You are viewing a plain text version of this content. The canonical link for it is here.
Posted to gitbox@activemq.apache.org by GitBox <gi...@apache.org> on 2022/04/26 16:37:34 UTC
[GitHub] [activemq-artemis] iliya-gr opened a new pull request, #4054: ARTEMIS-3759 Add mirror controller address filter support
iliya-gr opened a new pull request, #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054
Allow replication only certain addresses with mirror controller.
The configuration is similar to cluster address configuration.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] iliya-gr commented on a diff in pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
iliya-gr commented on code in PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#discussion_r859692047
##########
artemis-server/src/main/resources/schema/artemis-configuration.xsd:
##########
@@ -2442,6 +2442,13 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
+ <xsd:attribute name="address" type="xsd:string" use="optional">
+ <xsd:annotation>
+ <xsd:documentation>
+ name of the address this mirror connection applies to
Review Comment:
This is still questionable for me. I don't know what is the better name for this attribute. Because any *-match name will suggest same behavior as in receiver/sender and I suspect that nobody will want to create mirror for each address. I chose this name because it has same meaning as in the cluster configuration and I didn't want to expand terminology.
So I am still in doubt. I don't think we should align mirror and receiver/sender filter behavior because they have different use cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] iliya-gr commented on a diff in pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
iliya-gr commented on code in PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#discussion_r860274253
##########
artemis-server/src/main/resources/schema/artemis-configuration.xsd:
##########
@@ -2442,6 +2442,13 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
+ <xsd:attribute name="address" type="xsd:string" use="optional">
+ <xsd:annotation>
+ <xsd:documentation>
+ name of the address this mirror connection applies to
Review Comment:
I agree with all of this, maybe except reusing `address-match`. I think using the same term that has different semantic/behaviour is bad and will lead users to misconfiguration.
Anyway, I have opted for `address-filter`, but if there are any objections I can change it to anything else.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] gemmellr commented on a diff in pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
gemmellr commented on code in PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#discussion_r859618348
##########
artemis-server/src/main/resources/schema/artemis-configuration.xsd:
##########
@@ -2442,6 +2442,13 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
+ <xsd:attribute name="address" type="xsd:string" use="optional">
+ <xsd:annotation>
+ <xsd:documentation>
+ name of the address this mirror connection applies to
Review Comment:
Its not just a simple address name it is taking, so this feels like the wrong attribute name and description. Something conveying the 'filter' nature would seem appropriate. The format of what it does take isnt mentioned , so besides the test usage its unclear how anyone would know. There should be doc updates to e.g https://github.com/apache/activemq-artemis/blob/2.21.0/docs/user-manual/en/amqp-broker-connections.md#mirror-configuration to cover whatever the changes ends up being.
Other 'broker connections' stuff such as the sender+receiver already take "address-match" or "queue-match", as shown in the context of the config change in ConfigurationTest-full-config.xml. It would also seem reasonable to align the terminology between them if possible (and perhaps even the syntax and/or implementation; they are implemented a _completely_ different way currently, which doesnt seem that obvious).
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -337,4 +368,52 @@ public boolean isMirrorController() {
return true;
}
}
+
+ public static class AddressFilter {
Review Comment:
This seems like it could be private, or just be its own class, rather than being public so that AMQPMirrorControllerSourceTest can use it for testing, without ever testing anything general of AMQPMirrorControllerSource itself.
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPReplicaTest.java:
##########
@@ -473,6 +473,52 @@ public void testNoAddressWithAnnotations() throws Exception {
}
}
+ @Test
+ public void testAddressFilterMet() throws Exception {
+ addressTest("test", true, 0);
Review Comment:
Using a 0 wait seems potentially susceptible to failure, as the mirroring isnt synchronous. I know its only checking for > 0 messages being mirrored, but I'd still make it more robust personally. Perhaps just changing it to wait for them all to be mirrored as we do expect them to quickly be. Actually, no need for sending 200 messages either given it doesnt actually look at any of them, sending a couple is enough, and it should look.
I would also remove the 1 second wait in the 'no match' case and replace it with better verification by other means, e.g monitor the mirroring SnF queue having not had any messages added when the send is done, or just nothing being present for that target even when subsequently-sent matching stuff has already been received from the other, or both
E.g combine this into one test that sends a non-matching and then matching message in that order and then in the reverse order checks the matching consumer does get a message (receive(timeout)) and the non-matching consumer does not (receiveNoWait) from their respective target queues. Faster and more reliable, verification of what actually happened / actual use. (It could perhaps even monitor the mirror SnF queue during and after, verifying nothing got added to the SnF queue for the non-matching case, and is added for the matching case).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] clebertsuconic commented on pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
clebertsuconic commented on PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#issuecomment-1110127777
it looks good to me...
I will not merge it now as the release I'm doing today is named 2.21.1...
I will defer it to 2.22.0 in a couple weeks...
can you squash these commits and push -f?
(git rebase -I, using a fix on the second commit to the first)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] clebertsuconic commented on pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
clebertsuconic commented on PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#issuecomment-1118080195
thanks @iliya-gr and @gemmellr
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] gemmellr commented on a diff in pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
gemmellr commented on code in PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#discussion_r860865963
##########
docs/user-manual/en/amqp-broker-connections.md:
##########
@@ -100,6 +100,26 @@ Add a `<mirror>` element within the `<amqp-connection>` element to configure mir
The following optional arguments can be utilized:
+* `address-filter`: Each amqp connection mirror configuration only applies to addresses (and its queues) that match the specified `address-filter` attribute, that is, mirroring events will only be sent to the target broker for addresses that match this filter.
+ An address is matched on the mirror amqp connection when it begins with the string specified in this field. The `address-filters` attribute on a mirror amqp connection also supports comma separated lists and an exclude syntax `!`.
+ To prevent an address from being matched on this mirror amqp connection, prepend an address string with `!`. If this attribute is not specified, then all addresses and their queues will be mirrored.
Review Comment:
I would also possibly suggest moving the `address-filter` option down below the others, as they are relatively tiny by comparison.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] iliya-gr commented on a diff in pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
iliya-gr commented on code in PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#discussion_r861006446
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPReplicaTest.java:
##########
@@ -473,6 +473,73 @@ public void testNoAddressWithAnnotations() throws Exception {
}
}
+ @Test
+ public void testAddressFilter() throws Exception {
+ server.start();
+
+ server_2 = createServer(AMQP_PORT_2, false);
+ server_2.setIdentity("server_2");
+ server_2.getConfiguration().setName("server_2");
+
+ AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("mirror-source", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
+ AMQPMirrorBrokerConnectionElement replica = new AMQPMirrorBrokerConnectionElement().setDurable(true).setAddressFilter("replicated,!nonReplicated");
+ amqpConnection.addElement(replica);
+ server_2.getConfiguration().addAMQPConnection(amqpConnection);
+
+ server_2.start();
+
+ {
+ // Send to nonReplicated address
+ ConnectionFactory factory = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT_2);
+ Connection connection = factory.createConnection();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = session.createProducer(session.createQueue("nonReplicated"));
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+ producer.send(session.createTextMessage("never receive"));
+ connection.close();
+ }
+
+ // Check nothing was added to SnF queue
+ Assert.assertEquals(0, server_2.locateQueue(replica.getMirrorSNF()).getMessagesAdded());
+
+ {
+ // Send to replicated address
+ ConnectionFactory factory = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT_2);
+ Connection connection = factory.createConnection();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = session.createProducer(session.createQueue("replicated"));
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+ producer.send(session.createTextMessage("will receive"));
+ connection.close();
+ }
+
+ // Check some messages were sent to SnF queue
+ Assert.assertTrue(server_2.locateQueue(replica.getMirrorSNF()).getMessagesAdded() > 0);
+
+ {
+ ConnectionFactory factory = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT);
+ Connection connection = factory.createConnection();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ connection.start();
+ MessageConsumer consumer = session.createConsumer(session.createQueue("replicated"));
+ Message message = consumer.receive(3000);
+ Assert.assertNotNull(message);
+ Assert.assertEquals("will receive", message.getBody(String.class));
+ connection.close();
+ }
+
+ {
+ ConnectionFactory factory = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT);
+ Connection connection = factory.createConnection();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ connection.start();
+ MessageConsumer consumer = session.createConsumer(session.createQueue("replicated"));
Review Comment:
Indeed, it should be `nonReplicated`. I have added double send to prevent such mistype.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] iliya-gr commented on pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
iliya-gr commented on PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#issuecomment-1116497576
> I will spend some time reviewing this tomorrow after I finish the release on 2.22.
Do I need to squash commits before?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] gemmellr commented on a diff in pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
gemmellr commented on code in PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#discussion_r860822346
##########
docs/user-manual/en/amqp-broker-connections.md:
##########
@@ -100,6 +100,26 @@ Add a `<mirror>` element within the `<amqp-connection>` element to configure mir
The following optional arguments can be utilized:
+* `address-filter`: Each amqp connection mirror configuration only applies to addresses (and its queues) that match the specified `address-filter` attribute, that is, mirroring events will only be sent to the target broker for addresses that match this filter.
+ An address is matched on the mirror amqp connection when it begins with the string specified in this field. The `address-filters` attribute on a mirror amqp connection also supports comma separated lists and an exclude syntax `!`.
+ To prevent an address from being matched on this mirror amqp connection, prepend an address string with `!`. If this attribute is not specified, then all addresses and their queues will be mirrored.
Review Comment:
```suggestion
* `address-filter`: An optional comma-separated list of inclusion and/or exclusion filter entries used to govern which addresses (and related queues) mirroring events will be created for on this broker-connection. That is, events will only be mirrored to the target broker for addresses that match the filter.
An address is matched when it begins with an inclusion entry specified in this field, unless the address is also explicitly excluded by another entry. An exclusion entry is prefixed with `!` to denote any address beginning with that value does not match.
If no inclusion entry is specified in the list, all addresses not explicitly excluded will match. If the address-filter attribute is not specified, then all addresses (and related queues) will match and be mirrored.
```
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPReplicaTest.java:
##########
@@ -473,6 +473,73 @@ public void testNoAddressWithAnnotations() throws Exception {
}
}
+ @Test
+ public void testAddressFilter() throws Exception {
+ server.start();
+
+ server_2 = createServer(AMQP_PORT_2, false);
+ server_2.setIdentity("server_2");
+ server_2.getConfiguration().setName("server_2");
+
+ AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("mirror-source", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
+ AMQPMirrorBrokerConnectionElement replica = new AMQPMirrorBrokerConnectionElement().setDurable(true).setAddressFilter("replicated,!nonReplicated");
+ amqpConnection.addElement(replica);
+ server_2.getConfiguration().addAMQPConnection(amqpConnection);
+
+ server_2.start();
+
+ {
+ // Send to nonReplicated address
+ ConnectionFactory factory = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT_2);
+ Connection connection = factory.createConnection();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = session.createProducer(session.createQueue("nonReplicated"));
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+ producer.send(session.createTextMessage("never receive"));
+ connection.close();
+ }
+
+ // Check nothing was added to SnF queue
+ Assert.assertEquals(0, server_2.locateQueue(replica.getMirrorSNF()).getMessagesAdded());
+
+ {
+ // Send to replicated address
+ ConnectionFactory factory = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT_2);
+ Connection connection = factory.createConnection();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = session.createProducer(session.createQueue("replicated"));
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+ producer.send(session.createTextMessage("will receive"));
Review Comment:
Defining a variable or constnat would make it nicer, and easier to see the relation to the assertion.
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPReplicaTest.java:
##########
@@ -473,6 +473,73 @@ public void testNoAddressWithAnnotations() throws Exception {
}
}
+ @Test
+ public void testAddressFilter() throws Exception {
+ server.start();
+
+ server_2 = createServer(AMQP_PORT_2, false);
+ server_2.setIdentity("server_2");
+ server_2.getConfiguration().setName("server_2");
+
+ AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("mirror-source", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
+ AMQPMirrorBrokerConnectionElement replica = new AMQPMirrorBrokerConnectionElement().setDurable(true).setAddressFilter("replicated,!nonReplicated");
Review Comment:
Defining a variable or constant for the queue names would be nicer to follow than all the literals.
##########
artemis-server/src/main/resources/schema/artemis-configuration.xsd:
##########
@@ -2442,6 +2442,13 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
+ <xsd:attribute name="address" type="xsd:string" use="optional">
+ <xsd:annotation>
+ <xsd:documentation>
+ name of the address this mirror connection applies to
Review Comment:
I agree address-match shouldn't be used if the configuration within it is actually entirely different, as it is now. The comment had the second aspect that its not so obvious to me that it should be entirely different when both things are about governing matched addresses.
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPReplicaTest.java:
##########
@@ -473,6 +473,73 @@ public void testNoAddressWithAnnotations() throws Exception {
}
}
+ @Test
+ public void testAddressFilter() throws Exception {
+ server.start();
+
+ server_2 = createServer(AMQP_PORT_2, false);
+ server_2.setIdentity("server_2");
+ server_2.getConfiguration().setName("server_2");
+
+ AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("mirror-source", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
+ AMQPMirrorBrokerConnectionElement replica = new AMQPMirrorBrokerConnectionElement().setDurable(true).setAddressFilter("replicated,!nonReplicated");
+ amqpConnection.addElement(replica);
+ server_2.getConfiguration().addAMQPConnection(amqpConnection);
+
+ server_2.start();
+
+ {
+ // Send to nonReplicated address
+ ConnectionFactory factory = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT_2);
+ Connection connection = factory.createConnection();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = session.createProducer(session.createQueue("nonReplicated"));
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+ producer.send(session.createTextMessage("never receive"));
+ connection.close();
+ }
Review Comment:
Using try-with-resources around the Connection objects would be slightly simpler + more robust, and implicitly their own code blocks.
EDIT: Although, noting all the connections...reusing the connection on the same broker would seem reasonable in most if not all of the cases, you could just close the producers/consumers.
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPReplicaTest.java:
##########
@@ -473,6 +473,73 @@ public void testNoAddressWithAnnotations() throws Exception {
}
}
+ @Test
+ public void testAddressFilter() throws Exception {
+ server.start();
+
+ server_2 = createServer(AMQP_PORT_2, false);
+ server_2.setIdentity("server_2");
+ server_2.getConfiguration().setName("server_2");
+
+ AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("mirror-source", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
+ AMQPMirrorBrokerConnectionElement replica = new AMQPMirrorBrokerConnectionElement().setDurable(true).setAddressFilter("replicated,!nonReplicated");
+ amqpConnection.addElement(replica);
+ server_2.getConfiguration().addAMQPConnection(amqpConnection);
+
+ server_2.start();
+
+ {
+ // Send to nonReplicated address
+ ConnectionFactory factory = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT_2);
+ Connection connection = factory.createConnection();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = session.createProducer(session.createQueue("nonReplicated"));
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+ producer.send(session.createTextMessage("never receive"));
+ connection.close();
+ }
+
+ // Check nothing was added to SnF queue
+ Assert.assertEquals(0, server_2.locateQueue(replica.getMirrorSNF()).getMessagesAdded());
+
+ {
+ // Send to replicated address
+ ConnectionFactory factory = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT_2);
+ Connection connection = factory.createConnection();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = session.createProducer(session.createQueue("replicated"));
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+ producer.send(session.createTextMessage("will receive"));
+ connection.close();
+ }
+
+ // Check some messages were sent to SnF queue
+ Assert.assertTrue(server_2.locateQueue(replica.getMirrorSNF()).getMessagesAdded() > 0);
+
+ {
+ ConnectionFactory factory = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT);
+ Connection connection = factory.createConnection();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ connection.start();
+ MessageConsumer consumer = session.createConsumer(session.createQueue("replicated"));
+ Message message = consumer.receive(3000);
+ Assert.assertNotNull(message);
+ Assert.assertEquals("will receive", message.getBody(String.class));
+ connection.close();
+ }
+
+ {
+ ConnectionFactory factory = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT);
+ Connection connection = factory.createConnection();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ connection.start();
+ MessageConsumer consumer = session.createConsumer(session.createQueue("replicated"));
Review Comment:
Should this not be nonReplicated?
##########
docs/user-manual/en/amqp-broker-connections.md:
##########
@@ -100,6 +100,26 @@ Add a `<mirror>` element within the `<amqp-connection>` element to configure mir
The following optional arguments can be utilized:
+* `address-filter`: Each amqp connection mirror configuration only applies to addresses (and its queues) that match the specified `address-filter` attribute, that is, mirroring events will only be sent to the target broker for addresses that match this filter.
+ An address is matched on the mirror amqp connection when it begins with the string specified in this field. The `address-filters` attribute on a mirror amqp connection also supports comma separated lists and an exclude syntax `!`.
+ To prevent an address from being matched on this mirror amqp connection, prepend an address string with `!`. If this attribute is not specified, then all addresses and their queues will be mirrored.
+
+ Examples:
+
+ - 'eu'
+ matches all addresses starting with 'eu'
+ - '!eu'
+ matches all address except for those starting with 'eu'
+ - 'eu.uk,eu.de'
+ matches all addresses starting with either 'eu.uk' or 'eu.de'
+ - 'eu,!eu.uk'
+ matches all addresses starting with 'eu' but not those starting with 'eu.uk'
+
+ **Note:**
+
+ - Address exclusion will always take precedence over address inclusion.
+ - Address matching on mirror amqp connections does not support wild-card matching.
Review Comment:
```suggestion
- Address matching on mirror elements is prefix-based and does not support wild-card matching.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] gemmellr commented on pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
gemmellr commented on PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#issuecomment-1119604952
Just to be sure; you didnt have any concerns with the different broker-connections bits having a completely different forms of governing applicable addresses?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] gemmellr commented on pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
gemmellr commented on PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#issuecomment-1119676704
The question was for @clebertsuconic, about the long discussion we already had, where I said it is not obvious to me that two broker-connection elements (which could even be on the same broker-connection at this point) should have completely different ways for doing what is largely the same thing, governing the applicable addresses.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] gemmellr commented on a diff in pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
gemmellr commented on code in PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#discussion_r861671176
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPReplicaTest.java:
##########
@@ -473,6 +473,69 @@ public void testNoAddressWithAnnotations() throws Exception {
}
}
+ @Test
+ public void testAddressFilter() throws Exception {
+ String REPLICATED = "replicated";
+ String NON_REPLICATED = "nonReplicated";
+ String ADDRESS_FILTER = REPLICATED + "," + "!" + NON_REPLICATED;
Review Comment:
Might as well add final if using the caps.
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPReplicaTest.java:
##########
@@ -473,6 +473,69 @@ public void testNoAddressWithAnnotations() throws Exception {
}
}
+ @Test
+ public void testAddressFilter() throws Exception {
+ String REPLICATED = "replicated";
+ String NON_REPLICATED = "nonReplicated";
+ String ADDRESS_FILTER = REPLICATED + "," + "!" + NON_REPLICATED;
+
+ server.start();
+
+ server_2 = createServer(AMQP_PORT_2, false);
+ server_2.setIdentity("server_2");
+ server_2.getConfiguration().setName("server_2");
+
+ AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("mirror-source", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
+ AMQPMirrorBrokerConnectionElement replica = new AMQPMirrorBrokerConnectionElement().setDurable(true).setAddressFilter(ADDRESS_FILTER);
+ amqpConnection.addElement(replica);
+ server_2.getConfiguration().addAMQPConnection(amqpConnection);
+
+ server_2.start();
+
+ try (Connection connection = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT_2).createConnection()) {
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+ // Send to non replicated address
+ try (MessageProducer producer = session.createProducer(session.createQueue(NON_REPLICATED))) {
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+ for (int i = 0; i < 2; i++) {
+ producer.send(session.createTextMessage("never receive"));
+ }
+ }
+
+ // Check nothing was added to SnF queue
+ Assert.assertEquals(0, server_2.locateQueue(replica.getMirrorSNF()).getMessagesAdded());
+
+ // Send to replicated address
+ try (MessageProducer producer = session.createProducer(session.createQueue(REPLICATED))) {
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+ for (int i = 0; i < 2; i++) {
+ producer.send(session.createTextMessage("will receive"));
+ }
Review Comment:
As before, and like the queue/address names, using a variable / constant for the payload here and in the assertion would be good to make things clearer.
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -208,6 +229,10 @@ public void sendMessage(Message message, RoutingContext context, List<MessageRef
return;
}
+ if (ignoreAddress(context.getAddress(message))) {
+ return;
+ }
Review Comment:
Since all the other early-exits trace log why they are doing so, its probably worth this one doing so.
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/MirrorAddressFilter.java:
##########
@@ -0,0 +1,70 @@
+/*
+ * 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.protocol.amqp.connect.mirror;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.activemq.artemis.api.core.SimpleString;
+
+public class MirrorAddressFilter {
+
+ private final SimpleString[] allowList;
+
+ private final SimpleString[] denyList;
+
+ public MirrorAddressFilter(String address) {
Review Comment:
address -> filter
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] clebertsuconic commented on pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
clebertsuconic commented on PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#issuecomment-1115724471
I will spend some time reviewing this tomorrow after I finish the release on 2.22.
it seems in good shape.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] iliya-gr commented on pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
iliya-gr commented on PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#issuecomment-1119629490
Sorry, didn't understand your question. Do you mean wildcard settings? Can you explain in details?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] gemmellr commented on pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
gemmellr commented on PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#issuecomment-1119804229
Not that, although I guess its possible.
Mostly it just seems bizarre for elements of the same thing (a broker connection) to support such entirely distinct ways of doing what amounts to almost exactly the same thing (configuring the applicable addresses). One way supports a list of entries, the other doesnt. One supports wildcarding, the other doesnt. One supports exclusions, the other doesnt.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] clebertsuconic commented on a diff in pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
clebertsuconic commented on code in PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#discussion_r858997595
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -337,4 +368,52 @@ public boolean isMirrorController() {
return true;
}
}
+
+ public static class AddressFilter {
+
+ private final SimpleString[] whiteList;
Review Comment:
whiteList, blackList are now considered offensive terms...
please.. name then allowList, and denyList :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] clebertsuconic merged pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
clebertsuconic merged PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] gemmellr commented on a diff in pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
gemmellr commented on code in PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#discussion_r864642643
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -208,6 +232,13 @@ public void sendMessage(Message message, RoutingContext context, List<MessageRef
return;
}
+ if (ignoreAddress(context.getAddress(message))) {
+ if (logger.isTraceEnabled()) {
+ logger.trace("server " + server + " is discarding send to excluded address");
Review Comment:
In addition to the "excluded" comment, for this one I'd also note it should say _which address_ is in play otherwise its not very helpful.
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -162,6 +172,12 @@ public void createQueue(QueueConfiguration queueConfiguration) throws Exception
}
return;
}
+ if (ignoreAddress(queueConfiguration.getAddress())) {
+ if (logger.isTraceEnabled()) {
+ logger.trace("Skipping create " + queueConfiguration + ", queue address" + queueConfiguration.getAddress() + " is excluded");
Review Comment:
Given the filter does actually have 'exclusions' and it might not actually be excluded, so much as just not-included, I would reword this to avoid saying it is 'excluded' specifically, but instead something more general like 'doesnt match filter'.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] gemmellr commented on pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
gemmellr commented on PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#issuecomment-1117121572
> > I will spend some time reviewing this tomorrow after I finish the release on 2.22.
>
> Do I need to squash commits before?
It can be reviewed without squashing them but its never going to be merged like this without squashing, so I'd suggest just doing so.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] iliya-gr commented on a diff in pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
iliya-gr commented on code in PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#discussion_r864707438
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -208,6 +232,13 @@ public void sendMessage(Message message, RoutingContext context, List<MessageRef
return;
}
+ if (ignoreAddress(context.getAddress(message))) {
+ if (logger.isTraceEnabled()) {
+ logger.trace("server " + server + " is discarding send to excluded address");
Review Comment:
Agree. I have changed both messages.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] clebertsuconic commented on pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
clebertsuconic commented on PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#issuecomment-1118080111
this looks awesome.. merging it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] iliya-gr commented on a diff in pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
iliya-gr commented on code in PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#discussion_r861007512
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPReplicaTest.java:
##########
@@ -473,6 +473,73 @@ public void testNoAddressWithAnnotations() throws Exception {
}
}
+ @Test
+ public void testAddressFilter() throws Exception {
+ server.start();
+
+ server_2 = createServer(AMQP_PORT_2, false);
+ server_2.setIdentity("server_2");
+ server_2.getConfiguration().setName("server_2");
+
+ AMQPBrokerConnectConfiguration amqpConnection = new AMQPBrokerConnectConfiguration("mirror-source", "tcp://localhost:" + AMQP_PORT).setReconnectAttempts(-1).setRetryInterval(100);
+ AMQPMirrorBrokerConnectionElement replica = new AMQPMirrorBrokerConnectionElement().setDurable(true).setAddressFilter("replicated,!nonReplicated");
+ amqpConnection.addElement(replica);
+ server_2.getConfiguration().addAMQPConnection(amqpConnection);
+
+ server_2.start();
+
+ {
+ // Send to nonReplicated address
+ ConnectionFactory factory = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT_2);
+ Connection connection = factory.createConnection();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = session.createProducer(session.createQueue("nonReplicated"));
+ producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+ producer.send(session.createTextMessage("never receive"));
+ connection.close();
+ }
Review Comment:
Everything is in try-with-resources block now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] clebertsuconic commented on pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
clebertsuconic commented on PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#issuecomment-1119793573
@gemmellr can we have a separate discussion on that? I missed it... and I'm not sure I understand it. are you concerned with users messing up the config and having two connections filtering the same address?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] iliya-gr commented on a diff in pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
iliya-gr commented on code in PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#discussion_r859006088
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/AMQPMirrorControllerSource.java:
##########
@@ -337,4 +368,52 @@ public boolean isMirrorController() {
return true;
}
}
+
+ public static class AddressFilter {
+
+ private final SimpleString[] whiteList;
Review Comment:
Thanks for pointing this. We definitely need a maven plugin for this :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [activemq-artemis] gemmellr commented on a diff in pull request #4054: ARTEMIS-3759 Add mirror controller address filter support
Posted by GitBox <gi...@apache.org>.
gemmellr commented on code in PR #4054:
URL: https://github.com/apache/activemq-artemis/pull/4054#discussion_r859833745
##########
artemis-server/src/main/resources/schema/artemis-configuration.xsd:
##########
@@ -2442,6 +2442,13 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
+ <xsd:attribute name="address" type="xsd:string" use="optional">
+ <xsd:annotation>
+ <xsd:documentation>
+ name of the address this mirror connection applies to
Review Comment:
I dont think the use case for this configuration is all that different, in this case its about what things you want to mirror from, vs the others being about what you want to send from or recieve to. An address 'match' for a mirror definition certainly doesnt need to mean 'create a seperate mirror for every address' just because it might mean that for e.g a sender. I also presume it couldnt really mean that given each mirror uses a specific queue named based on its parent broker-connections name.
Using *address-match*, *address-pattern(s)*, *address-filter(s)*, or anything to similar effect, all seem equally reasonable and far more appropriate to me than just "address" does for a configuration that generally seems to be expected _not_ to just have a single address in it.
I'm not saying the config has to be the same, but I am saying it seems more obvious, and I dont as yet see any compelling reason it should be entirely different when its basically for doing entirely equivalent things, configuring applicable addresses of interest.
In any case how it is configured and what it does should be documented.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org