Fail consumer creation when the remote will not apply the selector

qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/builders/AmqpConsumerBuilder.java

@@ -65,6 +65,7 @@ public class AmqpConsumerBuilder extends AmqpResourceBuilder<AmqpConsumer, AmqpS
 
     boolean validateSharedSubsLinkCapability;
     boolean sharedSubsNotSupported;
+    boolean selectorNotSupported;
 
     public AmqpConsumerBuilder(AmqpSession parent, JmsConsumerInfo consumerInfo) {
         super(parent, consumerInfo);
@@ -156,6 +157,19 @@ protected void afterOpened() {
                 }
             }
         }
+
+        // "the sending endpoint sets the filter actually in place" (AMQP 1.0 section 3.5.3), so a
+        // requested selector that is absent from the attach response is one the remote will not
+        // apply. Carrying on would silently hand the application messages the selector excludes.
+        if (!sharedSubsNotSupported && isSelectorRequested() && !isSelectorFilterInPlace()) {
+            selectorNotSupported = true;
+
+            if (resourceInfo.isDurable()) {
+                endpoint.detach();
+            } else {
+                endpoint.close();
+            }
+        }
     }
 
     @Override
@@ -183,6 +197,11 @@ protected ProviderException getDefaultOpenAbortException() {
             return new ProviderUnsupportedOperationException("Remote peer does not support shared subscriptions");
         }
 
