[
https://issues.apache.org/jira/browse/ARTEMIS-5037?focusedWorklogId=946832&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-946832
]
ASF GitHub Bot logged work on ARTEMIS-5037:
-------------------------------------------
Author: ASF GitHub Bot
Created on: 05/Dec/24 09:37
Start Date: 05/Dec/24 09:37
Worklog Time Spent: 10m
Work Description: lavocatt commented on code in PR #5220:
URL: https://github.com/apache/activemq-artemis/pull/5220#discussion_r1871001332
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPMirrorConnectionTest.java:
##########
@@ -229,6 +271,173 @@ public void testBrokerAddsAddressAndQueue() throws
Exception {
}
}
+ @Override
+ protected void configureAMQPAcceptorParameters(Map<String, Object> params) {
+ params.put("amqpMinLargeMessageSize", 100 * 1024);
+ }
+
+ @Test
+ @Timeout(20)
+ public void testNoForwardBlocksMessagesAndControlsPropagation() throws
Exception {
+ doTestNoForwardBlocksMessagesAndControlsPropagation(false, false);
+ }
+
+ @Test
+ @Timeout(20)
+ public void
testNoForwardBlocksMessagesAndControlsPropagationWithTunneling() throws
Exception {
+ doTestNoForwardBlocksMessagesAndControlsPropagation(true, false);
+ }
+
+ @Test
+ @Timeout(20)
+ public void testNoForwardBlocksLargeMessagesAndControlsPropagation() throws
Exception {
+ doTestNoForwardBlocksMessagesAndControlsPropagation(false, true);
+ }
+
+ @Test
+ @Timeout(20)
+ public void
testNoForwardBlocksLargeMessagesAndControlsPropagationWithTunneling() throws
Exception {
+ doTestNoForwardBlocksMessagesAndControlsPropagation(true, true);
+ }
+
+ private void doTestNoForwardBlocksMessagesAndControlsPropagation(boolean
tunneling, boolean longMessage) throws Exception {
+ final Map<String, Object> brokerProperties = new HashMap<>();
+ brokerProperties.put(AMQPMirrorControllerSource.BROKER_ID.toString(),
"Test-Broker");
+
+ final String[] capabilities;
+ ArrayList<String> capabilitiesList = new ArrayList<>();
+ int messageFormat = 0;
+
+ capabilitiesList.add("amq.mirror");
+ if (tunneling) {
+
capabilitiesList.add(AmqpSupport.CORE_MESSAGE_TUNNELING_SUPPORT.toString());
+ messageFormat = AMQP_TUNNELED_CORE_MESSAGE_FORMAT;
+ }
+ capabilities = capabilitiesList.toArray(new String[]{});
+
+ // Topology of the test: server -(noForward)-> server_2 -> peer_3
+ try (ProtonTestServer peer_3 = new ProtonTestServer()) {
+ peer_3.expectSASLPlainConnect("user", "pass", "PLAIN", "ANONYMOUS");
+ peer_3.expectOpen().respond();
+ peer_3.expectBegin().respond();
+ peer_3.expectAttach().ofSender()
+ .withName(Matchers.startsWith("$ACTIVEMQ_ARTEMIS_MIRROR"))
+ .withDesiredCapabilities(capabilities)
+ .respond()
+ .withOfferedCapabilities(capabilities)
+ .withPropertiesMap(brokerProperties);
+ peer_3.remoteFlow().withLinkCredit(10).queue();
+ peer_3.start();
+
+ final URI remoteURI = peer_3.getServerURI();
+ logger.info("Connect test started, peer listening on: {}", remoteURI);
+
+ final int AMQP_PORT_2 = BROKER_PORT_NUM + 1;
+ final ActiveMQServer server_2 = createServer(AMQP_PORT_2, false);
+ {
+ AMQPBrokerConnectConfiguration amqpConnection = new
AMQPBrokerConnectConfiguration(getTestMethodName() + "toPeer3", "tcp://" +
remoteURI.getHost() + ":" + remoteURI.getPort());
+ amqpConnection.setReconnectAttempts(0);// No reconnects
+ amqpConnection.setUser("user");
+ amqpConnection.setPassword("pass");
+ amqpConnection.addElement(new
AMQPMirrorBrokerConnectionElement().addProperty(TUNNEL_CORE_MESSAGES,
Boolean.toString(tunneling)).setQueueCreation(true).setAddressFilter(getQueueName()
+ ",sometest"));
+ server_2.getConfiguration().addAMQPConnection(amqpConnection);
+
+ amqpConnection = new
AMQPBrokerConnectConfiguration(getTestMethodName() + "toServer",
"tcp://localhost:" + BROKER_PORT_NUM);
+ amqpConnection.addElement(new
AMQPMirrorBrokerConnectionElement().setAddressFilter(getQueueName()));
+ server_2.getConfiguration().addAMQPConnection(amqpConnection);
+ }
+
+ {
+ AMQPBrokerConnectConfiguration amqpConnection = new
AMQPBrokerConnectConfiguration(getTestMethodName() + "toServer2",
"tcp://localhost:" + AMQP_PORT_2).setRetryInterval(50);
Review Comment:
We don't really need to set this retryInterval, I prefer to simply remove it.
Issue Time Tracking
-------------------
Worklog Id: (was: 946832)
Time Spent: 7h (was: 6h 50m)
> AMQ Broker Mirroring: One to Many - avoid the infinite loop of the messages
> ---------------------------------------------------------------------------
>
> Key: ARTEMIS-5037
> URL: https://issues.apache.org/jira/browse/ARTEMIS-5037
> Project: ActiveMQ Artemis
> Issue Type: Bug
> Reporter: Thomas Lavocat
> Assignee: Thomas Lavocat
> Priority: Major
> Labels: pull-request-available
> Time Spent: 7h
> Remaining Estimate: 0h
>
> AMQ Broker Mirroring: One to Many - avoid the infinite loop of the messages:
> if we have a->b-c->a..
> a message will circulate forever in the mirrors
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
For further information, visit: https://activemq.apache.org/contact
