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 2015/05/12 23:24:47 UTC
[1/3] activemq git commit: add some more logging if tests fails
Repository: activemq
Updated Branches:
refs/heads/master e19293de5 -> c89bb7a31
add some more logging if tests fails
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/d4672200
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/d4672200
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/d4672200
Branch: refs/heads/master
Commit: d46722008591a575815471ff3c3a50e92a20e681
Parents: e19293d
Author: gtully <ga...@gmail.com>
Authored: Mon May 11 16:45:59 2015 +0100
Committer: gtully <ga...@gmail.com>
Committed: Tue May 12 22:10:56 2015 +0100
----------------------------------------------------------------------
.../activemq/network/FailoverStaticNetworkTest.java | 14 +++++++++-----
1 file changed, 9 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/d4672200/activemq-unit-tests/src/test/java/org/apache/activemq/network/FailoverStaticNetworkTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/network/FailoverStaticNetworkTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/network/FailoverStaticNetworkTest.java
index 4113f14..fe1cddc 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/network/FailoverStaticNetworkTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/network/FailoverStaticNetworkTest.java
@@ -41,6 +41,7 @@ import javax.net.ssl.KeyManager;
import javax.net.ssl.TrustManager;
import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.AutoFailTestSupport;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.SslContext;
import org.apache.activemq.broker.TransportConnector;
@@ -346,7 +347,7 @@ public class FailoverStaticNetworkTest {
while (!done.get()) {
brokerA = createBroker("tcp", "61610", null);
brokerA.setBrokerName("Pair");
- brokerA.setBrokerObjectName(new ObjectName(brokerA.getManagementContext().getJmxDomainName() + ":" + "BrokerName="
+ brokerA.setBrokerObjectName(new ObjectName(brokerA.getManagementContext().getJmxDomainName() + ":" + "brokerName="
+ JMXSupport.encodeObjectNamePart("A") + "," + "Type=Broker"));
((KahaDBPersistenceAdapter)brokerA.getPersistenceAdapter()).getLocker().setLockAcquireSleepInterval(1000);
brokerA.start();
@@ -377,7 +378,7 @@ public class FailoverStaticNetworkTest {
brokerA1 = createBroker("tcp", "61611", null);
brokerA1.setBrokerName("Pair");
// so they can coexist in local jmx we set the object name b/c the brokername identifies the shared store
- brokerA1.setBrokerObjectName(new ObjectName(brokerA.getManagementContext().getJmxDomainName() + ":" + "BrokerName="
+ brokerA1.setBrokerObjectName(new ObjectName(brokerA.getManagementContext().getJmxDomainName() + ":" + "brokerName="
+ JMXSupport.encodeObjectNamePart("A1") + "," + "Type=Broker"));
((KahaDBPersistenceAdapter)brokerA1.getPersistenceAdapter()).getLocker().setLockAcquireSleepInterval(1000);
brokerA1.start();
@@ -394,12 +395,12 @@ public class FailoverStaticNetworkTest {
for (int i=0; i<4; i++) {
BrokerService currentMaster = (i%2 == 0 ? brokerA : brokerA1);
- LOG.info("iteration: " + i + ", using: " + currentMaster.getBrokerObjectName().getKeyProperty("BrokerName"));
+ LOG.info("iteration: " + i + ", using: " + currentMaster.getBrokerObjectName().getKeyProperty("brokerName"));
currentMaster.waitUntilStarted();
doTestNetworkSendReceive(brokerB, currentMaster);
- LOG.info("Stopping " + currentMaster.getBrokerObjectName().getKeyProperty("BrokerName"));
+ LOG.info("Stopping " + currentMaster.getBrokerObjectName().getKeyProperty("brokerName"));
currentMaster.stop();
currentMaster.waitUntilStopped();
}
@@ -434,10 +435,13 @@ public class FailoverStaticNetworkTest {
@Override
public boolean isSatisified() throws Exception {
Message message = consumer.receive(5000);
- LOG.info("from: " + from.getBrokerObjectName().getKeyProperty("BrokerName") + ", received: " + message);
+ LOG.info("from: " + from.getBrokerObjectName().getKeyProperty("brokerName") + ", received: " + message);
return message != null;
}
});
+ if (!gotMessage) {
+ AutoFailTestSupport.dumpAllThreads("noMessage");
+ }
try {
consConn.close();
} catch (JMSException ignored) {
[2/3] activemq git commit:
https://issues.apache.org/jira/browse/AMQ-4897 - rework with test that shows
the blocking inactivity monitor task. Removed some redundant state
Posted by gt...@apache.org.
https://issues.apache.org/jira/browse/AMQ-4897 - rework with test that shows the blocking inactivity monitor task. Removed some redundant state
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/3ef53896
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/3ef53896
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/3ef53896
Branch: refs/heads/master
Commit: 3ef53896917076981c505b087e9f6ff99042aa3f
Parents: d467220
Author: gtully <ga...@gmail.com>
Authored: Tue May 12 21:55:07 2015 +0100
Committer: gtully <ga...@gmail.com>
Committed: Tue May 12 22:10:57 2015 +0100
----------------------------------------------------------------------
.../transport/failover/FailoverTransport.java | 41 +++--
...adInactivityBlockWriteTimeoutClientTest.java | 152 +++++++++++++++++++
2 files changed, 169 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/3ef53896/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
----------------------------------------------------------------------
diff --git a/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java b/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
index c9ef141..728d4b7 100755
--- a/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
+++ b/activemq-client/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
@@ -74,7 +74,6 @@ public class FailoverTransport implements CompositeTransport {
private static final int INFINITE = -1;
private TransportListener transportListener;
private boolean disposed;
- private boolean connected;
private final CopyOnWriteArrayList<URI> uris = new CopyOnWriteArrayList<URI>();
private final CopyOnWriteArrayList<URI> updated = new CopyOnWriteArrayList<URI>();
@@ -91,7 +90,6 @@ public class FailoverTransport implements CompositeTransport {
private final TaskRunnerFactory reconnectTaskFactory;
private final TaskRunner reconnectTask;
private boolean started;
- private boolean initialized;
private long initialReconnectDelay = DEFAULT_INITIAL_RECONNECT_DELAY;
private long maxReconnectDelay = 1000 * 30;
private double backOffMultiplier = 2d;
@@ -198,9 +196,6 @@ public class FailoverTransport implements CompositeTransport {
((Tracked) object).onResponses(command);
}
}
- if (!initialized) {
- initialized = true;
- }
if (command.isConnectionControl()) {
handleConnectionControl((ConnectionControl) command);
@@ -242,35 +237,35 @@ public class FailoverTransport implements CompositeTransport {
}
public final void handleTransportFailure(IOException e) throws InterruptedException {
- synchronized (reconnectMutex) {
- if (shuttingDown) {
- // shutdown info sent and remote socket closed and we see that before a local close
- // let the close do the work
- return;
- }
-
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + " handleTransportFailure: " + e, e);
- }
+ if (shuttingDown) {
+ // shutdown info sent and remote socket closed and we see that before a local close
+ // let the close do the work
+ return;
+ }
- Transport transport = connectedTransport.getAndSet(null);
+ if (LOG.isTraceEnabled()) {
+ LOG.trace(this + " handleTransportFailure: " + e, e);
+ }
- if (transport != null) {
+ // could be blocked in write with the reconnectMutex held, but still needs to be whacked
+ Transport transport = connectedTransport.getAndSet(null);
+ if (transport != null) {
+ disposeTransport(transport);
+ }
- disposeTransport(transport);
+ synchronized (reconnectMutex) {
+ if (transport != null && connectedTransport.get() == null) {
boolean reconnectOk = false;
if (canReconnect()) {
reconnectOk = true;
}
- LOG.warn("Transport (" + transport + ") failed"
+ LOG.warn("Transport (" + connectedTransportURI + ") failed"
+ (reconnectOk ? "," : ", not") + " attempting to automatically reconnect", e);
- initialized = false;
failedConnectTransportURI = connectedTransportURI;
connectedTransportURI = null;
- connected = false;
connectedToPriority = false;
if (reconnectOk) {
@@ -377,7 +372,6 @@ public class FailoverTransport implements CompositeTransport {
}
started = false;
disposed = true;
- connected = false;
if (connectedTransport.get() != null) {
transportToStop = connectedTransport.getAndSet(null);
@@ -1067,7 +1061,6 @@ public class FailoverTransport implements CompositeTransport {
LOG.info("Successfully reconnected to " + uri);
}
- connected = true;
return false;
} catch (Exception e) {
failure = e;
@@ -1262,7 +1255,7 @@ public class FailoverTransport implements CompositeTransport {
@Override
public boolean isConnected() {
- return connected;
+ return connectedTransport.get() != null;
}
@Override
http://git-wip-us.apache.org/repos/asf/activemq/blob/3ef53896/activemq-unit-tests/src/test/java/org/apache/activemq/transport/failover/FailoverReadInactivityBlockWriteTimeoutClientTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/transport/failover/FailoverReadInactivityBlockWriteTimeoutClientTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/transport/failover/FailoverReadInactivityBlockWriteTimeoutClientTest.java
new file mode 100644
index 0000000..95fce82
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/transport/failover/FailoverReadInactivityBlockWriteTimeoutClientTest.java
@@ -0,0 +1,152 @@
+/**
+ * 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.transport.failover;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import javax.jms.Connection;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import org.apache.activemq.ActiveMQConnection;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.JmsTestSupport;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter;
+import org.apache.activemq.transport.TransportListener;
+import org.apache.activemq.util.SocketProxy;
+import org.apache.activemq.util.URISupport;
+import org.apache.activemq.util.Wait;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class FailoverReadInactivityBlockWriteTimeoutClientTest extends JmsTestSupport {
+ private static final Logger LOG = LoggerFactory.getLogger(FailoverReadInactivityBlockWriteTimeoutClientTest.class);
+
+ protected BrokerService createBroker() throws Exception {
+ BrokerService broker = new BrokerService();
+ broker.setDeleteAllMessagesOnStartup(true);
+ KahaDBPersistenceAdapter adapter = new KahaDBPersistenceAdapter();
+ adapter.setConcurrentStoreAndDispatchQueues(false);
+ broker.setPersistenceAdapter(adapter);
+ broker.addConnector("tcp://localhost:0?wireFormat.maxInactivityDuration=0");
+ return broker;
+ }
+
+ public void testBlockedFailoverSendWillReactToReadInactivityTimeout() throws Exception {
+ final ActiveMQQueue dest = new ActiveMQQueue("testClientWriteTimeout");
+ messageTextPrefix = initMessagePrefix(80*1024);
+
+ URI tcpBrokerUri = URISupport.removeQuery(broker.getTransportConnectors().get(0).getConnectUri());
+ LOG.info("consuming using uri: " + tcpBrokerUri);
+
+ ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(tcpBrokerUri);
+ Connection c = factory.createConnection();
+ c.start();
+ Session session = c.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageConsumer consumer = session.createConsumer(dest);
+
+ SocketProxy proxy = new SocketProxy();
+ proxy.setTarget(tcpBrokerUri);
+ proxy.open();
+
+ ActiveMQConnectionFactory pFactory = new ActiveMQConnectionFactory("failover:(" + proxy.getUrl() + "?wireFormat.maxInactivityDuration=5000&ignoreRemoteWireFormat=true)?jms.useAsyncSend=true&trackMessages=true&maxCacheSize=6638400");
+ final ActiveMQConnection pc = (ActiveMQConnection) pFactory.createConnection();
+ final AtomicInteger interruptCounter = new AtomicInteger(0);
+ pc.addTransportListener(new TransportListener() {
+ @Override
+ public void onCommand(Object command) {
+
+ }
+
+ @Override
+ public void onException(IOException error) {
+ LOG.info("Got: " + error);
+
+ }
+
+ @Override
+ public void transportInterupted() {
+ interruptCounter.incrementAndGet();
+ }
+
+ @Override
+ public void transportResumed() {
+
+ }
+ });
+ pc.start();
+
+
+ final int messageCount = 200;
+ final CountDownLatch sentOne = new CountDownLatch(1);
+ ExecutorService executorService = Executors.newCachedThreadPool();
+ executorService.execute(new Runnable() {
+ @Override
+ public void run() {
+ try{
+ Session session = pc.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = session.createProducer(dest);
+ for (int i = 0; i < messageCount; i++) {
+ producer.send(session.createTextMessage(messageTextPrefix + i));
+ sentOne.countDown();
+ }
+ producer.close();
+ session.close();
+ LOG.info("Done with send of: " + messageCount);
+ } catch (Exception ignored) {
+ ignored.printStackTrace();
+ }
+ }
+ });
+
+ sentOne.await(5, TimeUnit.SECONDS);
+ proxy.pause();
+
+ assertTrue("Got interrupted", Wait.waitFor(new Wait.Condition() {
+ @Override
+ public boolean isSatisified() throws Exception {
+ return interruptCounter.get() > 0;
+ }
+ }));
+
+ proxy.goOn();
+ for (int i=0; i<messageCount; i++) {
+ assertNotNull("Got message " + i + " after reconnect", consumer.receive(5000));
+ }
+
+ assertTrue("no pending messages when done", Wait.waitFor(new Wait.Condition() {
+ @Override
+ public boolean isSatisified() throws Exception {
+
+ LOG.info("current total message count: " + broker.getAdminView().getTotalMessageCount());
+ return broker.getAdminView().getTotalMessageCount() == 0;
+ }
+ }));
+ }
+
+ private String initMessagePrefix(int i) {
+ byte[] content = new byte[i];
+ return new String(content);
+ }
+}
\ No newline at end of file
[3/3] activemq git commit: fix intermittent failure of
FailoverStaticNetworkTest. Bridge fails to start
triggerStartAsyncNetworkBridgeCreation thread waiting on localBrokerInfo. The
command was dropped due to contention between dispatch and peer start. F
Posted by gt...@apache.org.
fix intermittent failure of FailoverStaticNetworkTest. Bridge fails to start triggerStartAsyncNetworkBridgeCreation thread waiting on localBrokerInfo. The command was dropped due to contention between dispatch and peer start. Fix and test. Relates to https://issues.apache.org/jira/browse/AMQ-3684
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/c89bb7a3
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/c89bb7a3
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/c89bb7a3
Branch: refs/heads/master
Commit: c89bb7a316e3bc499e33d4dd5b14349f82ae1d1f
Parents: 3ef5389
Author: gtully <ga...@gmail.com>
Authored: Tue May 12 22:10:26 2015 +0100
Committer: gtully <ga...@gmail.com>
Committed: Tue May 12 22:10:57 2015 +0100
----------------------------------------------------------------------
.../activemq/transport/vm/VMTransport.java | 22 +++++----
activemq-unit-tests/pom.xml | 1 -
.../transport/vm/VMTransportBrokerNameTest.java | 48 ++++++++++++++++++++
3 files changed, 61 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/c89bb7a3/activemq-broker/src/main/java/org/apache/activemq/transport/vm/VMTransport.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/transport/vm/VMTransport.java b/activemq-broker/src/main/java/org/apache/activemq/transport/vm/VMTransport.java
index ec1a423..e7aa729 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/transport/vm/VMTransport.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/transport/vm/VMTransport.java
@@ -92,12 +92,24 @@ public class VMTransport implements Transport, Task {
throw new TransportDisposedIOException("Peer (" + peer.toString() + ") disposed.");
}
- if (peer.async || !peer.started.get()) {
+ if (peer.async) {
peer.getMessageQueue().put(command);
peer.wakeup();
return;
}
+ if (!peer.started.get()) {
+ LinkedBlockingQueue<Object> pending = peer.getMessageQueue();
+ boolean accepted = false;
+ do {
+ synchronized (peer.started) {
+ accepted = pending.offer(command);
+ }
+ } while (!accepted && !peer.started.get());
+ if (accepted) {
+ return;
+ }
+ }
} catch (InterruptedException e) {
InterruptedIOException iioe = new InterruptedIOException(e.getMessage());
iioe.initCause(e);
@@ -259,14 +271,6 @@ public class VMTransport implements Transport, Task {
this.transportListener = commandListener;
}
- public void setMessageQueue(LinkedBlockingQueue<Object> asyncQueue) {
- synchronized (this) {
- if (messageQueue == null) {
- messageQueue = asyncQueue;
- }
- }
- }
-
public LinkedBlockingQueue<Object> getMessageQueue() throws TransportDisposedIOException {
LinkedBlockingQueue<Object> result = messageQueue;
if (result == null) {
http://git-wip-us.apache.org/repos/asf/activemq/blob/c89bb7a3/activemq-unit-tests/pom.xml
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/pom.xml b/activemq-unit-tests/pom.xml
index 3ca9b1a..7ea4ce3 100755
--- a/activemq-unit-tests/pom.xml
+++ b/activemq-unit-tests/pom.xml
@@ -583,7 +583,6 @@
<exclude>**/amq1490/*</exclude>
<exclude>**/archive/*</exclude>
<exclude>**/NetworkFailoverTest.*/**</exclude>
- <exclude>**/vm/VMTransportBrokerTest.*</exclude>
<exclude>**/broker/MarshallingBrokerTest.*</exclude>
<exclude>**/AMQDeadlockTest3.*</exclude>
<!-- https://issues.apache.org/activemq/browse/AMQ-2050 -->
http://git-wip-us.apache.org/repos/asf/activemq/blob/c89bb7a3/activemq-unit-tests/src/test/java/org/apache/activemq/transport/vm/VMTransportBrokerNameTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/transport/vm/VMTransportBrokerNameTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/transport/vm/VMTransportBrokerNameTest.java
index ac05393..5183db8 100644
--- a/activemq-unit-tests/src/test/java/org/apache/activemq/transport/vm/VMTransportBrokerNameTest.java
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/transport/vm/VMTransportBrokerNameTest.java
@@ -16,7 +16,10 @@
*/
package org.apache.activemq.transport.vm;
+import java.io.IOException;
import java.net.URI;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import javax.jms.Connection;
@@ -25,6 +28,10 @@ import junit.framework.TestCase;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerRegistry;
+import org.apache.activemq.command.BrokerInfo;
+import org.apache.activemq.transport.Transport;
+import org.apache.activemq.transport.TransportFactory;
+import org.apache.activemq.transport.TransportListener;
public class VMTransportBrokerNameTest extends TestCase {
@@ -47,4 +54,45 @@ public class VMTransportBrokerNameTest extends TestCase {
c1.close();
c2.close();
}
+
+ public void testBrokerInfoClientAsync() throws Exception {
+
+ ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(new URI(vmUrl));
+ ActiveMQConnection c1 = (ActiveMQConnection) cf.createConnection();
+ assertTrue("Transport has name in it: " + c1.getTransport(), c1.getTransport().toString().contains(MY_BROKER));
+
+ for (int i=0;i<20; i++) {
+ final CountDownLatch gotBrokerInfo = new CountDownLatch(1);
+ Transport transport = TransportFactory.connect(new URI("vm://" + MY_BROKER + "?async=false"));
+ transport.setTransportListener(new TransportListener() {
+ @Override
+ public void onCommand(Object command) {
+ if (command instanceof BrokerInfo) {
+ gotBrokerInfo.countDown();
+ }
+ }
+
+ @Override
+ public void onException(IOException error) {
+
+ }
+
+ @Override
+ public void transportInterupted() {
+
+ }
+
+ @Override
+ public void transportResumed() {
+
+ }
+ });
+ transport.start();
+
+ assertTrue("got broker info on iteration:" + i, gotBrokerInfo.await(5, TimeUnit.SECONDS));
+
+ transport.stop();
+ }
+ c1.close();
+ }
}