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:12 UTC
(activemq-artemis) branch main updated (2bbf056edf -> 6c1134388b)
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
from 2bbf056edf ARTEMIS-4892: update to micrometer-core 1.13.2
new 5af90578cb ARTEMIS-4925 Test with mirror and replication.
new 6c1134388b ARTEMIS-4925 Fixing issue with mirroring and replication
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../api/core/management/SimpleManagement.java | 18 +
.../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 +-
.../mirror/ReplicatedBothNodesMirrorTest.java | 413 +++++++++++++++++++++
14 files changed, 485 insertions(+), 30 deletions(-)
create mode 100644 tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/brokerConnection/mirror/ReplicatedBothNodesMirrorTest.java
---------------------------------------------------------------------
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
(activemq-artemis) 02/02: ARTEMIS-4925 Fixing issue with mirroring and replication
Posted by cl...@apache.org.
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
(activemq-artemis) 01/02: ARTEMIS-4925 Test with mirror and replication.
Posted by cl...@apache.org.
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 5af90578cb155e9456745d63d632b0b65eb27ac0
Author: Clebert Suconic <cl...@apache.org>
AuthorDate: Mon Jul 15 23:39:23 2024 -0400
ARTEMIS-4925 Test with mirror and replication.
Files will leak on the target and messages will not be received after failover.
Also messages will be written to wrong destinations on the replica. The leak is actually between destinations, and the consequence is the file leak.
I usually keep test and fix on the same commit, but in this case I have been heavily validating the server with and without the fix,
so I will open an exception in this case and keep the fix and test separated.
---
.../api/core/management/SimpleManagement.java | 18 +
.../mirror/ReplicatedBothNodesMirrorTest.java | 413 +++++++++++++++++++++
2 files changed, 431 insertions(+)
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/SimpleManagement.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/SimpleManagement.java
index ece36b40b7..5152b36e4e 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/SimpleManagement.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/SimpleManagement.java
@@ -20,6 +20,7 @@ package org.apache.activemq.artemis.api.core.management;
import java.lang.invoke.MethodHandles;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
@@ -85,6 +86,10 @@ public class SimpleManagement implements AutoCloseable {
return simpleManagementLong("broker", "getCurrentTimeMillis");
}
+ public boolean isReplicaSync() throws Exception {
+ return simpleManagementBoolean("broker", "isReplicaSync");
+ }
+
public void rebuildPageCounters() throws Exception {
simpleManagementVoid("broker", "rebuildPageCounters");
}
@@ -103,6 +108,13 @@ public class SimpleManagement implements AutoCloseable {
return responseLong.get();
}
+ /** Simple helper for management returning a long.*/
+ public boolean simpleManagementBoolean(String resource, String method, Object... parameters) throws Exception {
+ AtomicBoolean responseBoolean = new AtomicBoolean();
+ doManagement((m) -> setupCall(m, resource, method, parameters), m -> setBooleanResult(m, responseBoolean), SimpleManagement::failed);
+ return responseBoolean.get();
+ }
+
/** Simple helper for management void calls.*/
public void simpleManagementVoid(String resource, String method, Object... parameters) throws Exception {
doManagement((m) -> setupCall(m, resource, method, parameters), null, SimpleManagement::failed);
@@ -191,6 +203,12 @@ public class SimpleManagement implements AutoCloseable {
result.set(resultLong);
}
+ protected static void setBooleanResult(ClientMessage m, AtomicBoolean result) throws Exception {
+ boolean resultBoolean = (boolean)ManagementHelper.getResult(m, Boolean.class);
+ logger.debug("management result:: {}", resultBoolean);
+ result.set(resultBoolean);
+ }
+
protected static void setIntResult(ClientMessage m, AtomicInteger result) throws Exception {
int resultInt = (int)ManagementHelper.getResult(m, Integer.class);
logger.debug("management result:: {}", resultInt);
diff --git a/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/brokerConnection/mirror/ReplicatedBothNodesMirrorTest.java b/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/brokerConnection/mirror/ReplicatedBothNodesMirrorTest.java
new file mode 100644
index 0000000000..8a40036db2
--- /dev/null
+++ b/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/brokerConnection/mirror/ReplicatedBothNodesMirrorTest.java
@@ -0,0 +1,413 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.activemq.artemis.tests.soak.brokerConnection.mirror;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.JMSException;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.StringWriter;
+import java.lang.invoke.MethodHandles;
+import java.util.HashSet;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
+
+import org.apache.activemq.artemis.api.core.management.SimpleManagement;
+import org.apache.activemq.artemis.core.config.amqpBrokerConnectivity.AMQPBrokerConnectionAddressType;
+import org.apache.activemq.artemis.tests.soak.SoakTestBase;
+import org.apache.activemq.artemis.tests.util.CFUtil;
+import org.apache.activemq.artemis.util.ServerUtil;
+import org.apache.activemq.artemis.utils.FileUtil;
+import org.apache.activemq.artemis.utils.TestParameters;
+import org.apache.activemq.artemis.utils.Wait;
+import org.apache.activemq.artemis.utils.cli.helper.HelperCreate;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Assumptions;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class ReplicatedBothNodesMirrorTest extends SoakTestBase {
+
+ private static final String TEST_NAME = "REPLICATED_BOTH_NODES_MIRROR_SOAK";
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ // Set this to true and log4j will be configured with some relevant log.trace for the AckManager at the server's
+ private static final boolean TRACE_LOGS = Boolean.parseBoolean(TestParameters.testProperty(TEST_NAME, "TRACE_LOGS", "false"));
+ private static final int NUMBER_MESSAGES = TestParameters.testProperty(TEST_NAME, "NUMBER_MESSAGES", 200);
+
+ private static final int SEND_COMMIT = TestParameters.testProperty(TEST_NAME, "SEND_COMMIT", 50);
+
+ /*
+ * Time each consumer takes to process a message received to allow some messages accumulating.
+ * This sleep happens right before the commit.
+ */
+ private static final String QUEUE_NAME = "queueTest";
+
+ private static String body;
+
+ static {
+ StringWriter writer = new StringWriter();
+ while (writer.getBuffer().length() < 30 * 1024) {
+ writer.append("The sky is blue, ..... watch out for poop from the birds though!...");
+ }
+ body = writer.toString();
+ }
+
+ public static final String DC1_NODE = "ReplicatedBothNodesMirrorTest/DC1";
+ public static final String DC2_NODE = "ReplicatedBothNodesMirrorTest/DC2";
+ public static final String DC2_REPLICA_NODE = "ReplicatedBothNodesMirrorTest/DC2_REPLICA";
+ public static final String DC1_REPLICA_NODE = "ReplicatedBothNodesMirrorTest/DC1_REPLICA";
+
+ volatile Process processDC1;
+ volatile Process processDC2;
+ volatile Process processDC1_REPLICA;
+ volatile Process processDC2_REPLICA;
+
+ @AfterEach
+ public void destroyServers() throws Exception {
+ if (processDC2_REPLICA != null) {
+ processDC2_REPLICA.destroyForcibly();
+ processDC2_REPLICA.waitFor(1, TimeUnit.MINUTES);
+ processDC2_REPLICA = null;
+ }
+ if (processDC1_REPLICA != null) {
+ processDC1_REPLICA.destroyForcibly();
+ processDC1_REPLICA.waitFor(1, TimeUnit.MINUTES);
+ processDC1_REPLICA = null;
+ }
+ if (processDC1 != null) {
+ processDC1.destroyForcibly();
+ processDC1.waitFor(1, TimeUnit.MINUTES);
+ processDC1 = null;
+ }
+ if (processDC2 != null) {
+ processDC2.destroyForcibly();
+ processDC2.waitFor(1, TimeUnit.MINUTES);
+ processDC2 = null;
+ }
+ }
+
+ private static final String DC1_IP = "localhost:61616";
+ private static final String DC1_BACKUP_IP = "localhost:61617";
+ private static final String DC2_IP = "localhost:61618";
+ private static final String DC2_BACKUP_IP = "localhost:61619";
+
+ private static String uri(String ip) {
+ return "tcp://" + ip;
+ }
+ private static String uriWithAlternate(String ip, String alternate) {
+ return "tcp://" + ip + "#tcp://" + alternate;
+ }
+
+ private void startDC2(SimpleManagement managementDC2) throws Exception {
+ processDC2 = startServer(DC2_NODE, -1, -1, new File(getServerLocation(DC2_NODE), "broker.properties"));
+ ServerUtil.waitForServerToStart(2, 10_000);
+ processDC2_REPLICA = startServer(DC2_REPLICA_NODE, -1, -1, new File(getServerLocation(DC2_REPLICA_NODE), "broker.properties"));
+ Wait.assertTrue(managementDC2::isReplicaSync);
+ }
+
+ private void startDC1(SimpleManagement managementDC1) throws Exception {
+ processDC1 = startServer(DC1_NODE, -1, -1, new File(getServerLocation(DC1_NODE), "broker.properties"));
+ ServerUtil.waitForServerToStart(0, 10_000);
+ processDC1_REPLICA = startServer(DC1_REPLICA_NODE, -1, -1, new File(getServerLocation(DC1_REPLICA_NODE), "broker.properties"));
+ Wait.assertTrue(managementDC1::isReplicaSync);
+ }
+
+ private static void createMirroredServer(String serverName,
+ String connectionName,
+ String mirrorURI,
+ int porOffset,
+ boolean replicated,
+ String clusterStatic) throws Exception {
+ File serverLocation = getFileServerLocation(serverName);
+ deleteDirectory(serverLocation);
+
+ HelperCreate cliCreateServer = new HelperCreate();
+ cliCreateServer.setAllowAnonymous(true).setArtemisInstance(serverLocation);
+ cliCreateServer.setNoWeb(false);
+ cliCreateServer.setArgs("--no-stomp-acceptor", "--no-hornetq-acceptor", "--no-mqtt-acceptor", "--no-amqp-acceptor", "--max-hops", "1", "--name", DC1_NODE);
+ cliCreateServer.addArgs("--queues", QUEUE_NAME);
+ cliCreateServer.setPortOffset(porOffset);
+ if (replicated) {
+ cliCreateServer.setReplicated(true);
+ cliCreateServer.setStaticCluster(clusterStatic);
+ cliCreateServer.setClustered(true);
+ } else {
+ cliCreateServer.setClustered(false);
+ }
+
+ cliCreateServer.createServer();
+
+ Properties brokerProperties = new Properties();
+ brokerProperties.put("messageExpiryScanPeriod", "1000");
+ brokerProperties.put("AMQPConnections." + connectionName + ".uri", mirrorURI);
+ brokerProperties.put("AMQPConnections." + connectionName + ".retryInterval", "1000");
+ brokerProperties.put("AMQPConnections." + connectionName + ".type", AMQPBrokerConnectionAddressType.MIRROR.toString());
+ brokerProperties.put("AMQPConnections." + connectionName + ".connectionElements.mirror.sync", "false");
+ brokerProperties.put("largeMessageSync", "false");
+
+ brokerProperties.put("addressSettings.#.maxSizeMessages", "50");
+ brokerProperties.put("addressSettings.#.maxReadPageMessages", "2000");
+ brokerProperties.put("addressSettings.#.maxReadPageBytes", "-1");
+ brokerProperties.put("addressSettings.#.prefetchPageMessages", "500");
+ // if we don't use pageTransactions we may eventually get a few duplicates
+ brokerProperties.put("mirrorPageTransaction", "true");
+ File brokerPropertiesFile = new File(serverLocation, "broker.properties");
+ saveProperties(brokerProperties, brokerPropertiesFile);
+
+ File brokerXml = new File(serverLocation, "/etc/broker.xml");
+ assertTrue(brokerXml.exists());
+ // Adding redistribution delay to broker configuration
+ assertTrue(FileUtil.findReplace(brokerXml, "<address-setting match=\"#\">", "<address-setting match=\"#\">\n\n" + " <redistribution-delay>0</redistribution-delay>\n"));
+ assertTrue(FileUtil.findReplace(brokerXml, "<page-size-bytes>10M</page-size-bytes>", "<page-size-bytes>100K</page-size-bytes>"));
+
+ if (TRACE_LOGS) {
+ replaceLogs(serverLocation);
+ }
+
+ }
+
+ private static void replaceLogs(File serverLocation) throws Exception {
+ File log4j = new File(serverLocation, "/etc/log4j2.properties");
+ assertTrue(FileUtil.findReplace(log4j, "logger.artemis_utils.level=INFO", "logger.artemis_utils.level=INFO\n" + "\n" + "logger.endpoint.name=org.apache.activemq.artemis.core.replication.ReplicationEndpoint\n" + "logger.endpoint.level=DEBUG\n" + "appender.console.filter.threshold.type = ThresholdFilter\n" + "appender.console.filter.threshold.level = info"));
+ }
+
+ private static void createMirroredBackupServer(String serverName, int porOffset, String clusterStatic, String mirrorURI) throws Exception {
+ File serverLocation = getFileServerLocation(serverName);
+ deleteDirectory(serverLocation);
+
+ HelperCreate cliCreateServer = new HelperCreate();
+ cliCreateServer.setAllowAnonymous(true).setArtemisInstance(serverLocation);
+ cliCreateServer.setMessageLoadBalancing("ON_DEMAND");
+ cliCreateServer.setNoWeb(false);
+ cliCreateServer.setArgs("--no-stomp-acceptor", "--no-hornetq-acceptor", "--no-mqtt-acceptor", "--no-amqp-acceptor", "--max-hops", "1", "--name", DC1_NODE);
+ cliCreateServer.setPortOffset(porOffset);
+ cliCreateServer.setClustered(true);
+ cliCreateServer.setReplicated(true);
+ cliCreateServer.setBackup(true);
+ cliCreateServer.setStaticCluster(clusterStatic);
+ cliCreateServer.createServer();
+
+ Properties brokerProperties = new Properties();
+ brokerProperties.put("messageExpiryScanPeriod", "1000");
+ brokerProperties.put("AMQPConnections.mirror.uri", mirrorURI);
+ brokerProperties.put("AMQPConnections.mirror.retryInterval", "1000");
+ brokerProperties.put("AMQPConnections.mirror.type", AMQPBrokerConnectionAddressType.MIRROR.toString());
+ brokerProperties.put("AMQPConnections.mirror.connectionElements.mirror.sync", "false");
+ brokerProperties.put("largeMessageSync", "false");
+
+ brokerProperties.put("addressSettings.#.maxSizeMessages", "1");
+ brokerProperties.put("addressSettings.#.maxReadPageMessages", "2000");
+ brokerProperties.put("addressSettings.#.maxReadPageBytes", "-1");
+ brokerProperties.put("addressSettings.#.prefetchPageMessages", "500");
+ // if we don't use pageTransactions we may eventually get a few duplicates
+ brokerProperties.put("mirrorPageTransaction", "true");
+ File brokerPropertiesFile = new File(serverLocation, "broker.properties");
+ saveProperties(brokerProperties, brokerPropertiesFile);
+
+
+ File brokerXml = new File(serverLocation, "/etc/broker.xml");
+ assertTrue(brokerXml.exists());
+ // Adding redistribution delay to broker configuration
+ assertTrue(FileUtil.findReplace(brokerXml, "<address-setting match=\"#\">", "<address-setting match=\"#\">\n\n" + " <redistribution-delay>0</redistribution-delay> <!-- added by SimpleMirrorSoakTest.java --> \n"));
+ assertTrue(FileUtil.findReplace(brokerXml, "<page-size-bytes>10M</page-size-bytes>", "<page-size-bytes>100K</page-size-bytes>"));
+
+ if (TRACE_LOGS) {
+ replaceLogs(serverLocation);
+ }
+ }
+
+ public static void createRealServers() throws Exception {
+ createMirroredServer(DC1_NODE, "mirror", uriWithAlternate(DC2_IP, DC2_BACKUP_IP), 0, true, uri(DC1_BACKUP_IP));
+ createMirroredBackupServer(DC1_REPLICA_NODE, 1, uri(DC1_IP), uriWithAlternate(DC2_IP, DC2_BACKUP_IP));
+ createMirroredServer(DC2_NODE, "mirror", uriWithAlternate(DC1_IP, DC1_BACKUP_IP), 2, true, uri(DC2_BACKUP_IP));
+ createMirroredBackupServer(DC2_REPLICA_NODE, 3, uri(DC2_IP), uriWithAlternate(DC1_IP, DC1_BACKUP_IP));
+ }
+
+ @Test
+ public void testFailoverLaterStart() throws Exception {
+ testMirror(true);
+ }
+
+ @Test
+ public void testFailoverWhileMirroring() throws Exception {
+ testMirror(false);
+ }
+
+ private void testMirror(boolean laterStart) throws Exception {
+ createRealServers();
+
+ SimpleManagement managementDC1 = new SimpleManagement(uri(DC1_IP), null, null);
+ SimpleManagement managementDC2 = new SimpleManagement(uri(DC2_IP), null, null);
+
+ startDC1(managementDC1);
+
+ if (!laterStart) {
+ startDC2(managementDC2);
+ }
+
+ runAfter(() -> managementDC1.close());
+ runAfter(() -> managementDC2.close());
+
+ sendMessages(QUEUE_NAME);
+
+ processDC1.destroyForcibly();
+ processDC1.waitFor(10, TimeUnit.SECONDS);
+
+ if (laterStart) {
+ startDC2(managementDC2);
+ }
+
+ // Mirror failover could challenge the order
+ HashSet<Integer> receivedIDs = new HashSet<>();
+
+ ConnectionFactory connectionFactoryDC2 = CFUtil.createConnectionFactory("amqp", uri(DC2_IP));
+ try (Connection connection = connectionFactoryDC2.createConnection()) {
+ Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+ connection.start();
+ MessageConsumer consumer = session.createConsumer(session.createQueue(QUEUE_NAME));
+ for (int i = 0; i < NUMBER_MESSAGES; i++) {
+ TextMessage message = (TextMessage) consumer.receive(30_000);
+ Assertions.assertNotNull(message);
+ receivedIDs.add(message.getIntProperty("i"));
+ if (i > 0 && i % SEND_COMMIT == 0) {
+ logger.info("Received {} messages", i);
+ session.commit();
+ }
+ }
+
+ session.commit();
+ }
+
+ Assertions.assertEquals(NUMBER_MESSAGES, receivedIDs.size());
+ for (int i = 0; i < NUMBER_MESSAGES; i++) {
+ Assertions.assertTrue(receivedIDs.contains(i));
+ }
+ }
+
+
+ @Test
+ public void testMultipleSenders() throws Exception {
+ try {
+ lsof();
+ } catch (IOException e) {
+ logger.warn("lsof is not available in this platform, we will ignore this test - {}", e.getMessage(), e);
+ Assumptions.abort("lsof is not available");
+ }
+ createRealServers();
+
+ SimpleManagement managementDC1 = new SimpleManagement(uri(DC1_IP), null, null);
+ SimpleManagement managementDC2 = new SimpleManagement(uri(DC2_IP), null, null);
+
+ startDC1(managementDC1);
+ startDC2(managementDC2);
+
+ runAfter(() -> managementDC1.close());
+ runAfter(() -> managementDC2.close());
+
+ int destinations = 5;
+ ExecutorService executorService = Executors.newFixedThreadPool(destinations);
+ runAfter(executorService::shutdownNow);
+
+ CountDownLatch latch = new CountDownLatch(destinations);
+ AtomicInteger errors = new AtomicInteger(0);
+
+ for (int i = 0; i < destinations; i++) {
+ String destination = "queue" + i;
+ executorService.execute(() -> {
+ try {
+ sendMessages(destination);
+ } catch (Throwable e) {
+ logger.warn(e.getMessage(), e);
+ errors.incrementAndGet();
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+
+ Assertions.assertTrue(latch.await(5, TimeUnit.MINUTES));
+
+ int openFiles = lsof();
+
+ logger.info("There are {} open files", openFiles);
+
+ // lsof is showing a file descriptor associated with multiple threads. So it is expected to have quite a few repetitions
+ // when the issue is happening we would have around 40k, 50k entries or a lot more if you add more messages.
+ Assertions.assertTrue(openFiles < 2500, () -> "There was " + openFiles + " open files");
+
+ }
+
+ private int lsof() throws IOException, InterruptedException {
+ ProcessBuilder lsofBuilder = new ProcessBuilder();
+ lsofBuilder.command("lsof", "-n", "-P");
+
+ Process process = lsofBuilder.start();
+ runAfter(process::destroyForcibly);
+
+ InputStream inputStream = process.getInputStream();
+ BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
+ AtomicInteger filesCounter = new AtomicInteger();
+ try (Stream<String> lines = reader.lines()) {
+ lines.filter(line -> line.contains("paging") && line.contains(basedir)).forEach(l -> {
+ logger.info("file {}", l);
+ filesCounter.incrementAndGet();
+ });
+ }
+ Assertions.assertTrue(process.waitFor(10, TimeUnit.SECONDS));
+ return filesCounter.get();
+ }
+
+ private static void sendMessages(String queueName) throws JMSException {
+ ConnectionFactory connectionFactoryDC1A = CFUtil.createConnectionFactory("amqp", uri(DC1_IP));
+ try (Connection connection = connectionFactoryDC1A.createConnection()) {
+ Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+ MessageProducer producer = session.createProducer(session.createQueue(queueName));
+ for (int i = 0; i < NUMBER_MESSAGES; i++) {
+ TextMessage message = session.createTextMessage(body);
+ message.setIntProperty("i", i);
+ message.setBooleanProperty("large", false);
+ producer.send(message);
+ if (i > 0 && i % SEND_COMMIT == 0) {
+ logger.info("Sent {} messages on {}", i, queueName);
+ session.commit();
+ }
+ }
+
+ session.commit();
+ }
+ }
+}
\ No newline at end of file
---------------------------------------------------------------------
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