Throw exception if echoed filters do not match requested filters

Fixes #217

Author: acogoluegnes

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

@@ -219,6 +219,7 @@ private ClientReceiver createNativeReceiver(
           (ClientReceiver)
               ExceptionUtils.wrapGet(
                   nativeSession.openReceiver(address, receiverOptions).openFuture());
+      boolean filterOk = true;
       if (!filters.isEmpty()) {
         Map<String, String> remoteSourceFilters = receiver.source().filters();
         for (Map.Entry<String, Object> localEntry : localSourceFilters.entrySet()) {
@@ -227,9 +228,15 @@ private ClientReceiver createNativeReceiver(
                 "Missing filter value in attach response: {} => {}",
                 localEntry.getKey(),
                 localEntry.getValue());
+            filterOk = false;
           }
         }
       }
+      if (!filterOk) {
+        receiver.close();
+        throw new AmqpException(
+            "The sending endpoint filters do not match the receiving endpoint filters");
+      }
       return receiver;
     } catch (ClientException e) {
       throw ExceptionUtils.convert(e, "Error while creating receiver from '%s'", address);

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

@@ -28,6 +28,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
+import com.rabbitmq.client.amqp.AmqpException;
 import com.rabbitmq.client.amqp.Connection;
 import com.rabbitmq.client.amqp.Consumer;
 import com.rabbitmq.client.amqp.ConsumerBuilder;
@@ -472,7 +473,7 @@ void filterExpressionStringModifier() {
   @Test
   // TODO should be 4.2
   @BrokerVersionAtLeast(RABBITMQ_4_1_0)
-  void filterExpressionSql() {
+  void sqlFilterExpressionsShouldFilterMessages() {
     publish(1, m -> m.subject("abc 123"));
     publish(1, m -> m.subject("foo bar"));
     publish(1, m -> m.subject("ab 12"));
@@ -484,6 +485,23 @@ void filterExpressionSql() {
     msgs.forEach(m -> assertThat(m).hasSubject("foo bar"));
   }
 
+  @Test
+  // TODO should be 4.2
+  @BrokerVersionAtLeast(RABBITMQ_4_1_0)
+  void incorrectFilterShouldThrowException() {
+    assertThatThrownBy(
+            () ->
+                connection.consumerBuilder().queue(name).messageHandler((ctx, msg) -> {}).stream()
+                    .offset(FIRST)
+                    .filter()
+                    .sql("TRUE TRUE")
+                    .stream()
+                    .builder()
+                    .build())
+        .isInstanceOf(AmqpException.class)
+        .hasMessageContaining("filters do not match");
+  }
+
   void publish(int messageCount) {
     this.publish(messageCount, UnaryOperator.identity());
   }