+        if (selectorNotSupported) {
+            return new ProviderUnsupportedOperationException(
+                "Remote peer does not support message selectors on this destination");
+        }
+
         // Verify the attach response contained a non-null Source
         org.apache.qpid.proton.amqp.transport.Source source = endpoint.getRemoteSource();
         if (source != null) {
@@ -197,11 +216,25 @@ protected ProviderException getDefaultOpenAbortException() {
     protected boolean isClosePending() {
         // When no link terminus was created, the peer will now detach/close us otherwise
         // we need to validate the returned remote source prior to open completion.
-        return sharedSubsNotSupported || endpoint.getRemoteSource() == null;
+        return sharedSubsNotSupported || selectorNotSupported || endpoint.getRemoteSource() == null;
     }
 
     //----- Internal implementation ------------------------------------------//
 
+    private boolean isSelectorRequested() {
+        return resourceInfo.getSelector() != null && !resourceInfo.getSelector().trim().equals("");
+    }
+
+    private boolean isSelectorFilterInPlace() {
+        org.apache.qpid.proton.amqp.transport.Source remoteSource = endpoint.getRemoteSource();
+        if (remoteSource instanceof Source) {
+            Map<Symbol, Object> filter = ((Source) remoteSource).getFilter();
+            return filter != null && filter.containsKey(JMS_SELECTOR_SYMBOL);
+        }
+
+        return false;
+    }
+
     private void configureSource(Source source) {
         Map<Symbol, DescribedType> filters = new HashMap<Symbol, DescribedType>();
         Symbol[] outcomes = new Symbol[]{ Accepted.DESCRIPTOR_SYMBOL, Rejected.DESCRIPTOR_SYMBOL,

qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SessionIntegrationTest.java

@@ -34,7 +34,9 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Random;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -55,6 +57,7 @@
 import org.apache.qpid.jms.test.testpeer.basictypes.TerminusDurability;
 import org.apache.qpid.jms.test.testpeer.describedtypes.Accepted;
 import org.apache.qpid.jms.test.testpeer.describedtypes.Rejected;
+import org.apache.qpid.jms.test.testpeer.describedtypes.Source;
 import org.apache.qpid.jms.test.testpeer.describedtypes.sections.AmqpValueDescribedType;
 import org.apache.qpid.jms.test.testpeer.describedtypes.sections.HeaderDescribedType;
 import org.apache.qpid.jms.test.testpeer.matchers.AcceptedMatcher;
@@ -378,6 +381,60 @@ private void doCreateConsumerWithSelectorTestImpl(String messageSelector, boolea
         }
     }
 
+    @Test
+    @Timeout(20)
+    public void testCreateConsumerFailsWhenSelectorFilterNotEchoedAtAll() throws Exception {
+        // Peer answers with a source carrying no filter section.
+        doCreateConsumerFailsWhenSelectorFilterNotInPlaceTestImpl(null);
+    }
+
+    @Test
+    @Timeout(20)
+    public void testCreateConsumerFailsWhenSelectorFilterOmittedFromEchoedFilterSet() throws Exception {
+        // Peer answers with an empty filter set, which is what a broker sends when it strips a
+        // filter it will not apply from the set it echoes back.
+        doCreateConsumerFailsWhenSelectorFilterNotInPlaceTestImpl(new HashMap<Symbol, Object>());
+    }
+
+    private void doCreateConsumerFailsWhenSelectorFilterNotInPlaceTestImpl(Map<Symbol, Object> responseFilter) throws Exception {
+        try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
+            Connection connection = testFixture.establishConnecton(testPeer);
+            connection.start();
+
+            testPeer.expectBegin();
+
+            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+            Queue queue = session.createQueue("myQueue");
+            String selector = "myProp = 'foo'";
+
+            // "the sending endpoint sets the filter actually in place" (AMQP 1.0 section 3.5.3),
+            // so a response without the selector filter says the peer will not apply it. The
+            // consumer must fail rather than silently deliver messages the selector excludes.
+            Source responseSource = new Source();
+            responseSource.setAddress("myQueue");
+            responseSource.setFilter(responseFilter);
+
+            Matcher<?> filterMapMatcher = hasEntry(equalTo(Symbol.valueOf("jms-selector")), notNullValue());
+            SourceMatcher sourceMatcher = new SourceMatcher();
+            sourceMatcher.withFilter(filterMapMatcher);
+
+            testPeer.expectReceiverAttachWithResponseSource(notNullValue(), sourceMatcher, responseSource);
+            testPeer.expectDetach(true, true, true);
+
+            try {
+                session.createConsumer(queue, selector);
+                fail("Expected an exception to be thrown");
+            } catch (JMSException jmse) {
+                // Expected
+            }
+
+            testPeer.expectClose();
+            connection.close();
+
+            testPeer.waitForAllHandlersToComplete(1000);
+        }
+    }
+
     @Test
     @Timeout(20)
     public void testCreateConsumerFailsWhenLinkRefusedAndAttachResponseWriteIsNotDeferred() throws Exception {

qpid-jms-client/src/test/java/org/apache/qpid/jms/test/testpeer/TestAmqpPeer.java

@@ -1519,6 +1519,16 @@ public void expectReceiverAttach(final Matcher<?> linkNameMatcher, final Matcher
         expectReceiverAttach(linkNameMatcher, sourceMatcher, settled, refuseLink, omitDetach, deferAttachResponseWrite, errorType, errorMessage, null, null, null);
     }
 
+    /**
+     * Accepts the receiver attach but answers with the given source instead of echoing back the
+     * source that was received, so that a test can control which filters the peer reports as
+     * actually being in place.
+     */
+    public void expectReceiverAttachWithResponseSource(final Matcher<?> linkNameMatcher, final Matcher<?> sourceMatcher, final Source responseSource)
+    {
+        expectReceiverAttach(linkNameMatcher, sourceMatcher, false, false, false, false, null, null, responseSource, null, null);
+    }
+
     private void expectReceiverAttach(final Matcher<?> linkNameMatcher, final Matcher<?> sourceMatcher, final boolean settled, final boolean refuseLink,
                                      boolean omitDetach, boolean deferAttachResponseWrite, Symbol errorType, String errorMessage, final Source responseSourceOverride,
                                      Matcher<?> desiredCapabilitiesMatcher, Symbol[] offeredCapabilitiesResponse)