3:35 p.m.
Author: clebert.suconic(a)jboss.com
Date: 2011-12-02 16:35:46 -0500 (Fri, 02 Dec 2011)
New Revision: 11820
Modified:
branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java
branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java
branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java
branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/DelegatingSession.java
branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java
branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/util/InVMNodeManager.java
Log:
https://issues.jboss.org/browse/JBPAPP-6543, https://issues.jboss.org/browse/HORNETQ-685
(Fix provided by Andy Taylor.. I'm just doing the commit after a review)
Modified:
branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java
===================================================================
---
branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2011-12-02
21:23:10 UTC (rev 11819)
+++
branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2011-12-02
21:35:46 UTC (rev 11820)
@@ -921,6 +921,17 @@
*/
private void reconnectSessions(final CoreRemotingConnection oldConnection, final int
reconnectAttempts)
{
+ HashSet<ClientSessionInternal> sessionsToFailover;
+ synchronized (sessions)
+ {
+ sessionsToFailover = new HashSet<ClientSessionInternal>(sessions);
+ }
+
+ for (ClientSessionInternal session : sessionsToFailover)
+ {
+ session.preHandleFailover(connection);
+ }
+
getConnectionWithRetry(reconnectAttempts);
if (connection == null)
@@ -946,12 +957,6 @@
connection.setFailureListeners(newListeners);
- HashSet<ClientSessionInternal> sessionsToFailover;
- synchronized (sessions)
- {
- sessionsToFailover = new HashSet<ClientSessionInternal>(sessions);
- }
-
for (ClientSessionInternal session : sessionsToFailover)
{
session.handleFailover(connection);
@@ -968,7 +973,7 @@
" multiplier = " +
retryIntervalMultiplier, new
Exception("trace"));
}
-
+
long interval = retryInterval;
int count = 0;
@@ -1042,6 +1047,10 @@
}
else
{
+ if (log.isDebugEnabled())
+ {
+ log.debug("Reconnection successfull");
+ }
return;
}
}
Modified:
branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java
===================================================================
---
branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2011-12-02
21:23:10 UTC (rev 11819)
+++
branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2011-12-02
21:35:46 UTC (rev 11820)
@@ -757,7 +757,10 @@
}
checkClosed();
-
+ if (log.isDebugEnabled())
+ {
+ log.debug("client ack messageID = " + messageID);
+ }
SessionAcknowledgeMessage message = new SessionAcknowledgeMessage(consumerID,
messageID, blockOnAcknowledge);
if (blockOnAcknowledge)
@@ -935,6 +938,14 @@
sendAckHandler = handler;
}
+ public void preHandleFailover(CoreRemotingConnection connection)
+ {
+ // We lock the channel to prevent any packets to be added to the resend
+ // cache during the failover process
+ //we also do this before the connection fails over to give the session a chance to
block for failover
+ channel.lock();
+ }
+
// Needs to be synchronized to prevent issues with occurring concurrently with
close()
public void handleFailover(final CoreRemotingConnection backupConnection)
@@ -948,9 +959,6 @@
boolean resetCreditManager = false;
- // We lock the channel to prevent any packets to be added to the resend
- // cache during the failover process
- channel.lock();
try
{
channel.transferConnection(backupConnection);
Modified:
branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java
===================================================================
---
branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java 2011-12-02
21:23:10 UTC (rev 11819)
+++
branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java 2011-12-02
21:35:46 UTC (rev 11820)
@@ -59,6 +59,8 @@
void handleReceiveContinuation(long consumerID, SessionReceiveContinuationMessage
continuation) throws Exception;
+ void preHandleFailover(CoreRemotingConnection connection);
+
void handleFailover(CoreRemotingConnection backupConnection);
RemotingConnection getConnection();
@@ -92,5 +94,4 @@
void setPacketSize(int packetSize);
void resetIfNeeded() throws HornetQException;
-
}
Modified:
branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/DelegatingSession.java
===================================================================
---
branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/DelegatingSession.java 2011-12-02
21:23:10 UTC (rev 11819)
+++
branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/DelegatingSession.java 2011-12-02
21:35:46 UTC (rev 11820)
@@ -373,6 +373,11 @@
return session.getXAResource();
}
+ public void preHandleFailover(CoreRemotingConnection connection)
+ {
+ session.preHandleFailover(connection);
+ }
+
public void handleFailover(final CoreRemotingConnection backupConnection)
{
session.handleFailover(backupConnection);
Modified:
branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java
===================================================================
---
branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java 2011-12-02
21:23:10 UTC (rev 11819)
+++
branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java 2011-12-02
21:35:46 UTC (rev 11820)
@@ -44,6 +44,7 @@
import org.hornetq.core.logging.Logger;
import org.hornetq.core.transaction.impl.XidImpl;
import org.hornetq.jms.client.HornetQTextMessage;
+import org.hornetq.tests.integration.cluster.util.InVMNodeManager;
import org.hornetq.tests.integration.cluster.util.TestableServer;
import org.hornetq.tests.util.RandomUtil;
@@ -129,6 +130,246 @@
return sf.createSession(xa, autoCommitSends, autoCommitAcks);
}
+ // https://issues.jboss.org/browse/HORNETQ-685
+ public void testTimeoutOnFailover() throws Exception
+ {
+ locator.setCallTimeout(5000);
+ locator.setBlockOnNonDurableSend(true);
+ locator.setBlockOnDurableSend(true);
+ locator.setAckBatchSize(0);
+ locator.setReconnectAttempts(-1);
+ ((InVMNodeManager)nodeManager).failoverPause = 5000l;
+
+ ClientSessionFactoryInternal sf =
(ClientSessionFactoryInternal)locator.createSessionFactory();
+
+ final ClientSession session = createSession(sf, true, true);
+
+ session.createQueue(FailoverTestBase.ADDRESS, FailoverTestBase.ADDRESS, null,
true);
+
+ final ClientProducer producer = session.createProducer(FailoverTestBase.ADDRESS);
+
+ final CountDownLatch latch = new CountDownLatch(10);
+
+ Runnable r = new Runnable()
+ {
+ public void run()
+ {
+ for (int i = 0; i < 500; i++)
+ {
+ ClientMessage message = session.createMessage(true);
+ message.putIntProperty("counter", i);
+ try
+ {
+ System.out.println("sending message: " + i);
+ producer.send(message);
+ if (i < 10)
+ {
+ latch.countDown();
+ }
+ }
+ catch (HornetQException e)
+ {
+ // this is our retry
+ try
+ {
+ producer.send(message);
+ }
+ catch (HornetQException e1)
+ {
+ e1.printStackTrace();
+ }
+ }
+ }
+ }
+ };
+ Thread t = new Thread(r);
+ t.start();
+ latch.await(10, TimeUnit.SECONDS);
+ log.info("crashing session");
+ crash(session);
+ t.join(5000);
+ ClientConsumer consumer = session.createConsumer(FailoverTestBase.ADDRESS);
+ session.start();
+ for (int i = 0; i < 500; i++)
+ {
+ ClientMessage m = consumer.receive(1000);
+ assertNotNull(m);
+ System.out.println("received message " + i);
+ // assertEquals(i, m.getIntProperty("counter").intValue());
+ }
+ }
+
+ // https://issues.jboss.org/browse/HORNETQ-685
+ public void testTimeoutOnFailoverConsume() throws Exception
+ {
+ locator.setCallTimeout(5000);
+ locator.setBlockOnNonDurableSend(true);
+ locator.setBlockOnDurableSend(true);
+ locator.setAckBatchSize(0);
+ locator.setBlockOnAcknowledge(true);
+ locator.setReconnectAttempts(-1);
+ locator.setRetryInterval(500);
+ locator.setAckBatchSize(0);
+ ((InVMNodeManager)nodeManager).failoverPause = 5000l;
+
+ ClientSessionFactoryInternal sf =
(ClientSessionFactoryInternal)locator.createSessionFactory();
+
+ final ClientSession session = createSession(sf, true, true);
+
+ session.createQueue(FailoverTestBase.ADDRESS, FailoverTestBase.ADDRESS, null,
true);
+
+ final ClientProducer producer = session.createProducer(FailoverTestBase.ADDRESS);
+
+ for (int i = 0; i < 500; i++)
+ {
+ ClientMessage message = session.createMessage(true);
+ message.putIntProperty("counter", i);
+ producer.send(message);
+ }
+
+ final CountDownLatch latch = new CountDownLatch(1);
+ final CountDownLatch endLatch = new CountDownLatch(1);
+
+ final ClientConsumer consumer = session.createConsumer(FailoverTestBase.ADDRESS);
+ session.start();
+
+ final Map<Integer, ClientMessage> received = new HashMap<Integer,
ClientMessage>();
+
+ consumer.setMessageHandler(new MessageHandler()
+ {
+
+ public void onMessage(ClientMessage message)
+ {
+ Integer counter = message.getIntProperty("counter");
+ received.put(counter, message);
+ try
+ {
+ log.info("acking message = id = " + message.getMessageID() +
+ ", counter = " +
+ message.getIntProperty("counter"));
+ message.acknowledge();
+ }
+ catch (HornetQException e)
+ {
+ e.printStackTrace();
+ return;
+ }
+ log.info("Acked counter = " + counter);
+ if (counter.equals(10))
+ {
+ latch.countDown();
+ }
+ if (received.size() == 500)
+ {
+ endLatch.countDown();
+ }
+ }
+
+ });
+ latch.await(10, TimeUnit.SECONDS);
+ log.info("crashing session");
+ crash(session);
+ endLatch.await(60, TimeUnit.SECONDS);
+ assertTrue("received only " + received.size(), received.size() == 500);
+
+ session.close();
+ }
+
+ // https://issues.jboss.org/browse/HORNETQ-685
+ public void testTimeoutOnFailoverTransactionCommit() throws Exception
+ {
+ locator.setCallTimeout(2000);
+ locator.setBlockOnNonDurableSend(true);
+ locator.setBlockOnDurableSend(true);
+ locator.setAckBatchSize(0);
+ locator.setReconnectAttempts(-1);
+ ((InVMNodeManager)nodeManager).failoverPause = 5000l;
+
+ ClientSessionFactoryInternal sf =
(ClientSessionFactoryInternal)locator.createSessionFactory();
+
+ final ClientSession session = createSession(sf, true, false, false);
+
+ session.createQueue(FailoverTestBase.ADDRESS, FailoverTestBase.ADDRESS, null,
true);
+
+ final ClientProducer producer = session.createProducer(FailoverTestBase.ADDRESS);
+
+ Xid xid = new XidImpl("uhuhuhu".getBytes(), 126512,
"auhsduashd".getBytes());
+
+ session.start(xid, XAResource.TMNOFLAGS);
+
+ for (int i = 0; i < 500; i++)
+ {
+ ClientMessage message = session.createMessage(true);
+ message.putIntProperty("counter", i);
+
+ System.out.println("sending message: " + i);
+ producer.send(message);
+
+ }
+ session.end(xid, XAResource.TMSUCCESS);
+ session.prepare(xid);
+ System.out.println("crashing session");
+ crash(false, session);
+
+ session.commit(xid, false);
+
+ ClientConsumer consumer = session.createConsumer(FailoverTestBase.ADDRESS);
+ session.start();
+ for (int i = 0; i < 500; i++)
+ {
+ ClientMessage m = consumer.receive(1000);
+ assertNotNull(m);
+ System.out.println("received message " + i);
+ assertEquals(i, m.getIntProperty("counter").intValue());
+ }
+ }
+
+ // https://issues.jboss.org/browse/HORNETQ-685
+ public void testTimeoutOnFailoverTransactionRollback() throws Exception
+ {
+ locator.setCallTimeout(2000);
+ locator.setBlockOnNonDurableSend(true);
+ locator.setBlockOnDurableSend(true);
+ locator.setAckBatchSize(0);
+ locator.setReconnectAttempts(-1);
+ ((InVMNodeManager)nodeManager).failoverPause = 5000l;
+
+ ClientSessionFactoryInternal sf =
(ClientSessionFactoryInternal)locator.createSessionFactory();
+
+ final ClientSession session = createSession(sf, true, false, false);
+
+ session.createQueue(FailoverTestBase.ADDRESS, FailoverTestBase.ADDRESS, null,
true);
+
+ final ClientProducer producer = session.createProducer(FailoverTestBase.ADDRESS);
+
+ Xid xid = new XidImpl("uhuhuhu".getBytes(), 126512,
"auhsduashd".getBytes());
+
+ session.start(xid, XAResource.TMNOFLAGS);
+
+ for (int i = 0; i < 500; i++)
+ {
+ ClientMessage message = session.createMessage(true);
+ message.putIntProperty("counter", i);
+
+ System.out.println("sending message: " + i);
+ producer.send(message);
+ }
+
+ session.end(xid, XAResource.TMSUCCESS);
+ session.prepare(xid);
+ System.out.println("crashing session");
+ crash(false, session);
+
+ session.rollback(xid);
+
+ ClientConsumer consumer = session.createConsumer(FailoverTestBase.ADDRESS);
+ session.start();
+
+ ClientMessage m = consumer.receive(1000);
+ assertNull(m);
+
+ }
+
// https://jira.jboss.org/browse/HORNETQ-522
public void testNonTransactedWithZeroConsumerWindowSize() throws Exception
{
@@ -1334,7 +1575,7 @@
session2.end(xid, XAResource.TMSUCCESS);
- // session2.prepare(xid);
+ // session2.prepare(xid);
crash(session2);
Modified:
branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/util/InVMNodeManager.java
===================================================================
---
branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/util/InVMNodeManager.java 2011-12-02
21:23:10 UTC (rev 11819)
+++
branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/util/InVMNodeManager.java 2011-12-02
21:35:46 UTC (rev 11820)
@@ -37,6 +37,8 @@
public State state = NOT_STARTED;
+ public long failoverPause = 0l;
+
public InVMNodeManager()
{
liveLock = new Semaphore(1);
@@ -73,6 +75,10 @@
}
}
while (true);
+ if(failoverPause > 0l)
+ {
+ Thread.sleep(failoverPause);
+ }
}
@Override