You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ro...@apache.org on 2025/03/28 10:18:08 UTC
(activemq-artemis) branch main updated: ARTEMIS-5368 Propagate Federation consumer priority unchanged
This is an automated email from the ASF dual-hosted git repository.
robbie pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
The following commit(s) were added to refs/heads/main by this push:
new cfffb46149 ARTEMIS-5368 Propagate Federation consumer priority unchanged
cfffb46149 is described below
commit cfffb46149a66a5aedeb5edde017472e5e8d6d71
Author: Timothy Bish <ta...@gmail.com>
AuthorDate: Thu Mar 27 14:52:36 2025 -0400
ARTEMIS-5368 Propagate Federation consumer priority unchanged
When passing along a federation queue consumer that matches the includes
of a federation queue policy retain the original consumer priority so
that if the consumers loop the propagation will stop at the source that
started the loop.
---
.../AMQPFederationAddressPolicyManager.java | 16 +-
.../federation/AMQPFederationConsumerManager.java | 42 +++++-
.../AMQPFederationQueuePolicyManager.java | 165 ++++++++++++---------
.../artemis/core/server/ServerConsumer.java | 40 +++++
.../core/server/impl/ServerConsumerImpl.java | 36 +++++
.../connect/AMQPFederationQueuePolicyTest.java | 104 ++++++++++++-
.../connect/AMQPFederationServerToServerTest.java | 25 +++-
.../tests/integration/cli/DummyServerConsumer.java | 22 +++
8 files changed, 362 insertions(+), 88 deletions(-)
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationAddressPolicyManager.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationAddressPolicyManager.java
index 04540e7d84..10f9bb3fa3 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationAddressPolicyManager.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationAddressPolicyManager.java
@@ -93,7 +93,7 @@ public final class AMQPFederationAddressPolicyManager extends AMQPFederationLoca
@Override
public synchronized void afterRemoveAddress(SimpleString address, AddressInfo addressInfo) throws ActiveMQException {
if (isActive()) {
- final AMQPFederationConsumerManager entry = federationConsumers.remove(address.toString());
+ final AMQPFederationAddressConsumerManager entry = federationConsumers.remove(address.toString());
if (entry != null) {
logger.trace("Federated address {} was removed, closing federation consumer", address);
@@ -357,7 +357,7 @@ public final class AMQPFederationAddressPolicyManager extends AMQPFederationLoca
// current demand and don't need to re-check the server state before trying to create the
// remote address consumer.
if (isActive() && testIfAddressMatchesPolicy(addressName, RoutingType.MULTICAST)) {
- final AMQPFederationConsumerManager entry = federationConsumers.get(addressName);
+ final AMQPFederationAddressConsumerManager entry = federationConsumers.get(addressName);
if (entry != null) {
entry.recover();
@@ -451,7 +451,7 @@ public final class AMQPFederationAddressPolicyManager extends AMQPFederationLoca
return false;
}
- private static class AMQPFederationAddressConsumerManager extends AMQPFederationConsumerManager {
+ private static class AMQPFederationAddressConsumerManager extends AMQPFederationConsumerManager<Binding> {
private final AMQPFederationAddressPolicyManager manager;
private final AddressInfo addressInfo;
@@ -486,5 +486,15 @@ public final class AMQPFederationAddressPolicyManager extends AMQPFederationLoca
protected boolean isPluginBlockingFederationConsumerCreate() {
return manager.isPluginBlockingFederationConsumerCreate(addressInfo);
}
+
+ @Override
+ protected void whenDemandTrackingEntryAdded(Binding entry) {
+ // No current action needed
+ }
+
+ @Override
+ protected void whenDemandTrackingEntryRemoved(Binding entry) {
+ // No current action needed
+ }
}
}
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationConsumerManager.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationConsumerManager.java
index 6c0577ed84..0c8f9968ed 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationConsumerManager.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationConsumerManager.java
@@ -32,8 +32,10 @@ import org.slf4j.LoggerFactory;
* <p>
* All interactions with the consumer tracking entry should occur under the lock of the parent manager instance and this
* manager will perform any asynchronous work with a lock held on the parent manager instance.
+ *
+ * @param <E> The type used in the demand tracking collection.
*/
-public abstract class AMQPFederationConsumerManager {
+public abstract class AMQPFederationConsumerManager<E> {
private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@@ -48,7 +50,7 @@ public abstract class AMQPFederationConsumerManager {
private final AMQPFederation federation;
private final AMQPFederationLocalPolicyManager manager;
- private final Set<Object> demandTracking = new HashSet<>();
+ private final Set<E> demandTracking = new HashSet<>();
private State state = State.READY;
private AMQPFederationConsumer consumer;
@@ -94,6 +96,14 @@ public abstract class AMQPFederationConsumerManager {
} finally {
consumer = null;
+ demandTracking.forEach(entry -> {
+ try {
+ whenDemandTrackingEntryRemoved(entry);
+ } catch (Exception ignore) {
+ }
+ });
+ demandTracking.clear();
+
// Stop handler in closed state won't schedule idle timeout, ensure
// any pending task is canceled just for proper cleanup purposes.
if (pendingIdleTimeout != null) {
@@ -135,10 +145,12 @@ public abstract class AMQPFederationConsumerManager {
*
* @param demand A new unit of demand to add to this consumer manager.
*/
- public void addDemand(Object demand) {
+ public void addDemand(E demand) {
checkClosed();
- demandTracking.add(demand);
+ if (demandTracking.add(demand)) {
+ whenDemandTrackingEntryAdded(demand);
+ }
// This will create a new consumer only if there isn't one currently assigned to this entry and any configured
// federation plugins don't block it from doing so. An already stopping consumer will check on stop if it should
@@ -159,10 +171,12 @@ public abstract class AMQPFederationConsumerManager {
*
* @param demand The element of demand that should be removed from tracking.
*/
- public void removeDemand(Object demand) {
+ public void removeDemand(E demand) {
checkClosed();
- demandTracking.remove(demand);
+ if (demandTracking.remove(demand)) {
+ whenDemandTrackingEntryRemoved(demand);
+ }
if (hasDemand() || state == State.READY) {
return;
@@ -372,6 +386,22 @@ public abstract class AMQPFederationConsumerManager {
}
}
+ /**
+ * An event point that a subclass can use to perform an initialization action whenever an entry is added to
+ * demand tracking.
+ *
+ * @param entry The entry that is being added to demand tracking.
+ */
+ protected abstract void whenDemandTrackingEntryAdded(E entry);
+
+ /**
+ * An event point that a subclass can use to perform a cleanup action whenever an entry is removed from
+ * demand tracking.
+ *
+ * @param entry The entry that is being removed from demand tracking.
+ */
+ protected abstract void whenDemandTrackingEntryRemoved(E entry);
+
/**
* Creates a new federation consumer that this manager will monitor and maintain. The returned consumer should be in
* an initial state ready for this manager to initialize once it is fully configured.
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationQueuePolicyManager.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationQueuePolicyManager.java
index 412004fb89..d6c5bb31ab 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationQueuePolicyManager.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationQueuePolicyManager.java
@@ -23,15 +23,13 @@ import java.lang.invoke.MethodHandles;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
-import java.util.function.Predicate;
+import java.util.UUID;
import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.core.filter.Filter;
-import org.apache.activemq.artemis.core.filter.impl.FilterImpl;
import org.apache.activemq.artemis.core.postoffice.Binding;
import org.apache.activemq.artemis.core.postoffice.QueueBinding;
-import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.ServerConsumer;
import org.apache.activemq.artemis.core.server.ServerSession;
@@ -52,17 +50,22 @@ public final class AMQPFederationQueuePolicyManager extends AMQPFederationLocalP
private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- protected final Predicate<ServerConsumer> federationConsumerMatcher;
protected final FederationReceiveFromQueuePolicy policy;
protected final Map<FederationConsumerInfo, AMQPFederationQueueConsumerManager> federationConsumers = new HashMap<>();
+ /*
+ * Unique for the lifetime of this policy manager which is either the lifetime of the broker or until the
+ * configuration is reloaded and this federation instance happens to be updated but not removed in which
+ * case the full federation instance is shutdown and then removed and re-added as if it was new.
+ */
+ private final String CONSUMER_INFO_ATTACHMENT_KEY = UUID.randomUUID().toString();
+
public AMQPFederationQueuePolicyManager(AMQPFederation federation, AMQPFederationMetrics metrics, FederationReceiveFromQueuePolicy queuePolicy) throws ActiveMQException {
super(federation, metrics, queuePolicy);
Objects.requireNonNull(queuePolicy, "The Queue match policy cannot be null");
this.policy = queuePolicy;
- this.federationConsumerMatcher = createFederationConsumerMatcher(server, queuePolicy);
}
/**
@@ -100,12 +103,12 @@ public final class AMQPFederationQueuePolicyManager extends AMQPFederationLocalP
@Override
public synchronized void afterCloseConsumer(ServerConsumer consumer, boolean failed) {
if (isActive()) {
- final FederationConsumerInfo consumerInfo = createConsumerInfo(consumer);
- final AMQPFederationQueueConsumerManager entry = federationConsumers.get(consumerInfo);
+ final FederationConsumerInfo consumerInfo = (FederationConsumerInfo) consumer.getAttachment(CONSUMER_INFO_ATTACHMENT_KEY);
- if (entry != null) {
+ if (consumerInfo != null && federationConsumers.containsKey(consumerInfo)) {
+ final AMQPFederationQueueConsumerManager entry = federationConsumers.get(consumerInfo);
logger.trace("Reducing demand on federated queue {}", entry.getQueueName());
- entry.removeDemand(identifyConsumer(consumer));
+ entry.removeDemand(consumer);
}
}
}
@@ -156,17 +159,19 @@ public final class AMQPFederationQueuePolicyManager extends AMQPFederationLocalP
final String queueName = consumer.getQueue().getName().toString();
if (testIfQueueMatchesPolicy(consumer.getQueueAddress().toString(), queueName)) {
+ final boolean federationConsumer = isFederationConsumer(consumer);
+
// We should ignore federation consumers from remote peers but configuration does allow
// these to be federated again for some very specific use cases so we check before then
// moving onto any server plugin checks kick in.
- if (federationConsumerMatcher.test(consumer)) {
+ if (federationConsumer && !policy.isIncludeFederated()) {
return;
}
logger.trace("Federation Policy matched on consumer for binding: {}", consumer.getBinding());
final AMQPFederationQueueConsumerManager entry;
- final FederationConsumerInfo consumerInfo = createConsumerInfo(consumer);
+ final FederationConsumerInfo consumerInfo = createConsumerInfo(consumer, federationConsumer);
// Check for existing consumer add demand from a additional local consumer to ensure
// the remote consumer remains active until all local demand is withdrawn.
@@ -174,13 +179,13 @@ public final class AMQPFederationQueuePolicyManager extends AMQPFederationLocalP
logger.trace("Federation Queue Policy manager found existing demand for queue: {}, adding demand", queueName);
entry = federationConsumers.get(consumerInfo);
} else {
- federationConsumers.put(consumerInfo, entry = new AMQPFederationQueueConsumerManager(this, consumerInfo, consumer.getQueue()));
+ federationConsumers.put(consumerInfo, entry = new AMQPFederationQueueConsumerManager(this, CONSUMER_INFO_ATTACHMENT_KEY, consumerInfo, consumer.getQueue()));
}
// Demand passed all binding plugin blocking checks so we track it, plugin can still
// stop federation of the queue based on some external criteria but once it does
// (if ever) allow it we will have tracked all allowed demand.
- entry.addDemand(identifyConsumer(consumer));
+ entry.addDemand(consumer);
}
}
@@ -190,6 +195,7 @@ public final class AMQPFederationQueuePolicyManager extends AMQPFederationLocalP
*
* @param addressName The address that was added on the remote.
* @param queueName The queue that was added on the remote.
+ *
* @throws Exception if an error occurs while processing the queue added event.
*/
public synchronized void afterRemoteQueueAdded(String addressName, String queueName) throws Exception {
@@ -219,6 +225,7 @@ public final class AMQPFederationQueuePolicyManager extends AMQPFederationLocalP
*
* @param address The address that is being tested for a policy match.
* @param queueName The name of the queue that is being tested for a policy match.
+ *
* @return {@code true} if the address given is a match against the policy
*/
private boolean testIfQueueMatchesPolicy(String address, String queueName) {
@@ -231,6 +238,7 @@ public final class AMQPFederationQueuePolicyManager extends AMQPFederationLocalP
* configured matching policy.
*
* @param queueName The name of the queue that is being tested for a policy match.
+ *
* @return {@code true} if the address given is a match against the policy
*/
private boolean testIfQueueMatchesPolicy(String queueName) {
@@ -239,23 +247,20 @@ public final class AMQPFederationQueuePolicyManager extends AMQPFederationLocalP
/**
* Create a new {@link FederationConsumerInfo} based on the given {@link ServerConsumer} and the configured
- * {@link FederationReceiveFromQueuePolicy}. A subclass must override this method to return a consumer information
- * object with additional data used be that implementation.
+ * {@link FederationReceiveFromQueuePolicy}. This should only be called once when a consumer is added and
+ * we begin tracking it as demand on a federated queue.
*
* @param consumer The {@link ServerConsumer} to use as a basis for the consumer information object.
+ * @param federationConsumer Is the consumer one that was created by a remote federation controller
+ *
* @return a new {@link FederationConsumerInfo} instance based on the server consumer
*/
- private FederationConsumerInfo createConsumerInfo(ServerConsumer consumer) {
+ private FederationConsumerInfo createConsumerInfo(ServerConsumer consumer, boolean federationConsumer) {
final Queue queue = consumer.getQueue();
final String queueName = queue.getName().toString();
final String address = queue.getAddress().toString();
-
- final int priority = configuration.isIgnoreSubscriptionPriorities() ?
- ActiveMQDefaultConfiguration.getDefaultConsumerPriority() + policy.getPriorityAjustment() :
- consumer.getPriority() + policy.getPriorityAjustment();
-
- final String filterString =
- selectFilter(queue.getFilter(), configuration.isIgnoreSubscriptionFilters() ? null : consumer.getFilter());
+ final int priority = selectPriority(consumer, federationConsumer);
+ final String filterString = selectFilter(consumer);
return new AMQPFederationGenericConsumerInfo(Role.QUEUE_CONSUMER,
address,
@@ -266,78 +271,84 @@ public final class AMQPFederationQueuePolicyManager extends AMQPFederationLocalP
priority);
}
- @Override
- protected AMQPFederationConsumer createFederationConsumer(FederationConsumerInfo consumerInfo) {
- Objects.requireNonNull(consumerInfo, "Federation Queue consumer information object was null");
+ private String selectFilter(ServerConsumer consumer) {
+ final Filter consumerFilter;
+ final Filter queueFilter = consumer.getQueue().getFilter();
- if (logger.isTraceEnabled()) {
- logger.trace("AMQP Federation {} creating queue consumer: {} for policy: {}", federation.getName(), consumerInfo, policy.getPolicyName());
+ if (!configuration.isIgnoreSubscriptionFilters()) {
+ consumerFilter = consumer.getFilter();
+ } else {
+ consumerFilter = null;
}
- // Don't initiate anything yet as the caller might need to register error handlers etc
- // before the attach is sent otherwise they could miss the failure case.
- return new AMQPFederationQueueConsumer(this, configuration, session, consumerInfo, metrics.newConsumerMetrics());
+ if (consumerFilter != null) {
+ return consumerFilter.getFilterString().toString();
+ } else if (queueFilter != null) {
+ return queueFilter.getFilterString().toString();
+ } else {
+ return null;
+ }
}
- /**
- * Creates a {@link Predicate} that should return true if the given consumer is a federation created consumer which
- * should not be further federated.
- *
- * @param server The server instance for use in creating the filtering {@link Predicate}.
- * @param policy The configured Queue matching policy that can provide additional match criteria.
- * @return a {@link Predicate} that will return true if the consumer should be filtered
- * @throws ActiveMQException if an error occurs while creating the new consumer filter.
- */
- private Predicate<ServerConsumer> createFederationConsumerMatcher(ActiveMQServer server, FederationReceiveFromQueuePolicy policy) throws ActiveMQException {
- if (policy.isIncludeFederated()) {
- return (consumer) -> false; // Configuration says to federate these
+ private int selectPriority(ServerConsumer consumer, boolean federationConsumer) {
+ // Use the priority from the federation consumer as indicated and only choose values for
+ // non-federation consumers to help avoid an infinite descending priority loop when the
+ // federation consumers are included and the configured brokers loop or are bi-directional.
+ if (federationConsumer) {
+ return consumer.getPriority();
+ } else if (configuration.isIgnoreSubscriptionPriorities()) {
+ return ActiveMQDefaultConfiguration.getDefaultConsumerPriority() + policy.getPriorityAjustment();
} else {
- // This filter matches on the same criteria as the original Core client based
- // Federation code which allows this implementation to see those consumers as
- // well as its own which in this methods implementation must also use this same
- // mechanism to mark federation resources.
-
- final Filter metaDataMatcher =
- FilterImpl.createFilter("\"" + FEDERATION_NAME + "\" IS NOT NULL");
-
- return (consumer) -> {
- final ServerSession serverSession = server.getSessionByID(consumer.getSessionID());
-
- if (serverSession != null && serverSession.getMetaData() != null) {
- return metaDataMatcher.match(serverSession.getMetaData());
- } else {
- return false;
- }
- };
+ return consumer.getPriority() + policy.getPriorityAjustment();
}
}
- private static String identifyConsumer(ServerConsumer consumer) {
- return consumer.getConnectionID().toString() + ":" +
- consumer.getSessionID() + ":" +
- consumer.getID();
+ /*
+ * This method matches on the same criteria as the original Core client based Federation code which
+ * allows this implementation to see those consumers as well as its own which in this methods
+ * implementation must also use this same mechanism to mark federation resources.
+ */
+ private boolean isFederationConsumer(ServerConsumer consumer) {
+ final ServerSession serverSession = server.getSessionByID(consumer.getSessionID());
+
+ // Care must be taken to only check this on consumer added and not on other consumer removed
+ // events as the session can be removed before those events are triggered and this will falsely
+ // indicate that the consumer is not a federation consumer. This check works for both AMQP and
+ // Core federation consumers.
+ if (serverSession != null && serverSession.getMetaData() != null) {
+ return serverSession.getMetaData(FEDERATION_NAME) != null;
+ } else {
+ return false;
+ }
}
- private static String selectFilter(Filter queueFilter, Filter consumerFilter) {
- if (consumerFilter != null) {
- return consumerFilter.getFilterString().toString();
- } else {
- return queueFilter != null ? queueFilter.getFilterString().toString() : null;
+ @Override
+ protected AMQPFederationConsumer createFederationConsumer(FederationConsumerInfo consumerInfo) {
+ Objects.requireNonNull(consumerInfo, "Federation Queue consumer information object was null");
+
+ if (logger.isTraceEnabled()) {
+ logger.trace("AMQP Federation {} creating queue consumer: {} for policy: {}", federation.getName(), consumerInfo, policy.getPolicyName());
}
+
+ // Don't initiate anything yet as the caller might need to register error handlers etc
+ // before the attach is sent otherwise they could miss the failure case.
+ return new AMQPFederationQueueConsumer(this, configuration, session, consumerInfo, metrics.newConsumerMetrics());
}
- private static class AMQPFederationQueueConsumerManager extends AMQPFederationConsumerManager {
+ private static class AMQPFederationQueueConsumerManager extends AMQPFederationConsumerManager<ServerConsumer> {
private final AMQPFederationQueuePolicyManager manager;
private final Queue queue;
private final FederationConsumerInfo consumerInfo;
+ private final String policyInfoKey;
- AMQPFederationQueueConsumerManager(AMQPFederationQueuePolicyManager manager, FederationConsumerInfo consumerInfo, Queue queue) {
+ AMQPFederationQueueConsumerManager(AMQPFederationQueuePolicyManager manager, String policyInfoKey, FederationConsumerInfo consumerInfo, Queue queue) {
super(manager);
this.manager = manager;
this.queue = queue;
this.consumerInfo = consumerInfo;
+ this.policyInfoKey = policyInfoKey;
}
/**
@@ -356,5 +367,17 @@ public final class AMQPFederationQueuePolicyManager extends AMQPFederationLocalP
protected boolean isPluginBlockingFederationConsumerCreate() {
return manager.isPluginBlockingFederationConsumerCreate(queue);
}
+
+ @Override
+ protected void whenDemandTrackingEntryRemoved(ServerConsumer consumer) {
+ consumer.removeAttachment(policyInfoKey);
+ }
+
+ @Override
+ protected void whenDemandTrackingEntryAdded(ServerConsumer consumer) {
+ // Attach the consumer info to the server consumer for later use on consumer close or other
+ // operations that need to retrieve the data used to create the federation consumer identity.
+ consumer.addAttachment(policyInfoKey, consumerInfo);
+ }
}
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ServerConsumer.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ServerConsumer.java
index cc0bc74083..0b02a74f66 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ServerConsumer.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ServerConsumer.java
@@ -17,6 +17,7 @@
package org.apache.activemq.artemis.core.server;
import java.util.List;
+import java.util.Map;
import java.util.function.Function;
import org.apache.activemq.artemis.core.transaction.Transaction;
@@ -108,4 +109,43 @@ public interface ServerConsumer extends Consumer, ConsumerInfo {
* @param transaction the tx
*/
void metricsAcknowledge(MessageReference ref, Transaction transaction);
+
+ /**
+ * Adds the given attachment to the {@link ServerConsumer} which will overwrite any previously
+ * assigned value with the same key.
+ *
+ * @param key
+ * The key used to identify the attachment.
+ * @param attachment
+ * The actual value to store for the assigned key.
+ */
+ void addAttachment(String key, Object attachment);
+
+ /**
+ * Remove the any attachment entry from the {@link ServerConsumer} clearing any previously assigned value
+ *
+ * @param key
+ * The key used to identify the attachment.
+ */
+ void removeAttachment(String key);
+
+ /**
+ * Gets any attachment that has been assigned to this {@link ServerConsumer} using the provided key.
+ * If no value was assigned a null is returned.
+ *
+ * @param key
+ * The key identifying the target attachment.
+ *
+ * @return the assigned value associated with the given key or null if nothing assigned.
+ */
+ Object getAttachment(String key);
+
+ /**
+ * Provides access to the full {@link Map} of consumer attachments in an unmodifiable {@link Map} instance.
+ * If no attachments are assigned to the consumer an empty {@link Map} instance is returned, never null.
+ *
+ * @return an unmodifiable {@link Map} that carries all consumer attachments.
+ */
+ Map<String, Object> getAttachments();
+
}
\ No newline at end of file
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java
index d8b76a8b07..6977005735 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java
@@ -21,9 +21,11 @@ import java.nio.ByteBuffer;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
@@ -163,6 +165,8 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
private boolean isClosed = false;
+ private Map<String, Object> attachments;
+
@Override
public boolean isClosed() {
return isClosed;
@@ -638,6 +642,8 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
callback = null;
session = null;
+
+ attachments = null;
});
}
@@ -1609,6 +1615,36 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
return callback;
}
+ @Override
+ public synchronized void addAttachment(String key, Object attachment) {
+ if (attachments == null) {
+ attachments = new HashMap<>();
+ }
+
+ attachments.put(key, attachment);
+ }
+
+ @Override
+ public synchronized void removeAttachment(String key) {
+ if (attachments != null) {
+ attachments.remove(key);
+ }
+ }
+
+ @Override
+ public synchronized Object getAttachment(String key) {
+ if (attachments != null) {
+ return attachments.get(key);
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public synchronized Map<String, Object> getAttachments() {
+ return attachments != null ? Collections.unmodifiableMap(attachments) : Collections.emptyMap();
+ }
+
static class ServerConsumerMetrics extends TransactionOperationAbstract {
/**
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPFederationQueuePolicyTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPFederationQueuePolicyTest.java
index 146e3288eb..37785e53b8 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPFederationQueuePolicyTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPFederationQueuePolicyTest.java
@@ -2544,14 +2544,14 @@ public class AMQPFederationQueuePolicyTest extends AmqpClientTestSupport {
target.waitForScriptToComplete(5, TimeUnit.SECONDS);
// broker should create a new receiver that extends the federation receiver to this "broker"
- // but because this is a federation of a federation the priority should drop by an additional
- // increment as we apply the adjustment on each step
+ // but because this is a federation of a federation the priority should remain unchanged from
+ // what the source broker requested when creating the initial consumer.
target.expectAttach().ofReceiver().withDesiredCapability(FEDERATION_QUEUE_RECEIVER.toString())
.withName(allOf(containsString(getTestName()),
containsString("test::test"),
containsString("queue-receiver"),
containsString(server.getNodeID().toString())))
- .withProperty(FEDERATION_RECEIVER_PRIORITY.toString(), DEFAULT_QUEUE_RECEIVER_PRIORITY_ADJUSTMENT - 1).respond()
+ .withProperty(FEDERATION_RECEIVER_PRIORITY.toString(), DEFAULT_QUEUE_RECEIVER_PRIORITY_ADJUSTMENT).respond()
.withOfferedCapabilities(FEDERATION_QUEUE_RECEIVER.toString());
// Should get a flow but if the link goes away quick enough the broker won't get to this before detaching.
target.expectFlow().withLinkCredit(1000).optional();
@@ -4822,6 +4822,104 @@ public class AMQPFederationQueuePolicyTest extends AmqpClientTestSupport {
}
}
+ @Test
+ public void testOverlappingPolicyTargetsRetainDemandTrackingState() throws Exception {
+ try (ProtonTestServer peer = new ProtonTestServer()) {
+ peer.expectSASLAnonymousConnect();
+ peer.expectOpen().respond();
+ peer.expectBegin().respond();
+ peer.expectAttach().ofSender()
+ .withDesiredCapability(FEDERATION_CONTROL_LINK.toString())
+ .respond()
+ .withOfferedCapabilities(FEDERATION_CONTROL_LINK.toString());
+ peer.expectAttach().ofReceiver()
+ .withSenderSettleModeSettled()
+ .withDesiredCapability(FEDERATION_EVENT_LINK.toString())
+ .respondInKind();
+ peer.expectFlow().withLinkCredit(10);
+ peer.start();
+
+ final URI remoteURI = peer.getServerURI();
+ logger.info("Connect test started, peer listening on: {}", remoteURI);
+
+ // Ensure that policies that overlap on the same queue don't break the tracking of federation
+ // consumers for the other policy if they store state data in consumer attachments. We want the
+ // state data to not match between these two so that one could trip up the close event processing
+ // of the other if that was indeed broken.
+
+ final AMQPFederationQueuePolicyElement receiveFromQueue1 = new AMQPFederationQueuePolicyElement();
+ receiveFromQueue1.setName("first");
+ receiveFromQueue1.addToIncludes("#", getTestName());
+ receiveFromQueue1.addProperty(IGNORE_QUEUE_CONSUMER_FILTERS, Boolean.TRUE.toString());
+ receiveFromQueue1.addProperty(QUEUE_RECEIVER_IDLE_TIMEOUT, 10); // Stagger final drain to make frames predictable
+
+ final AMQPFederationQueuePolicyElement receiveFromQueue2 = new AMQPFederationQueuePolicyElement();
+ receiveFromQueue2.setName("second");
+ receiveFromQueue2.addToIncludes("#", getTestName());
+ receiveFromQueue2.addProperty(IGNORE_QUEUE_CONSUMER_FILTERS, Boolean.FALSE.toString());
+ receiveFromQueue2.addProperty(QUEUE_RECEIVER_IDLE_TIMEOUT, 20); // Stagger final drain to make frames predictable
+
+ final AMQPFederatedBrokerConnectionElement element = new AMQPFederatedBrokerConnectionElement();
+ element.setName(getTestName());
+ element.addLocalQueuePolicy(receiveFromQueue1);
+ element.addLocalQueuePolicy(receiveFromQueue2);
+
+ final AMQPBrokerConnectConfiguration amqpConnection =
+ new AMQPBrokerConnectConfiguration(getTestName(), "tcp://" + remoteURI.getHost() + ":" + remoteURI.getPort());
+ amqpConnection.setReconnectAttempts(0);// No reconnects
+ amqpConnection.addElement(element);
+
+ server.getConfiguration().addAMQPConnection(amqpConnection);
+ server.start();
+ server.createQueue(QueueConfiguration.of(getTestName()).setRoutingType(RoutingType.ANYCAST)
+ .setAddress("test")
+ .setAutoCreated(false));
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ peer.expectAttach().ofReceiver()
+ .withDesiredCapability(FEDERATION_QUEUE_RECEIVER.toString())
+ .withName(allOf(containsString(getTestName()),
+ containsString("test::" + getTestName()),
+ containsString(server.getNodeID().toString())))
+ .withProperty(FEDERATION_RECEIVER_PRIORITY.toString(), DEFAULT_QUEUE_RECEIVER_PRIORITY_ADJUSTMENT)
+ .respond()
+ .withOfferedCapabilities(FEDERATION_QUEUE_RECEIVER.toString());
+ peer.expectAttach().ofReceiver()
+ .withDesiredCapability(FEDERATION_QUEUE_RECEIVER.toString())
+ .withName(allOf(containsString(getTestName()),
+ containsString("test::" + getTestName()),
+ containsString(server.getNodeID().toString())))
+ .withProperty(FEDERATION_RECEIVER_PRIORITY.toString(), DEFAULT_QUEUE_RECEIVER_PRIORITY_ADJUSTMENT)
+ .respond()
+ .withOfferedCapabilities(FEDERATION_QUEUE_RECEIVER.toString());
+ peer.expectFlow().withLinkCredit(1000);
+ peer.expectFlow().withLinkCredit(1000);
+
+ final ConnectionFactory factory = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:" + AMQP_PORT);
+
+ try (Connection connection = factory.createConnection()) {
+ final Session session = connection.createSession(Session.AUTO_ACKNOWLEDGE);
+ final MessageConsumer consumer = session.createConsumer(session.createQueue(getTestName()), "color='red' OR color = 'blue'");
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ peer.expectFlow().withLinkCredit(1000).withDrain(true)
+ .respond()
+ .withLinkCredit(0).withDeliveryCount(1000).withDrain(true).afterDelay(2);
+ peer.expectFlow().withLinkCredit(1000).withDrain(true)
+ .respond()
+ .withLinkCredit(0).withDeliveryCount(1000).withDrain(true).afterDelay(2);
+
+ peer.expectDetach().respond();
+ peer.expectDetach().respond();
+
+ consumer.close();
+
+ peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
+ peer.close();
+ }
+ }
+ }
+
private static void sendQueueAddedEvent(ProtonTestPeer peer, String address, String queue, int handle, int deliveryId) {
final Map<String, Object> eventMap = new LinkedHashMap<>();
eventMap.put(REQUESTED_ADDRESS_NAME, address);
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPFederationServerToServerTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPFederationServerToServerTest.java
index ab0654e7ff..3e912931b1 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPFederationServerToServerTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/connect/AMQPFederationServerToServerTest.java
@@ -17,6 +17,7 @@
package org.apache.activemq.artemis.tests.integration.amqp.connect;
+import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.IGNORE_QUEUE_CONSUMER_PRIORITIES;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.PULL_RECEIVER_BATCH_SIZE;
import static org.apache.activemq.artemis.protocol.amqp.connect.federation.AMQPFederationConstants.RECEIVER_CREDITS;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
@@ -1493,28 +1494,40 @@ public class AMQPFederationServerToServerTest extends AmqpClientTestSupport {
@RepeatedTest(1)
@Timeout(20)
public void testTwoPullConsumerOnPullingFederationConfigurationEachCanTakeOneMessageProduceOnLocal() throws Exception {
- doTestTwoPullConsumerOnPullingFederationConfigurationEachCanTakeOneMessage(true, false);
+ doTestTwoPullConsumerOnPullingFederationConfigurationEachCanTakeOneMessage(true, false, true);
}
@RepeatedTest(1)
@Timeout(20)
public void testTwoPullConsumerOnPullingFederationConfigurationEachCanTakeOneMessageProduceOnRemote() throws Exception {
- doTestTwoPullConsumerOnPullingFederationConfigurationEachCanTakeOneMessage(false, false);
+ doTestTwoPullConsumerOnPullingFederationConfigurationEachCanTakeOneMessage(false, false, true);
}
@RepeatedTest(1)
@Timeout(20)
public void testTwoPullConsumerOnPullingFederationConfigurationEachCanTakeOneMessageProduceOnLocalIncludeFederated() throws Exception {
- doTestTwoPullConsumerOnPullingFederationConfigurationEachCanTakeOneMessage(true, true);
+ doTestTwoPullConsumerOnPullingFederationConfigurationEachCanTakeOneMessage(true, true, true);
}
@RepeatedTest(1)
@Timeout(20)
public void testTwoPullConsumerOnPullingFederationConfigurationEachCanTakeOneMessageProduceOnRemoteIncludeFederated() throws Exception {
- doTestTwoPullConsumerOnPullingFederationConfigurationEachCanTakeOneMessage(false, true);
+ doTestTwoPullConsumerOnPullingFederationConfigurationEachCanTakeOneMessage(false, true, true);
}
- public void doTestTwoPullConsumerOnPullingFederationConfigurationEachCanTakeOneMessage(boolean produceLocal, boolean includeFederated) throws Exception {
+ @RepeatedTest(1)
+ @Timeout(20)
+ public void testTwoPullConsumerOnPullingFederationConfigurationEachCanTakeOneMessageProduceOnLocalIncludeFederatedAmdPriority() throws Exception {
+ doTestTwoPullConsumerOnPullingFederationConfigurationEachCanTakeOneMessage(true, true, false);
+ }
+
+ @RepeatedTest(1)
+ @Timeout(20)
+ public void testTwoPullConsumerOnPullingFederationConfigurationEachCanTakeOneMessageProduceOnRemoteIncludeFederatedAndPriority() throws Exception {
+ doTestTwoPullConsumerOnPullingFederationConfigurationEachCanTakeOneMessage(false, true, false);
+ }
+
+ public void doTestTwoPullConsumerOnPullingFederationConfigurationEachCanTakeOneMessage(boolean produceLocal, boolean includeFederated, boolean ignorePriority) throws Exception {
logger.info("Test started: {}", getTestName());
final AMQPFederationQueuePolicyElement localQueuePolicy = new AMQPFederationQueuePolicyElement();
@@ -1523,6 +1536,7 @@ public class AMQPFederationServerToServerTest extends AmqpClientTestSupport {
localQueuePolicy.addProperty(RECEIVER_CREDITS, 0); // Enable Pull mode
localQueuePolicy.addProperty(PULL_RECEIVER_BATCH_SIZE, 1); // Pull mode batch is one
localQueuePolicy.setIncludeFederated(includeFederated);
+ localQueuePolicy.addProperty(IGNORE_QUEUE_CONSUMER_PRIORITIES, Boolean.toString(ignorePriority));
final AMQPFederationQueuePolicyElement remoteQueuePolicy = new AMQPFederationQueuePolicyElement();
remoteQueuePolicy.setName("test-policy-2");
@@ -1530,6 +1544,7 @@ public class AMQPFederationServerToServerTest extends AmqpClientTestSupport {
remoteQueuePolicy.addProperty(RECEIVER_CREDITS, 0); // Enable Pull mode
remoteQueuePolicy.addProperty(PULL_RECEIVER_BATCH_SIZE, 1); // Pull mode batch is one
remoteQueuePolicy.setIncludeFederated(includeFederated);
+ remoteQueuePolicy.addProperty(IGNORE_QUEUE_CONSUMER_PRIORITIES, Boolean.toString(ignorePriority));
final AMQPFederatedBrokerConnectionElement element = new AMQPFederatedBrokerConnectionElement();
element.setName(getTestName());
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cli/DummyServerConsumer.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cli/DummyServerConsumer.java
index 8fae3800bf..6483f07ce9 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cli/DummyServerConsumer.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cli/DummyServerConsumer.java
@@ -16,7 +16,9 @@
*/
package org.apache.activemq.artemis.tests.integration.cli;
+import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.function.Function;
import org.apache.activemq.artemis.api.core.RoutingType;
@@ -328,4 +330,24 @@ public class DummyServerConsumer implements ServerConsumer {
public int getMessagesAcknowledgedAwaitingCommit() {
return 0;
}
+
+ @Override
+ public void addAttachment(String key, Object attachment) {
+
+ }
+
+ @Override
+ public Object getAttachment(String key) {
+ return null;
+ }
+
+ @Override
+ public Map<String, Object> getAttachments() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public void removeAttachment(String key) {
+
+ }
}
---------------------------------------------------------------------
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