ARTEMIS-4642 Fix tracked federated demand when links are rejected or …

…blocked

Under some scenarios federation demand tracking is losing track of total demand
for a federated resource leading to teardown of federated links before all local
demand has been removed from the resource. This occurs most often if the attempts
to establish a federation link are refused because the resource hasn't yet been
created and an eventual attach succeeds, but can also occur in combination with
a plugin blocking or not blocking federation link creation in some cases.

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationAddressConsumer.java

@@ -165,7 +165,7 @@ public synchronized void close() {
          if (started) {
             started = false;
             connection.runLater(() -> {
-               federation.removeLinkClosedInterceptor(consumerInfo.getFqqn());
+               federation.removeLinkClosedInterceptor(consumerInfo.getId());
 
                if (receiver != null) {
                   try {
@@ -350,7 +350,7 @@ private void asyncCreateReceiver() {
                   // Intercept remote close and check for valid reasons for remote closure such as
                   // the remote peer not having a matching queue for this subscription or from an
                   // operator manually closing the link.
-                  federation.addLinkClosedInterceptor(consumerInfo.getFqqn(), remoteCloseInterceptor);
+                  federation.addLinkClosedInterceptor(consumerInfo.getId(), remoteCloseInterceptor);
 
                   receiver = new AMQPFederatedAddressDeliveryReceiver(session, consumerInfo, protonReceiver);
 

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/federation/AMQPFederationQueueConsumer.java

@@ -162,7 +162,7 @@ public synchronized void close() {
          if (started) {
             started = false;
             connection.runLater(() -> {
-               federation.removeLinkClosedInterceptor(consumerInfo.getFqqn());
+               federation.removeLinkClosedInterceptor(consumerInfo.getId());
 
                if (receiver != null) {
                   try {
@@ -341,7 +341,7 @@ private void asyncCreateReceiver() {
                   // Intercept remote close and check for valid reasons for remote closure such as
                   // the remote peer not having a matching queue for this subscription or from an
                   // operator manually closing the link.
-                  federation.addLinkClosedInterceptor(consumerInfo.getFqqn(), remoteCloseIntercepter);
+                  federation.addLinkClosedInterceptor(consumerInfo.getId(), remoteCloseIntercepter);
 
                   receiver = new AMQPFederatedQueueDeliveryReceiver(localQueue, protonReceiver);
 

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/federation/FederationConsumerInfo.java

@@ -38,6 +38,11 @@ enum Role {
       QUEUE_CONSUMER
    }
 
+   /**
+    * @return a unique Id for the consumer being represented.
+    */
+   String getId();
+
    /**
     * @return the type of federation consumer being represented.
     */

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/federation/internal/FederationAddressEntry.java

@@ -18,42 +18,59 @@
 package org.apache.activemq.artemis.protocol.amqp.federation.internal;
 
 import java.util.HashSet;
+import java.util.Objects;
 import java.util.Set;
 
 import org.apache.activemq.artemis.core.postoffice.Binding;
+import org.apache.activemq.artemis.core.server.impl.AddressInfo;
 
 /**
  * An entry type class used to hold a {@link FederationConsumerInternal} and
  * any other state data needed by the manager that is creating them based on the
- * policy configuration for the federation instance.  The entry can be extended
+ * policy configuration for the federation instance. The entry can be extended
  * by federation implementation to hold additional state data for the federation
- * consumer and the managing of its lifetime.
+ * consumer and the managing active demand.
  *
- * This entry type provides a reference counter that can be used to register demand
+ * This entry type provides reference tracking state for current demand (bindings)
  * on a federation resource such that it is not torn down until all demand has been
  * removed from the local resource.
  */
 public class FederationAddressEntry {
 
-   private final FederationConsumerInternal consumer;
-
+   private final AddressInfo addressInfo;
    private final Set<Binding> demandBindings = new HashSet<>();
 
+   private FederationConsumerInternal consumer;
+
    /**
-    * Creates a new address entry with a single reference
+    * Creates a new address entry for tracking demand on a federated address
     *
-    * @param consumer
-    *    The federation consumer that will be carried in this entry.
+    * @param addressInfo
+    *    The address information object that this entry hold demand state for.
     */
-   public FederationAddressEntry(FederationConsumerInternal consumer) {
-      this.consumer = consumer;
+   public FederationAddressEntry(AddressInfo addressInfo) {
+      this.addressInfo = addressInfo;
+   }
+
+   /**
+    * @return the address information that this entry is acting to federate.
+    */
+   public AddressInfo getAddressInfo() {
+      return addressInfo;
    }
 
    /**
     * @return the address that this entry is acting to federate.
     */
    public String getAddress() {
-      return consumer.getConsumerInfo().getAddress();
+      return addressInfo.getName().toString();
+   }
+
+   /**
+    * @return <code>true</code> if a consumer is currently set on this entry.
+    */
+   public boolean hasConsumer() {
+      return consumer != null;
    }
 
    /**
@@ -63,6 +80,30 @@ public FederationConsumerInternal getConsumer() {
       return consumer;
    }
 
+   /**
+    * Sets the consumer assigned to this entry to the given instance.
+    *
+    * @param consumer
+    *    The federation consumer that is currently active for this entry.
+    *
+    * @return this federation address consumer entry.
+    */
+   public FederationAddressEntry setConsumer(FederationConsumerInternal consumer) {
+      Objects.requireNonNull(consumer, "Cannot assign a null consumer to this entry, call clear to unset");
+      this.consumer = consumer;
+      return this;
+   }
+
+   /**
+    * Clears the currently assigned consumer from this entry.
+    *
+    * @return this federation address consumer entry.
+    */
+   public FederationAddressEntry clearConsumer() {
+      this.consumer = null;
+      return this;
+   }
+
    /**
     * @return true if there are bindings that are mapped to this federation entry.
     */
@@ -85,7 +126,7 @@ public FederationAddressEntry addDemand(Binding binding) {
     *
     * @return this federation address consumer entry.
     */
-   public FederationAddressEntry removeDenamd(Binding binding) {
+   public FederationAddressEntry removeDemand(Binding binding) {
       demandBindings.remove(binding);
       return this;
    }