https://issues.apache.org/jira/browse/AMQ-6240 use sendTimout on sync rollback on close such that a blocked connection won't block a close

activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java

@@ -1429,7 +1429,7 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon
         }
     }
 
-    private Response doSyncSendPacket(Command command, int timeout)
+    protected Response doSyncSendPacket(Command command, int timeout)
             throws JMSException {
         try {
             Response response = (Response) (timeout > 0

activemq-client/src/main/java/org/apache/activemq/TransactionContext.java

@@ -277,7 +277,7 @@ public class TransactionContext implements XAResource {
             TransactionInfo info = new TransactionInfo(getConnectionId(), transactionId, TransactionInfo.ROLLBACK);
             this.transactionId = null;
             //make this synchronous - see https://issues.apache.org/activemq/browse/AMQ-2364
-            this.connection.syncSendPacket(info);
+            this.connection.doSyncSendPacket(info, this.connection.getSendTimeout() > 0 && this.connection.isClosing() ? this.connection.getSendTimeout() : 0);
             // Notify the listener that the tx was rolled back
             if (localTransactionEventListener != null) {
                 localTransactionEventListener.rollbackEvent();

activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ6240Test.java

@@ -0,0 +1,92 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.bugs;
+
+import org.apache.activemq.ActiveMQConnection;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.EmbeddedBrokerTestSupport;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.transport.RequestTimedOutIOException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.jms.*;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class AMQ6240Test extends JmsTimeoutTest {
+
+    static final Logger LOG = LoggerFactory.getLogger(AMQ6240Test.class);
+
+
+    public boolean isPersistent() { return true;}
+
+    public void testBlockedTxProducerConnectionTimeoutConnectionCanClose() throws Exception {
+        final ActiveMQConnection cx = (ActiveMQConnection)createConnection();
+        final ActiveMQDestination queue = createDestination("noPfc");
+
+        // we should not take longer than 10 seconds to return from send
+        cx.setSendTimeout(10000);
+
+        Runnable r = new Runnable() {
+            public void run() {
+                try {
+                    LOG.info("Sender thread starting");
+                    Session session = cx.createSession(true, Session.SESSION_TRANSACTED);
+                    MessageProducer producer = session.createProducer(queue);
+                    producer.setDeliveryMode(DeliveryMode.PERSISTENT);
+
+                    TextMessage message = session.createTextMessage(createMessageText());
+                    for(int count=0; count<messageCount; count++){
+                        producer.send(message);
+                    }
+                    LOG.info("Done sending..");
+                } catch (JMSException e) {
+                    if (e.getCause() instanceof RequestTimedOutIOException) {
+                        exceptionCount.incrementAndGet();
+                    } else {
+                        e.printStackTrace();
+                    }
+                    return;
+                }
+
+            }
+        };
+        cx.start();
+        Thread producerThread = new Thread(r);
+        producerThread.start();
+        producerThread.join(15000);
+        cx.close();
+        // We should have a few timeout exceptions as memory store will fill up
+        assertTrue("No exception from the broker", exceptionCount.get() > 0);
+    }
+
+
+    protected void setUp() throws Exception {
+        super.setUp();
+
+        PolicyMap policyMap = new PolicyMap();
+        PolicyEntry noProducerFlowControl = new PolicyEntry();
+        noProducerFlowControl.setProducerFlowControl(false);
+        policyMap.put(new ActiveMQQueue("noPfc"), noProducerFlowControl);
+        broker.setDestinationPolicy(policyMap);
+        broker.getSystemUsage().getStoreUsage().setLimit(50*1024*1024);
+
+    }
+}

activemq-unit-tests/src/test/java/org/apache/activemq/bugs/JmsTimeoutTest.java

@@ -38,9 +38,9 @@ public class JmsTimeoutTest extends EmbeddedBrokerTestSupport {
 
     static final Logger LOG = LoggerFactory.getLogger(JmsTimeoutTest.class);
 
-    private final int messageSize=1024*64;
+    protected final int messageSize=1024*64;
-    private final int messageCount=10000;
+    protected final int messageCount=10000;
-    private final AtomicInteger exceptionCount = new AtomicInteger(0);
+    protected final AtomicInteger exceptionCount = new AtomicInteger(0);
 
     /**
      * Test the case where the broker is blocked due to a memory limit
@@ -145,7 +145,7 @@ public class JmsTimeoutTest extends EmbeddedBrokerTestSupport {
             broker.getTransportConnectors().get(0).getPublishableConnectString());
     }
 
-    private String createMessageText() {
+    protected String createMessageText() {
         StringBuffer buffer = new StringBuffer();
         buffer.append("<filler>");
         for (int i = buffer.length(); i < messageSize; i++) {