Check filter names in attach response

Author: acogoluegnes

.github/workflows/test-pr.yml

@@ -25,7 +25,7 @@ jobs:
       - name: Start broker
         run: ci/start-broker.sh
         env:
-          RABBITMQ_IMAGE: 'pivotalrabbitmq/rabbitmq:amqp-filtex'
+          RABBITMQ_IMAGE: 'pivotalrabbitmq/rabbitmq:main'
       - name: Start toxiproxy
         run: ci/start-toxiproxy.sh
       - name: Display Java version

src/main/java/com/rabbitmq/client/amqp/impl/AmqpConnection.java

@@ -72,6 +72,7 @@ final class AmqpConnection extends ResourceBase implements Connection {
   private final ConnectionSettings.AffinityStrategy affinityStrategy;
   private final String name;
   private final Lock instanceLock = new ReentrantLock();
+  private final boolean filterExpressionsSupported;
 
   AmqpConnection(AmqpConnectionBuilder builder) {
     super(builder.listeners());
@@ -126,6 +127,8 @@ final class AmqpConnection extends ResourceBase implements Connection {
             ConnectionUtils.NO_RETRY_STRATEGY,
             this.name());
     this.sync(ncw);
+    this.filterExpressionsSupported =
+        Utils.supportFilterExpressions(brokerVersion(this.nativeConnection));
     LOGGER.debug("Opened connection '{}' on node '{}'.", this.name(), this.connectionNodename());
     this.state(OPEN);
     this.environment.metricsCollector().openConnection();
@@ -672,6 +675,10 @@ ConnectionUtils.AffinityContext affinity() {
     return this.affinity;
   }
 
+  boolean filterExpressionsSupported() {
+    return this.filterExpressionsSupported;
+  }
+
   long id() {
     return this.id;
   }

src/main/java/com/rabbitmq/client/amqp/impl/AmqpConsumer.java

@@ -175,11 +175,27 @@ private ClientReceiver createNativeReceiver(
               .autoSettle(false)
               .creditWindow(0)
               .properties(properties);
+      Map<String, Object> localSourceFilters = Collections.emptyMap();
       if (!filters.isEmpty()) {
-        receiverOptions.sourceOptions().filters(Map.copyOf(filters));
+        localSourceFilters = Map.copyOf(filters);
+        receiverOptions.sourceOptions().filters(localSourceFilters);
       }
-      return (ClientReceiver)
-          ExceptionUtils.wrapGet(nativeSession.openReceiver(address, receiverOptions).openFuture());
+      ClientReceiver receiver =
+          (ClientReceiver)
+              ExceptionUtils.wrapGet(
+                  nativeSession.openReceiver(address, receiverOptions).openFuture());
+      if (!filters.isEmpty()) {
+        Map<String, String> remoteSourceFilters = receiver.source().filters();
+        for (Map.Entry<String, Object> localEntry : localSourceFilters.entrySet()) {
+          if (!remoteSourceFilters.containsKey(localEntry.getKey())) {
+            LOGGER.warn(
+                "Missing filter value in attach response: {} => {}",
+                localEntry.getKey(),
+                localEntry.getValue());
+          }
+        }
+      }
+      return receiver;
     } catch (ClientException e) {
       throw ExceptionUtils.convert(e, "Error while creating receiver from '%s'", address);
     }

src/main/java/com/rabbitmq/client/amqp/impl/AmqpConsumerBuilder.java

@@ -136,10 +136,10 @@ public Consumer build() {
   private static class DefaultStreamOptions implements StreamOptions {
 
     private final Map<String, DescribedType> filters;
-    private final ConsumerBuilder builder;
+    private final AmqpConsumerBuilder builder;
     private final StreamFilterOptions filterOptions;
 
-    private DefaultStreamOptions(ConsumerBuilder builder, Map<String, DescribedType> filters) {
+    private DefaultStreamOptions(AmqpConsumerBuilder builder, Map<String, DescribedType> filters) {
       this.builder = builder;
       this.filters = filters;
       this.filterOptions = new DefaultStreamFilterOptions(this, filters);
@@ -190,6 +190,10 @@ public StreamOptions filterMatchUnfiltered(boolean matchUnfiltered) {
 
     @Override
     public StreamFilterOptions filter() {
+      if (!this.builder.connection.filterExpressionsSupported()) {
+        throw new IllegalArgumentException(
+            "AMQP filter expressions requires at least RabbitMQ 4.1.0");
+      }
       return this.filterOptions;
     }
 

src/main/java/com/rabbitmq/client/amqp/impl/Utils.java

@@ -224,6 +224,14 @@ static boolean is4_0_OrMore(String brokerVersion) {
     return versionCompare(currentVersion(brokerVersion), "4.0.0") >= 0;
   }
 
+  static boolean is4_1_OrMore(String brokerVersion) {
+    return versionCompare(currentVersion(brokerVersion), "4.1.0") >= 0;
+  }
+
+  static boolean supportFilterExpressions(String brokerVersion) {
+    return is4_1_OrMore(brokerVersion);
+  }
+
   static final class ObservationConnectionInfo implements ObservationCollector.ConnectionInfo {
 
     private final String address;

src/test/java/com/rabbitmq/client/amqp/impl/SourceFiltersTest.java

@@ -20,6 +20,7 @@
 import static com.rabbitmq.client.amqp.ConsumerBuilder.StreamOffsetSpecification.*;
 import static com.rabbitmq.client.amqp.Management.QueueType.STREAM;
 import static com.rabbitmq.client.amqp.impl.Assertions.assertThat;
+import static com.rabbitmq.client.amqp.impl.TestConditions.BrokerVersion.RABBITMQ_4_1_0;
 import static com.rabbitmq.client.amqp.impl.TestUtils.sync;
 import static com.rabbitmq.client.amqp.impl.TestUtils.waitUntilStable;
 import static java.nio.charset.StandardCharsets.*;
@@ -28,6 +29,7 @@
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import com.rabbitmq.client.amqp.*;
+import com.rabbitmq.client.amqp.impl.TestConditions.BrokerVersionAtLeast;
 import com.rabbitmq.client.amqp.impl.TestUtils.Sync;
 import java.time.Duration;
 import java.time.Instant;
@@ -274,6 +276,7 @@ void streamFiltering() {
   }
 
   @Test
+  @BrokerVersionAtLeast(RABBITMQ_4_1_0)
   void filterExpressionApplicationProperties() {
     int messageCount = 10;
     UUID uuid = UUID.randomUUID();
@@ -326,6 +329,7 @@ void filterExpressionApplicationProperties() {
   }
 
   @Test
+  @BrokerVersionAtLeast(RABBITMQ_4_1_0)
   void filterExpressionProperties() {
     int messageCount = 10;
     byte[] userId = "guest".getBytes(UTF_8);
@@ -407,6 +411,7 @@ void filterExpressionProperties() {
   }
 
   @Test
+  @BrokerVersionAtLeast(RABBITMQ_4_1_0)
   void filterExpressionsPropertiesAndApplicationProperties() {
     int messageCount = 10;
     String subject = stringArbitrary.sample();
@@ -438,6 +443,7 @@ void filterExpressionsPropertiesAndApplicationProperties() {
   }
 
   @Test
+  @BrokerVersionAtLeast(RABBITMQ_4_1_0)
   void filterExpressionFilterFewMessagesFromManyToTestFlowControl() {
     String groupId = stringArbitrary.sample();
     publish(1, m -> m.groupId(groupId));
@@ -449,6 +455,7 @@ void filterExpressionFilterFewMessagesFromManyToTestFlowControl() {
   }
 
   @Test
+  @BrokerVersionAtLeast(RABBITMQ_4_1_0)
   void filterExpressionStringModifier() {
     publish(1, m -> m.subject("abc 123"));
     publish(1, m -> m.subject("foo bar"));

src/test/java/com/rabbitmq/client/amqp/impl/TestConditions.java

@@ -34,7 +34,8 @@ public final class TestConditions {
   private TestConditions() {}
 
   public enum BrokerVersion {
-    RABBITMQ_4_0_3("4.0.3");
+    RABBITMQ_4_0_3("4.0.3"),
+    RABBITMQ_4_1_0("4.1.0");
 
     final String value;