From f936806c9c9c9ca376c9baf180f8b07f91e16486 Mon Sep 17 00:00:00 2001 From: David Ansari Date: Wed, 26 Aug 2026 14:53:02 +0200 Subject: [PATCH] Fail consumer creation when the remote will not apply the selector A consumer created with a message selector silently behaved as if it had no selector at all when the remote declined to apply the requested filter. Every message on the destination was handed to the application, including the ones the selector was supposed to exclude. AMQP 1.0 negotiates a filter in both directions. Section 3.5.3 defines the source filter field as: > The receiving endpoint sets its desired filter, the sending endpoint sets the filter actually in place (including any filters defaulted at the node). The receiving endpoint MUST check that the filter in place meets its needs and take responsibility for detaching if it does not. Section 3.5.1 adds that a filter is a "restriction ... to only deliver a subset of messages" and that the values in the filter-set are the filters the sending endpoint has applied. A peer that will not honour a filter therefore leaves it out of the filter-set it sends in the attach response, and that response is the only way a client learns which filters took effect. We set the jms-selector filter on the source of the attach we send, but never looked at what came back, so we could not tell a peer that applies the selector from one that declines it. This matters wherever a peer supports selectors on some destinations and not others: RabbitMQ, for instance, applies a selector on its JMS queue type but not on classic queues, and correctly omits the filter from the attach response in the latter case. Against such a destination a JMS application asking for "myProp = 'foo'" received everything. Silently widening a selector is a correctness fault: the application cannot detect it, and Jakarta Messaging gives it no reason to expect it. Check the attach response instead, and if the requested selector filter is not in the filter-set actually in place, abort the consumer with a ProviderUnsupportedOperationException, detaching a durable subscription rather than closing it so an existing subscription is not destroyed on the way out. This mirrors how the shared-subscription link capability is already validated in afterOpened(). Tests cover both shapes a declining peer can send: no filter section at all, and an empty filter-set, which is what a broker produces when it strips a filter it will not apply from the set it echoes back. TestAmqpPeer gains a public expectReceiverAttachWithResponseSource so a test can control the filters the peer reports; by default the peer still echoes the source it received, so existing selector tests are unaffected. --- .../amqp/builders/AmqpConsumerBuilder.java | 35 +++++++++++- .../integration/SessionIntegrationTest.java | 57 +++++++++++++++++++ .../qpid/jms/test/testpeer/TestAmqpPeer.java | 10 ++++ 3 files changed, 101 insertions(+), 1 deletion(-) diff --git a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/builders/AmqpConsumerBuilder.java b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/builders/AmqpConsumerBuilder.java index 60b590f89..6a2c2635b 100644 --- a/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/builders/AmqpConsumerBuilder.java +++ b/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/builders/AmqpConsumerBuilder.java @@ -65,6 +65,7 @@ public class AmqpConsumerBuilder extends AmqpResourceBuilder filter = ((Source) remoteSource).getFilter(); + return filter != null && filter.containsKey(JMS_SELECTOR_SYMBOL); + } + + return false; + } + private void configureSource(Source source) { Map filters = new HashMap(); Symbol[] outcomes = new Symbol[]{ Accepted.DESCRIPTOR_SYMBOL, Rejected.DESCRIPTOR_SYMBOL, diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SessionIntegrationTest.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SessionIntegrationTest.java index c4c06d41d..0e2279aee 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/integration/SessionIntegrationTest.java +++ b/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()); + } + + private void doCreateConsumerFailsWhenSelectorFilterNotInPlaceTestImpl(Map 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 { diff --git a/qpid-jms-client/src/test/java/org/apache/qpid/jms/test/testpeer/TestAmqpPeer.java b/qpid-jms-client/src/test/java/org/apache/qpid/jms/test/testpeer/TestAmqpPeer.java index c56341777..2bd4a0ef7 100644 --- a/qpid-jms-client/src/test/java/org/apache/qpid/jms/test/testpeer/TestAmqpPeer.java +++ b/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)