You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by gt...@apache.org on 2016/04/13 13:02:34 UTC
activemq git commit: 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
Repository: activemq
Updated Branches:
refs/heads/master 08d797740 -> 77d46dc13
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
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/77d46dc1
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/77d46dc1
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/77d46dc1
Branch: refs/heads/master
Commit: 77d46dc139284141336591af649c0cc1da790cd0
Parents: 08d7977
Author: gtully <ga...@gmail.com>
Authored: Tue Apr 12 17:24:01 2016 +0100
Committer: gtully <ga...@gmail.com>
Committed: Wed Apr 13 11:55:59 2016 +0100
----------------------------------------------------------------------
.../org/apache/activemq/ActiveMQConnection.java | 2 +-
.../org/apache/activemq/TransactionContext.java | 2 +-
.../org/apache/activemq/bugs/AMQ6240Test.java | 92 ++++++++++++++++++++
.../apache/activemq/bugs/JmsTimeoutTest.java | 8 +-
4 files changed, 98 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/77d46dc1/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java
----------------------------------------------------------------------
diff --git a/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java b/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java
index f77f879..11cb74e 100755
--- a/activemq-client/src/main/java/org/apache/activemq/ActiveMQConnection.java
+++ b/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
http://git-wip-us.apache.org/repos/asf/activemq/blob/77d46dc1/activemq-client/src/main/java/org/apache/activemq/TransactionContext.java
----------------------------------------------------------------------
diff --git a/activemq-client/src/main/java/org/apache/activemq/TransactionContext.java b/activemq-client/src/main/java/org/apache/activemq/TransactionContext.java
index 77826b1..d59604a 100755
--- a/activemq-client/src/main/java/org/apache/activemq/TransactionContext.java
+++ b/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();
http://git-wip-us.apache.org/repos/asf/activemq/blob/77d46dc1/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ6240Test.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ6240Test.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ6240Test.java
new file mode 100644
index 0000000..fadb350
--- /dev/null
+++ b/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);
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq/blob/77d46dc1/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/JmsTimeoutTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/JmsTimeoutTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/JmsTimeoutTest.java
index 2858302..5a233ea 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/JmsTimeoutTest.java
+++ b/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;
- private final int messageCount=10000;
- private final AtomicInteger exceptionCount = new AtomicInteger(0);
+ protected final int messageSize=1024*64;
+ protected final int messageCount=10000;
+ 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++) {