Treat a released outcome for a topic send as a successful send

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

@@ -23,6 +23,7 @@
 import java.util.Map;
 import java.util.concurrent.ScheduledFuture;
 
+import org.apache.qpid.jms.JmsDestination;
 import org.apache.qpid.jms.message.JmsOutboundMessageDispatch;
 import org.apache.qpid.jms.meta.JmsConnectionInfo;
 import org.apache.qpid.jms.meta.JmsProducerInfo;
@@ -250,7 +251,11 @@ private void applyDeliveryStateUpdate(InFlightSend send, Delivery delivery, Deli
                 break;
             case Released:
                 LOG.trace("Outcome of delivery was released: {}", delivery);
-                deliveryError = new ProviderDeliveryReleasedException("Delivery failed: released by receiver");
+                if (isReleasedOutcomeSuccessful(send)) {
+                    send.onSuccess();
+                } else {
+                    deliveryError = new ProviderDeliveryReleasedException("Delivery failed: released by receiver");
+                }
                 break;
             case Modified:
                 LOG.trace("Outcome of delivery was modified: {}", delivery);
@@ -266,6 +271,28 @@ private void applyDeliveryStateUpdate(InFlightSend send, Delivery delivery, Deli
         }
     }
 
+    /**
+     * A released outcome at the target means the transfer "was not and will not be acted upon"
+     * (AMQP 1.0 section 3.4.4). Unlike rejected or modified it carries no indication that
+     * anything went wrong, and for a topic it is the expected result of publishing when no
+     * subscription matched: Jakarta Messaging section 4.2.2 states that a subscription only
+     * receives a copy of a message sent to the topic while the subscription exists, so a
+     * message published to a topic with no matching subscription is simply discarded and the
+     * send succeeds.
+     *
+     * For a queue the same outcome means the message was not stored, so it is still reported as
+     * a send failure rather than silently losing a point-to-point message.
+     *
+     * @param send
+     *      the in-flight send whose delivery was released by the remote.
+     *
+     * @return true if the released outcome should complete the send successfully.
+     */
+    private boolean isReleasedOutcomeSuccessful(InFlightSend send) {
+        JmsDestination destination = send.getEnvelope().getDestination();
+        return destination != null && destination.isTopic();
+    }
+
     public AmqpSession getSession() {
         return session;
     }

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

@@ -1580,6 +1580,112 @@ private void doSyncSendMessageNotAcceptedTestImpl(ListDescribedType responseStat
         }
     }
 
+    @Test
+    @Timeout(20)
+    public void testSyncSendToTopicMessageReleasedCompletesSuccessfully() throws Exception {
+        try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
+            Connection connection = testFixture.establishConnecton(testPeer);
+            testPeer.expectBegin();
+            testPeer.expectSenderAttach();
+
+            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+            Topic topic = session.createTopic("myTopic");
+            MessageProducer producer = session.createProducer(topic);
+            Message message = session.createTextMessage("content");
+
+            testPeer.expectTransfer(new TransferPayloadCompositeMatcher(), nullValue(), new Released(), true);
+            testPeer.expectClose();
+
+            // A released outcome means the remote did not act on the transfer. For a topic that is
+            // just "no subscription took a copy of this message", which the JMS pub/sub model
+            // defines as a successful send, so no exception must escape here.
+            producer.send(message);
+
+            connection.close();
+
+            testPeer.waitForAllHandlersToComplete(2000);
+        }
+    }
+
+    @Test
+    @Timeout(20)
+    public void testAsyncSendToTopicMessageReleasedCompletesSuccessfully() throws Exception {
+        try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
+            JmsConnection connection = (JmsConnection) testFixture.establishConnecton(testPeer);
+
+            final CountDownLatch asyncError = new CountDownLatch(1);
+
+            connection.setForceAsyncSend(true);
+            connection.setExceptionListener(new ExceptionListener() {
+
+                @Override
+                public void onException(JMSException exception) {
+                    LOG.debug("ExceptionListener got error: {}", exception.getMessage());
+                    asyncError.countDown();
+                }
+            });
+
+            testPeer.expectBegin();
+            testPeer.expectSenderAttach();
+            testPeer.expectSenderAttach();
+
+            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+            Topic topic = session.createTopic("myTopic");
+            MessageProducer producer = session.createProducer(topic);
+
+            // Create a second producer which allows for a safe wait for credit for the
+            // first producer without the need for a sleep.  Otherwise the first producer
+            // might not do an actual async send due to not having received credit yet.
+            session.createProducer(topic);
+
+            Message message = session.createTextMessage("content");
+
+            testPeer.expectTransfer(new TransferPayloadCompositeMatcher(), nullValue(), new Released(), true);
+
+            producer.send(message);
+
+            testPeer.waitForAllHandlersToComplete(2000);
+
+            assertFalse(asyncError.await(50, TimeUnit.MILLISECONDS),
+                "Released outcome for a topic send should not be reported as an async error");
+
+            testPeer.expectClose();
+            connection.close();
+
+            testPeer.waitForAllHandlersToComplete(2000);
+        }
+    }
+
+    @Test
+    @Timeout(20)
+    public void testAsyncCompletionSendToTopicMessageReleasedCompletesSuccessfully() throws Exception {
+        try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
+            Connection connection = testFixture.establishConnecton(testPeer);
+
+            testPeer.expectBegin();
+            testPeer.expectSenderAttach();
+
+            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+            Topic topic = session.createTopic("myTopic");
+            MessageProducer producer = session.createProducer(topic);
+            Message message = session.createTextMessage("content");
+
+            testPeer.expectTransfer(new TransferPayloadCompositeMatcher(), nullValue(), new Released(), true);
+
+            TestJmsCompletionListener listener = new TestJmsCompletionListener();
+            producer.send(message, listener);
+
+            assertTrue(listener.awaitCompletion(5, TimeUnit.SECONDS), "Did not get async callback");
+            assertNull(listener.exception, "Released outcome for a topic send should complete the send");
+            assertNotNull(listener.message);
+
+            testPeer.expectClose();
+            connection.close();
+
+            testPeer.waitForAllHandlersToComplete(2000);
+        }
+    }
+
     @Test
     @Timeout(20)
     public void testAsyncSendMessageRejected() throws Exception {