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 2020/11/23 04:43:07 UTC
[activemq-artemis] branch master updated: ARTEMIS-3007 Checking if
a message becomes large during copy
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
The following commit(s) were added to refs/heads/master by this push:
new 686036b ARTEMIS-3007 Checking if a message becomes large during copy
new a5fd97d This closes #3356
686036b is described below
commit 686036b1a4e387ac5054f04a09a587a242a0c6f8
Author: Clebert Suconic <cl...@apache.org>
AuthorDate: Sun Nov 22 23:28:38 2020 -0500
ARTEMIS-3007 Checking if a message becomes large during copy
This is to avoid shutting down the server on a critical failure in case the message is a few bytes shy
from beyond the max buffer size.
This will prevent the issue.
---
.../artemis/core/server/impl/QueueImpl.java | 4 +-
.../artemis/tests/util/ActiveMQTestBase.java | 7 +++
.../tests/integration/client/LargeMessageTest.java | 58 ++++++++++++++++++++++
3 files changed, 68 insertions(+), 1 deletion(-)
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
index ec68f9e..c3247f7 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
@@ -65,6 +65,7 @@ import org.apache.activemq.artemis.core.paging.cursor.PagedReference;
import org.apache.activemq.artemis.core.persistence.AddressQueueStatus;
import org.apache.activemq.artemis.core.persistence.OperationContext;
import org.apache.activemq.artemis.core.persistence.StorageManager;
+import org.apache.activemq.artemis.core.persistence.impl.journal.LargeServerMessageImpl;
import org.apache.activemq.artemis.core.postoffice.Binding;
import org.apache.activemq.artemis.core.postoffice.Bindings;
import org.apache.activemq.artemis.core.postoffice.DuplicateIDCache;
@@ -3460,7 +3461,8 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
copy.reencode();
- return copy;
+ // in some edge cases a large message can become large during the copy
+ return LargeServerMessageImpl.checkLargeMessage(copy, storageManager);
}
private void expire(final Transaction tx, final MessageReference ref) throws Exception {
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/tests/util/ActiveMQTestBase.java b/artemis-server/src/test/java/org/apache/activemq/artemis/tests/util/ActiveMQTestBase.java
index 3f5436a..2e195f1 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/tests/util/ActiveMQTestBase.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/tests/util/ActiveMQTestBase.java
@@ -67,6 +67,7 @@ import org.apache.activemq.artemis.api.core.ICoreMessage;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.Pair;
import org.apache.activemq.artemis.api.core.QueueConfiguration;
+import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.TransportConfiguration;
import org.apache.activemq.artemis.api.core.client.ActiveMQClient;
@@ -125,6 +126,7 @@ import org.apache.activemq.artemis.core.server.cluster.RemoteQueueBinding;
import org.apache.activemq.artemis.core.server.cluster.impl.MessageLoadBalancingType;
import org.apache.activemq.artemis.core.server.impl.Activation;
import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.core.server.impl.LiveOnlyActivation;
import org.apache.activemq.artemis.core.server.impl.SharedNothingBackupActivation;
import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
@@ -1598,6 +1600,11 @@ public abstract class ActiveMQTestBase extends Assert {
}
}
+ protected void createAnycastPair(ActiveMQServer server, String queueName) throws Exception {
+ server.addAddressInfo(new AddressInfo(queueName).addRoutingType(RoutingType.ANYCAST).setAutoCreated(false));
+ server.createQueue(new QueueConfiguration(queueName).setRoutingType(RoutingType.ANYCAST).setAddress(queueName));
+ }
+
protected void createQueue(final String address, final String queue) throws Exception {
ServerLocator locator = createInVMNonHALocator();
ClientSessionFactory sf = locator.createSessionFactory();
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/LargeMessageTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/LargeMessageTest.java
index b0da405..ad3a48b 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/LargeMessageTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/LargeMessageTest.java
@@ -686,6 +686,64 @@ public class LargeMessageTest extends LargeMessageTestBase {
}
+
+ @Test
+ public void testDLQAlmostLarge() throws Exception {
+ SimpleString addressName = SimpleString.toSimpleString("SomewhatHugeNameToBeUsedxxxxxxxxxxxxxxxxxxxiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiixxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx");
+ SimpleString dlqName = SimpleString.toSimpleString("DLQ" + addressName.toString());
+
+
+ ClientSession session = null;
+
+ ActiveMQServer server = createServer(true, isNetty(), storeType);
+ server.getConfiguration().setJournalFileSize(1024 * 1024);
+ server.getConfiguration().setJournalBufferSize_AIO(100 * 1024);
+ server.start();
+
+ server.getAddressSettingsRepository().clear();
+ AddressSettings settings = new AddressSettings().setDeadLetterAddress(dlqName).setMaxDeliveryAttempts(1);
+ server.getAddressSettingsRepository().addMatch("#", settings);
+
+ createAnycastPair(server, dlqName.toString());
+ createAnycastPair(server, addressName.toString());
+
+ locator.setMinLargeMessageSize(1024 * 1024);
+ ClientSessionFactory sf = addSessionFactory(createSessionFactory(locator));
+
+ session = sf.createSession(false, false, false);
+
+ ClientProducer producer = session.createProducer(addressName);
+
+ ClientMessage clientMessage = session.createMessage(true);
+ clientMessage.getBodyBuffer().writeBytes(new byte[100 * 1024 - 900]);
+ producer.send(clientMessage);
+
+ session.commit();
+
+ session.start();
+
+ ClientConsumer consumer = session.createConsumer(addressName);
+
+ for (int i = 0; i < 2; i++) {
+ if (i == 0) {
+ ClientMessage msg = consumer.receive(10000);
+ Assert.assertNotNull(msg);
+ msg.acknowledge();
+ session.rollback();
+ } else {
+ ClientMessage msg = consumer.receiveImmediate();
+ Assert.assertNull(msg);
+ }
+ }
+
+ consumer.close();
+
+ consumer = session.createConsumer(dlqName);
+ ClientMessage msg = consumer.receive(1000);
+ Assert.assertNotNull(msg);
+
+ }
+
@Test
public void testDLAOnExpiryNonDurableMessage() throws Exception {
final int messageSize = (int) (3.5 * ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE);