You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2024/07/16 21:05:14 UTC
(activemq-artemis) 02/02: ARTEMIS-4925 Fixing issue with mirroring and replication
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
commit 6c1134388b4d0a3775b6b92b00e5e29baf1b683a
Author: Clebert Suconic <cl...@apache.org>
AuthorDate: Tue Jul 16 10:58:31 2024 -0400
ARTEMIS-4925 Fixing issue with mirroring and replication
the page writes on replica should include the address where they belong to avoid leaks between the mirrorSNF and the actual addresses.
---
.../activemq/artemis/core/paging/impl/Page.java | 2 +-
.../artemis/core/paging/impl/PagingStoreImpl.java | 2 +-
.../artemis/core/persistence/StorageManager.java | 6 ++--
.../impl/journal/JournalStorageManager.java | 11 ++++---
.../impl/nullpm/NullStorageManager.java | 2 +-
.../artemis/core/protocol/ServerPacketDecoder.java | 2 +-
.../wireformat/ReplicationPageWriteMessage.java | 37 ++++++++++++++++++----
.../core/replication/ReplicationEndpoint.java | 4 +--
.../core/replication/ReplicationManager.java | 4 +--
.../core/transaction/impl/TransactionImplTest.java | 2 +-
.../tests/integration/client/SendAckFailTest.java | 4 +--
.../integration/replication/ReplicationTest.java | 8 ++---
12 files changed, 54 insertions(+), 30 deletions(-)
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/Page.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/Page.java
index ffd6e4e064..c940c83907 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/Page.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/Page.java
@@ -180,7 +180,7 @@ public final class Page {
public synchronized void write(final PagedMessage message) throws Exception {
writeDirect(message);
- storageManager.pageWrite(message, pageId);
+ storageManager.pageWrite(storeName, message, pageId);
}
/** This write will not interact back with the storage manager.
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java
index 55bc1e3fcd..ab3115f97e 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java
@@ -1535,7 +1535,7 @@ public class PagingStoreImpl implements PagingStore {
final long newPageId = currentPageId + 1;
if (logger.isTraceEnabled()) {
- logger.trace("new pageNr={}", newPageId);
+ logger.trace("destination {} new pageNr={}", storeName, newPageId);
}
final Page oldPage = currentPage;
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java
index 81ca168588..4efa2101a4 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java
@@ -126,11 +126,11 @@ public interface StorageManager extends IDGenerator, ActiveMQComponent {
// Message related operations
- void pageClosed(SimpleString storeName, long pageNumber);
+ void pageClosed(SimpleString address, long pageNumber);
- void pageDeleted(SimpleString storeName, long pageNumber);
+ void pageDeleted(SimpleString address, long pageNumber);
- void pageWrite(PagedMessage message, long pageNumber);
+ void pageWrite(SimpleString address, PagedMessage message, long pageNumber);
void afterCompleteOperations(IOCallback run);
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManager.java
index 86bcdfe485..827cc5ff89 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManager.java
@@ -394,7 +394,7 @@ public class JournalStorageManager extends AbstractJournalStorageManager {
}
@Override
- public void pageWrite(final PagedMessage message, final long pageNumber) {
+ public void pageWrite(final SimpleString address, final PagedMessage message, final long pageNumber) {
if (messageJournal.isHistory()) {
try (ArtemisCloseable lock = closeableReadLock()) {
@@ -421,7 +421,7 @@ public class JournalStorageManager extends AbstractJournalStorageManager {
try (ArtemisCloseable lock = closeableReadLock()) {
if (isReplicated())
- replicator.pageWrite(message, pageNumber);
+ replicator.pageWrite(address, message, pageNumber);
}
}
}
@@ -703,8 +703,11 @@ public class JournalStorageManager extends AbstractJournalStorageManager {
Map<SimpleString, Collection<Integer>> info = new HashMap<>();
for (SimpleString storeName : pagingManager.getStoreNames()) {
PagingStore store = pagingManager.getPageStore(storeName);
- info.put(storeName, store.getCurrentIds());
- store.forceAnotherPage();
+ Collection<Integer> ids = store.getCurrentIds();
+ info.put(storeName, ids);
+ if (!ids.isEmpty()) {
+ store.forceAnotherPage();
+ }
}
return info;
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java
index 649bb742bb..6d19e1fa43 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java
@@ -402,7 +402,7 @@ public class NullStorageManager implements StorageManager {
}
@Override
- public void pageWrite(final PagedMessage message, final long pageNumber) {
+ public void pageWrite(final SimpleString address, final PagedMessage message, final long pageNumber) {
}
@Override
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/ServerPacketDecoder.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/ServerPacketDecoder.java
index a01b1301c1..fc28870e92 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/ServerPacketDecoder.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/ServerPacketDecoder.java
@@ -201,7 +201,7 @@ public class ServerPacketDecoder extends ClientPacketDecoder {
break;
}
case REPLICATION_PAGE_WRITE: {
- packet = new ReplicationPageWriteMessage(connection.isVersionUsingLongOnPageReplication());
+ packet = new ReplicationPageWriteMessage(connection.isVersionUsingLongOnPageReplication(), coreMessageObjectPools);
break;
}
case REPLICATION_PAGE_EVENT: {
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ReplicationPageWriteMessage.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ReplicationPageWriteMessage.java
index 0971fdd2a2..6c8b485cef 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ReplicationPageWriteMessage.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/ReplicationPageWriteMessage.java
@@ -18,8 +18,10 @@ package org.apache.activemq.artemis.core.protocol.core.impl.wireformat;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.paging.PagedMessage;
import org.apache.activemq.artemis.core.paging.impl.PagedMessageImpl;
+import org.apache.activemq.artemis.core.persistence.CoreMessageObjectPools;
import org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl;
import org.apache.activemq.artemis.utils.DataConstants;
@@ -29,28 +31,35 @@ public class ReplicationPageWriteMessage extends PacketImpl implements MessagePa
protected PagedMessage pagedMessage;
+ protected SimpleString address;
+
final boolean useLong;
- public ReplicationPageWriteMessage(final boolean useLong) {
+ final CoreMessageObjectPools coreMessageObjectPools;
+
+ public ReplicationPageWriteMessage(final boolean useLong, CoreMessageObjectPools coreMessageObjectPools) {
super(PacketImpl.REPLICATION_PAGE_WRITE);
this.useLong = useLong;
+ this.coreMessageObjectPools = coreMessageObjectPools;
}
- public ReplicationPageWriteMessage(final PagedMessage pagedMessage, final long pageNumber, final boolean useLong) {
- this(useLong);
+ public ReplicationPageWriteMessage(final PagedMessage pagedMessage, final long pageNumber, final boolean useLong, final SimpleString address) {
+ this(useLong, null);
this.pageNumber = pageNumber;
this.pagedMessage = pagedMessage;
+ this.address = address;
}
@Override
public int expectedEncodeSize() {
if (useLong) {
- return PACKET_HEADERS_SIZE + DataConstants.SIZE_LONG + // buffer.writeLong(pageNumber);
- pagedMessage.getEncodeSize(); // pagedMessage.encode(buffer);
+ return PACKET_HEADERS_SIZE + DataConstants.SIZE_LONG + // buffer.writeLong(pageNumber)
+ pagedMessage.getEncodeSize() + // pagedMessage.encode(buffer)
+ SimpleString.sizeofString(address); // SizeUtil.writeNullableString(address)
} else {
- return PACKET_HEADERS_SIZE + DataConstants.SIZE_INT + // buffer.writeInt(pageNumber);
- pagedMessage.getEncodeSize(); // pagedMessage.encode(buffer);
+ return PACKET_HEADERS_SIZE + DataConstants.SIZE_INT + // buffer.writeInt(pageNumber)
+ pagedMessage.getEncodeSize(); // pagedMessage.encode(buffer)
}
}
@@ -65,6 +74,9 @@ public class ReplicationPageWriteMessage extends PacketImpl implements MessagePa
buffer.writeInt((int) pageNumber);
}
pagedMessage.encode(buffer);
+ if (useLong) {
+ buffer.writeNullableSimpleString(address);
+ }
}
@Override
@@ -76,6 +88,17 @@ public class ReplicationPageWriteMessage extends PacketImpl implements MessagePa
}
pagedMessage = new PagedMessageImpl(0, null);
pagedMessage.decode(buffer);
+ if (buffer.readableBytes() > 0) {
+ address = SimpleString.readNullableSimpleString(buffer.byteBuf(), coreMessageObjectPools.getAddressDecoderPool());
+ }
+ }
+
+ public SimpleString getAddress() {
+ if (address != null) {
+ return address;
+ } else {
+ return pagedMessage.getMessage().getAddressSimpleString();
+ }
}
/**
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationEndpoint.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationEndpoint.java
index 09cf40567d..a548bca431 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationEndpoint.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationEndpoint.java
@@ -35,7 +35,6 @@ import java.util.concurrent.Executor;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.Interceptor;
-import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.io.SequentialFile;
import org.apache.activemq.artemis.core.journal.EncoderPersister;
@@ -831,8 +830,7 @@ public final class ReplicationEndpoint implements ChannelHandler, ActiveMQCompon
private void handlePageWrite(final ReplicationPageWriteMessage packet) throws Exception {
PagedMessage pgdMessage = packet.getPagedMessage();
pgdMessage.initMessage(storageManager);
- Message msg = pgdMessage.getMessage();
- Page page = getPage(msg.getAddressSimpleString(), packet.getPageNumber());
+ Page page = getPage(packet.getAddress(), packet.getPageNumber());
page.writeDirect(pgdMessage);
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationManager.java
index 3906af597e..55577d8b45 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationManager.java
@@ -299,9 +299,9 @@ public final class ReplicationManager implements ActiveMQComponent {
}
}
- public void pageWrite(final PagedMessage message, final long pageNumber) {
+ public void pageWrite(final SimpleString address, final PagedMessage message, final long pageNumber) {
if (started) {
- sendReplicatePacket(new ReplicationPageWriteMessage(message, pageNumber, remotingConnection.isVersionUsingLongOnPageReplication()));
+ sendReplicatePacket(new ReplicationPageWriteMessage(message, pageNumber, remotingConnection.isVersionUsingLongOnPageReplication(), address));
}
}
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/transaction/impl/TransactionImplTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/transaction/impl/TransactionImplTest.java
index ef77075744..deb5585e05 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/transaction/impl/TransactionImplTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/transaction/impl/TransactionImplTest.java
@@ -420,7 +420,7 @@ public class TransactionImplTest extends ServerTestBase {
}
@Override
- public void pageWrite(PagedMessage message, long pageNumber) {
+ public void pageWrite(SimpleString address, PagedMessage message, long pageNumber) {
}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/SendAckFailTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/SendAckFailTest.java
index daf6d5a235..03a126deb5 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/SendAckFailTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/SendAckFailTest.java
@@ -378,8 +378,8 @@ public class SendAckFailTest extends SpawnedTestBase {
}
@Override
- public void pageWrite(PagedMessage message, long pageNumber) {
- manager.pageWrite(message, pageNumber);
+ public void pageWrite(SimpleString address, PagedMessage message, long pageNumber) {
+ manager.pageWrite(address, message, pageNumber);
}
@Override
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/ReplicationTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/ReplicationTest.java
index 50f2728f6b..e0d2cb2762 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/ReplicationTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/ReplicationTest.java
@@ -275,10 +275,10 @@ public final class ReplicationTest extends ActiveMQTestBase {
replicatedJournal.appendAddRecordTransactional(23, 24, (byte) 1, new FakeData());
PagedMessage pgmsg = new PagedMessageImpl(msg, new long[0]);
- manager.pageWrite(pgmsg, 1);
- manager.pageWrite(pgmsg, 2);
- manager.pageWrite(pgmsg, 3);
- manager.pageWrite(pgmsg, 4);
+ manager.pageWrite(dummy, pgmsg, 1);
+ manager.pageWrite(dummy, pgmsg, 2);
+ manager.pageWrite(dummy, pgmsg, 3);
+ manager.pageWrite(dummy, pgmsg, 4);
blockOnReplication(storage, manager);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@activemq.apache.org
For additional commands, e-mail: commits-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact