Skip to content

ARTEMIS-6189 QoS 1 & 2 message handling is not resilient #6613

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jbertram wants to merge 4 commits into
base: main
Choose a base branch
from
+4,220 −957
Open

ARTEMIS-6189 QoS 1 & 2 message handling is not resilient #6613

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
e2d5177
ARTEMIS-6189 generify JournalHashMap collection ID
jbertram Aug 12, 2026
00d0b7d
ARTEMIS-6189 implement MQTT QoS 1 & 2 resiliency
jbertram Aug 12, 2026
f8f9d49
ARTEMIS-6189 add QoS 1 & 2 resiliency tests
jbertram Aug 12, 2026
10d430a
ARTEMIS-6189 fix flaky MQTT tests
jbertram Aug 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 15 additions & 9 deletions ...n/java/org/apache/activemq/artemis/core/journal/collections/AbstractHashMapPersister.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@
import org.apache.activemq.artemis.core.persistence.Persister;
import org.apache.activemq.artemis.utils.DataConstants;

public abstract class AbstractHashMapPersister<K, V> implements Persister<JournalHashMap.MapRecord<K, V>> {
public abstract class AbstractHashMapPersister<I, K, V> implements Persister<JournalHashMap.MapRecord<I, K, V>> {

private final byte VERSION = 0;

Expand All @@ -32,14 +32,20 @@ public byte getID() {
}

@Override
public final int getEncodeSize(JournalHashMap.MapRecord<K, V> record) {
public final int getEncodeSize(JournalHashMap.MapRecord<I, K, V> record) {
return DataConstants.SIZE_LONG + // recordID
DataConstants.SIZE_BYTE + // Version
DataConstants.SIZE_LONG + // collectionID
getCollectionIdSize(record.collectionID) +
getKeySize(record.key) +
getValueSize(record.value);
}

protected abstract int getCollectionIdSize(I collectionID);

protected abstract void encodeCollectionId(ActiveMQBuffer buffer, I collectionID);

protected abstract I decodeCollectionId(ActiveMQBuffer buffer);

protected abstract int getKeySize(K key);

protected abstract void encodeKey(ActiveMQBuffer buffer, K key);
Expand All @@ -53,28 +59,28 @@ public final int getEncodeSize(JournalHashMap.MapRecord<K, V> record) {
protected abstract V decodeValue(ActiveMQBuffer buffer, K key);

@Override
public final void encode(ActiveMQBuffer buffer, JournalHashMap.MapRecord<K, V> record) {
public final void encode(ActiveMQBuffer buffer, JournalHashMap.MapRecord<I, K, V> record) {
buffer.writeLong(record.id);
buffer.writeByte(VERSION);
buffer.writeLong(record.collectionID);
encodeCollectionId(buffer, record.collectionID);
encodeKey(buffer, record.key);
encodeValue(buffer, record.value);
}

@Override
public final JournalHashMap.MapRecord<K, V> decode(ActiveMQBuffer buffer,
JournalHashMap.MapRecord<K, V> record,
public final JournalHashMap.MapRecord<I, K, V> decode(ActiveMQBuffer buffer,
JournalHashMap.MapRecord<I, K, V> record,
CoreMessageObjectPools pool) {
long id = buffer.readLong();

byte version = buffer.readByte();
assert version == VERSION;

long collectionID = buffer.readLong();
I collectionID = decodeCollectionId(buffer);
K key = decodeKey(buffer);
V value = decodeValue(buffer, key);

JournalHashMap.MapRecord<K, V> mapRecord = new JournalHashMap.MapRecord<>(collectionID, id, key, value);
JournalHashMap.MapRecord<I, K, V> mapRecord = new JournalHashMap.MapRecord<>(collectionID, id, key, value);
return mapRecord;
}
}
56 changes: 31 additions & 25 deletions ...al/src/main/java/org/apache/activemq/artemis/core/journal/collections/JournalHashMap.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@
import java.util.Objects;
import java.util.Set;
import java.util.function.BiConsumer;
import java.util.function.LongFunction;
import java.util.function.Function;
import java.util.function.LongSupplier;
import java.util.function.Supplier;

Expand All @@ -38,21 +38,22 @@
import org.slf4j.LoggerFactory;

/**
* I = Collection ID type
* K = Key
* V = Value
* C = Context
*/
public class JournalHashMap<K, V, C> implements Map<K, V> {
public class JournalHashMap<I, K, V, C> implements Map<K, V> {

private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());

public static class MapRecord<K, V> implements Entry<K, V> {
final long collectionID;
public static class MapRecord<I, K, V> implements Entry<K, V> {
final I collectionID;
final long id;
final K key;
V value;

MapRecord(long collectionID, long id, K key, V value) {
MapRecord(I collectionID, long id, K key, V value) {
this.collectionID = collectionID;
this.id = id;
this.key = key;
Expand Down Expand Up @@ -86,11 +87,11 @@ public boolean equals(Object obj) {
if (this == obj) {
return true;
}
if (!(obj instanceof MapRecord<?, ?> other)) {
if (!(obj instanceof MapRecord<?, ?, ?> other)) {
return false;
}

return collectionID == other.collectionID &&
return Objects.equals(collectionID, other.collectionID) &&
id == other.id &&
Objects.equals(key, other.key) &&
Objects.equals(value, other.value);
Expand All @@ -102,7 +103,7 @@ public int hashCode() {
}
}

public JournalHashMap(long collectionId, MapStorageManager journal, LongSupplier idGenerator, Persister<MapRecord<K, V>> persister, byte recordType, Supplier<IOCompletion> completionSupplier, LongFunction<C> contextProvider, IOCriticalErrorListener ioExceptionListener) {
public JournalHashMap(I collectionId, MapStorageManager journal, LongSupplier idGenerator, Persister<MapRecord<I, K, V>> persister, byte recordType, Supplier<IOCompletion> completionSupplier, Function<I, C> contextProvider, IOCriticalErrorListener ioExceptionListener) {
this.collectionId = collectionId;
this.journal = journal;
this.idGenerator = idGenerator;
Expand All @@ -115,13 +116,13 @@ public JournalHashMap(long collectionId, MapStorageManager journal, LongSupplier

C context;

LongFunction<C> contextProvider;
Function<I, C> contextProvider;

private final Persister<MapRecord<K, V>> persister;
private final Persister<MapRecord<I, K, V>> persister;

private final MapStorageManager journal;

private final long collectionId;
private final I collectionId;

private final byte recordType;

Expand All @@ -131,9 +132,9 @@ public JournalHashMap(long collectionId, MapStorageManager journal, LongSupplier

private final IOCriticalErrorListener exceptionListener;

private final Map<K, MapRecord<K, V>> map = new HashMap<>();
private final Map<K, MapRecord<I, K, V>> map = new HashMap<>();

public long getCollectionId() {
public I getCollectionId() {
return collectionId;
}

Expand All @@ -149,7 +150,7 @@ public C getContext() {
return context;
}

public JournalHashMap<K, V, C> setContext(C context) {
public JournalHashMap<I, K, V, C> setContext(C context) {
this.context = context;
return this;
}
Expand All @@ -166,7 +167,7 @@ public synchronized boolean containsKey(Object key) {

@Override
public synchronized boolean containsValue(Object value) {
for (Entry<K, MapRecord<K, V>> entry : map.entrySet()) {
for (Entry<K, MapRecord<I, K, V>> entry : map.entrySet()) {
if (value.equals(entry.getValue().value)) {
return true;
}
Expand All @@ -176,7 +177,7 @@ public synchronized boolean containsValue(Object value) {

@Override
public synchronized V get(Object key) {
MapRecord<K, V> record = map.get(key);
MapRecord<I, K, V> record = map.get(key);
if (record == null) {
return null;
} else {
Expand All @@ -187,17 +188,17 @@ public synchronized V get(Object key) {
/**
* This is to be called from a single thread during reload, no need to be synchronized
*/
public void reload(MapRecord<K, V> reloadValue) {
public void reload(MapRecord<I, K, V> reloadValue) {
map.put(reloadValue.getKey(), reloadValue);
}

@Override
public synchronized V put(K key, V value) {
logger.debug("adding {} = {}", key, value);
long id = idGenerator.getAsLong();
MapRecord<K, V> record = new MapRecord<>(collectionId, id, key, value);
MapRecord<I, K, V> record = new MapRecord<>(collectionId, id, key, value);
store(record);
MapRecord<K, V> oldRecord = map.put(key, record);
MapRecord<I, K, V> oldRecord = map.put(key, record);

if (oldRecord != null) {
removed(oldRecord);
Expand All @@ -208,13 +209,12 @@ public synchronized V put(K key, V value) {

}

private synchronized void store(MapRecord<K, V> record) {
private synchronized void store(MapRecord<I, K, V> record) {
try {
IOCompletion callback = null;
if (completionSupplier != null) {
callback = completionSupplier.get();
}

if (callback == null) {
journal.storeMapRecord(record.id, recordType, persister, record, false);
} else {
Expand All @@ -227,7 +227,7 @@ private synchronized void store(MapRecord<K, V> record) {
}

// callers must be synchronized
private void removed(MapRecord<K, V> record) {
private void removed(MapRecord<I, K, V> record) {
if (logger.isTraceEnabled()) {
logger.trace("Removing record {}", record);
}
Expand All @@ -239,7 +239,7 @@ private void removed(MapRecord<K, V> record) {
}

// callers must be synchronized
private void removed(MapRecord<K, V> record, long txid) {
private void removed(MapRecord<I, K, V> record, long txid) {
try {
journal.deleteMapRecordTx(txid, record.id);
} catch (Exception e) {
Expand All @@ -249,7 +249,10 @@ private void removed(MapRecord<K, V> record, long txid) {

@Override
public synchronized V remove(Object key) {
MapRecord<K, V> record = map.remove(key);
MapRecord<I, K, V> record = map.remove(key);
if (record == null) {
return null;
}
this.removed(record);
return record.value;
}
Expand All @@ -261,7 +264,10 @@ public synchronized V remove(Object key) {
* not expected.
*/
public synchronized V remove(Object key, long transactionID) {
MapRecord<K, V> record = map.remove(key);
MapRecord<I, K, V> record = map.remove(key);
if (record == null) {
return null;
}
this.removed(record, transactionID);
return record.value;
}
Expand Down
35 changes: 19 additions & 16 deletions ...ain/java/org/apache/activemq/artemis/core/journal/collections/JournalHashMapProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,30 +18,31 @@
package org.apache.activemq.artemis.core.journal.collections;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.function.LongFunction;
import java.util.Map;
import java.util.function.Function;
import java.util.function.LongSupplier;
import java.util.function.Supplier;

import io.netty.util.collection.LongObjectHashMap;
import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
import org.apache.activemq.artemis.core.journal.IOCompletion;
import org.apache.activemq.artemis.core.journal.RecordInfo;
import org.apache.activemq.artemis.core.persistence.Persister;

public class JournalHashMapProvider<K, V, C> {
public class JournalHashMapProvider<I, K, V, C> {

final MapStorageManager journal;
final Persister<JournalHashMap.MapRecord<K, V>> persister;
final LongObjectHashMap<JournalHashMap<K, V, C>> journalMaps = new LongObjectHashMap<>();
final Persister<JournalHashMap.MapRecord<I, K, V>> persister;
final Map<I, JournalHashMap<I, K, V, C>> journalMaps = new HashMap<>();
final LongSupplier idSupplier;
final byte recordType;
final IOCriticalErrorListener ioExceptionListener;
final Supplier<IOCompletion> ioCompletionSupplier;
final LongFunction<C> contextProvider;
final Function<I, C> contextProvider;

public JournalHashMapProvider(LongSupplier idSupplier, MapStorageManager journal, AbstractHashMapPersister<K, V> persister, byte recordType, Supplier<IOCompletion> ioCompletionSupplier, LongFunction<C> contextProvider, IOCriticalErrorListener ioExceptionListener) {
public JournalHashMapProvider(LongSupplier idSupplier, MapStorageManager journal, AbstractHashMapPersister<I, K, V> persister, byte recordType, Supplier<IOCompletion> ioCompletionSupplier, Function<I, C> contextProvider, IOCriticalErrorListener ioExceptionListener) {
this.idSupplier = idSupplier;
this.persister = persister;
this.journal = journal;
Expand All @@ -51,35 +52,37 @@ public JournalHashMapProvider(LongSupplier idSupplier, MapStorageManager journal
this.ioCompletionSupplier = ioCompletionSupplier;
}

public List<JournalHashMap<K, V, C>> getMaps() {
List<JournalHashMap<K, V, C>> maps = new ArrayList<>();
journalMaps.values().forEach(maps::add);
return maps;
public List<JournalHashMap<I, K, V, C>> getMaps() {
return new ArrayList<>(journalMaps.values());
}

public void clear() {
journalMaps.clear();
}

public void reload(RecordInfo recordInfo) {
JournalHashMap.MapRecord<K, V> mapRecord = persister.decode(recordInfo.wrapData(), null, null);
JournalHashMap.MapRecord<I, K, V> mapRecord = persister.decode(recordInfo.wrapData(), null, null);
getMap(mapRecord.collectionID, null).reload(mapRecord);
}

public Iterator<JournalHashMap<K, V, C>> iterMaps() {
public Iterator<JournalHashMap<I, K, V, C>> iterMaps() {
return journalMaps.values().iterator();
}

public synchronized JournalHashMap<K, V, C> getMap(long collectionID, C context) {
JournalHashMap<K, V, C> journalHashMap = journalMaps.get(collectionID);
public synchronized JournalHashMap<I, K, V, C> getMap(I collectionID, C context) {
JournalHashMap<I, K, V, C> journalHashMap = journalMaps.get(collectionID);
if (journalHashMap == null) {
journalHashMap = new JournalHashMap<>(collectionID, journal, idSupplier, persister, recordType, ioCompletionSupplier, contextProvider, ioExceptionListener).setContext(context);
journalMaps.put(collectionID, journalHashMap);
}
return journalHashMap;
}

public JournalHashMap<K, V, C> getMap(long collectionID) {
public JournalHashMap<I, K, V, C> getMap(I collectionID) {
return getMap(collectionID, null);
}

public boolean containsMap(I collectionID) {
return journalMaps.containsKey(collectionID);
}
}
5 changes: 5 additions & 0 deletions artemis-pom/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,6 +111,11 @@
<artifactId>org.eclipse.paho.mqttv5.client</artifactId>
<version>${paho.client.mqtt.version}</version>
</dependency>
<dependency>
<groupId>com.hivemq</groupId>
<artifactId>hivemq-mqtt-client</artifactId>
<version>${hive.client.mqtt.version}</version>
</dependency>
<dependency>
<groupId>org.fusesource.mqtt-client</groupId>
<artifactId>mqtt-client</artifactId>
Expand Down
Loading
Loading