JBoss hornetq SVN: r10326 - branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/jms/cluster.
by do-not-reply＠jboss.org
Author: clebert.suconic(a)jboss.com
Date: 2011-03-14 18:06:16 -0400 (Mon, 14 Mar 2011)
New Revision: 10326
Modified:
branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/jms/cluster/JMSReconnectTest.java
Log:
https://issues.jboss.org/browse/HORNETQ-653
Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/jms/cluster/JMSReconnectTest.java
===================================================================
--- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/jms/cluster/JMSReconnectTest.java 2011-03-14 16:46:25 UTC (rev 10325)
+++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/jms/cluster/JMSReconnectTest.java 2011-03-14 22:06:16 UTC (rev 10326)
@@ -168,9 +168,11 @@
conn.close();
- Assert.assertNotNull(listener.e);
- Assert.assertTrue(me == listener.e.getCause());
+// TODO - https://issues.jboss.org/browse/HORNETQ-653
+// Assert.assertNotNull(listener.e);
+//
+// Assert.assertTrue(me == listener.e.getCause());
}
public void testReconnectSameNodeServerRestartedWithNonDurableSub() throws Exception
@@ -255,7 +257,8 @@
conn.close();
- Assert.assertNotNull(listener.e);
+ // TODO - https://issues.jboss.org/browse/HORNETQ-653
+ //Assert.assertNotNull(listener.e);
}
//If the server is shutdown after a non durable sub is created, then close on the connection should proceed normally
13 years, 2 months
JBoss hornetq SVN: r10325 - in branches/Branch_2_2_EAP: src/main/org/hornetq/core/server/impl and 1 other directories.
by do-not-reply＠jboss.org
Author: clebert.suconic(a)jboss.com
Date: 2011-03-14 12:46:25 -0400 (Mon, 14 Mar 2011)
New Revision: 10325
Modified:
branches/Branch_2_2_EAP/src/main/org/hornetq/core/management/impl/QueueControlImpl.java
branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/QueueImpl.java
branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/management/QueueControlTest.java
Log:
https://issues.jboss.org/browse/JBPAPP-6091 - move message is not ignoring duplicates (feature was only working with moveMessages)
Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/management/impl/QueueControlImpl.java
===================================================================
--- branches/Branch_2_2_EAP/src/main/org/hornetq/core/management/impl/QueueControlImpl.java 2011-03-14 15:46:46 UTC (rev 10324)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/management/impl/QueueControlImpl.java 2011-03-14 16:46:25 UTC (rev 10325)
@@ -575,7 +575,7 @@
throw new IllegalArgumentException("No queue found for " + otherQueueName);
}
- return queue.moveReference(messageID, binding.getAddress());
+ return queue.moveReference(messageID, binding.getAddress(), rejectDuplicates);
}
finally
{
Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/QueueImpl.java
===================================================================
--- branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/QueueImpl.java 2011-03-14 15:46:46 UTC (rev 10324)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/QueueImpl.java 2011-03-14 16:46:25 UTC (rev 10325)
@@ -826,7 +826,7 @@
{
if (expiryAddress != null)
{
- move(expiryAddress, ref, true);
+ move(expiryAddress, ref, true, false);
}
else
{
@@ -1101,7 +1101,7 @@
deliveringCount.incrementAndGet();
try
{
- move(toAddress, ref);
+ move(toAddress, ref, false, rejectDuplicate);
}
catch (Exception e)
{
@@ -1627,11 +1627,6 @@
return messageReferences.size();
}
- private void move(final SimpleString toAddress, final MessageReference ref) throws Exception
- {
- move(toAddress, ref, false);
- }
-
private void move(final SimpleString toAddress,
final Transaction tx,
final MessageReference ref,
@@ -1711,7 +1706,7 @@
QueueImpl.log.warn("Message has reached maximum delivery attempts, sending it to Dead Letter Address " + deadLetterAddress +
" from " +
name);
- move(deadLetterAddress, ref, false);
+ move(deadLetterAddress, ref, false, false);
}
}
else
@@ -1723,7 +1718,7 @@
}
}
- private void move(final SimpleString address, final MessageReference ref, final boolean expiry) throws Exception
+ private void move(final SimpleString address, final MessageReference ref, final boolean expiry, final boolean rejectDuplicate) throws Exception
{
Transaction tx = new TransactionImpl(storageManager);
@@ -1731,7 +1726,7 @@
copyMessage.setAddress(address);
- postOffice.route(copyMessage, tx, false);
+ postOffice.route(copyMessage, tx, false, rejectDuplicate);
acknowledge(tx, ref);
Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/management/QueueControlTest.java
===================================================================
--- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/management/QueueControlTest.java 2011-03-14 15:46:46 UTC (rev 10324)
+++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/management/QueueControlTest.java 2011-03-14 16:46:25 UTC (rev 10325)
@@ -1345,6 +1345,92 @@
}
+ public void testMoveMessagesBack2() throws Exception
+ {
+ server.createQueue(new SimpleString("q1"), new SimpleString("q1"), null, true, false);
+ server.createQueue(new SimpleString("q2"), new SimpleString("q2"), null, true, false);
+
+ ServerLocator locator = createInVMNonHALocator();
+
+ ClientSessionFactory sf = locator.createSessionFactory();
+
+ ClientSession session = sf.createSession(true, true);
+
+ ClientProducer prod1 = session.createProducer("q1");
+
+ int NUMBER_OF_MSGS = 10;
+
+ for (int i = 0; i < NUMBER_OF_MSGS; i++)
+ {
+ ClientMessage msg = session.createMessage(true);
+
+ msg.putStringProperty(org.hornetq.api.core.Message.HDR_DUPLICATE_DETECTION_ID, new SimpleString("dupl-" + i));
+
+ prod1.send(msg);
+ }
+
+ session.commit();
+
+ ClientConsumer consumer = session.createConsumer("q1", true);
+ session.start();
+
+ assertNotNull(consumer.receive(5000));
+ consumer.close();
+
+ QueueControl q1Control = ManagementControlHelper.createQueueControl(new SimpleString("q1"),
+ new SimpleString("q1"),
+ mbeanServer);
+
+ QueueControl q2Control = ManagementControlHelper.createQueueControl(new SimpleString("q2"),
+ new SimpleString("q2"),
+ mbeanServer);
+
+ assertEquals(NUMBER_OF_MSGS, q1Control.moveMessages(null, "q2"));
+
+ long messageIDs[] = new long[NUMBER_OF_MSGS];
+
+ consumer = session.createConsumer("q2", true);
+
+ for (int i = 0 ; i < NUMBER_OF_MSGS; i++)
+ {
+ ClientMessage msg = consumer.receive(5000);
+ assertNotNull(msg);
+ messageIDs[i] = msg.getMessageID();
+ }
+
+ assertNull(consumer.receiveImmediate());
+
+ consumer.close();
+
+ for (int i = 0 ; i < NUMBER_OF_MSGS; i++)
+ {
+ q2Control.moveMessage(messageIDs[i], "q1");
+ }
+
+
+ session.start();
+ consumer = session.createConsumer("q1");
+
+ for (int i = 0; i < NUMBER_OF_MSGS; i++)
+ {
+ ClientMessage msg = consumer.receive(5000);
+ System.out.println("msg = " + msg);
+ assertNotNull(msg);
+ msg.acknowledge();
+ }
+
+ consumer.close();
+
+ session.deleteQueue("q1");
+
+ session.deleteQueue("q2");
+
+ session.close();
+
+ locator.close();
+
+ }
+
public void testPauseAndResume()
{
long counterPeriod = 1000;
13 years, 2 months
JBoss hornetq SVN: r10324 - in branches/Branch_2_2_EAP/src/main/org/hornetq: core/server/impl and 3 other directories.
by do-not-reply＠jboss.org
Author: ataylor
Date: 2011-03-14 11:46:46 -0400 (Mon, 14 Mar 2011)
New Revision: 10324
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/ClientSessionInternal.java
branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/DelegatingSession.java
branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java
branches/Branch_2_2_EAP/src/main/org/hornetq/jms/client/HornetQConnection.java
branches/Branch_2_2_EAP/src/main/org/hornetq/ra/HornetQRAXAResource.java
branches/Branch_2_2_EAP/src/main/org/hornetq/ra/inflow/HornetQActivation.java
branches/Branch_2_2_EAP/src/main/org/hornetq/ra/inflow/HornetQMessageHandler.java
Log:
https://issues.jboss.org/browse/JBPAPP-6049 - second part of fix to handle exceptions during xa calls and to handle xa retry properly
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-03-13 02:22:03 UTC (rev 10323)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2011-03-14 15:46:46 UTC (rev 10324)
@@ -190,6 +190,8 @@
private volatile SimpleString defaultAddress;
+ private boolean xaRetry = false;
+
// Constructors ----------------------------------------------------------------------------
public ClientSessionImpl(final ClientSessionFactoryInternal sessionFactory,
@@ -631,6 +633,15 @@
return xa;
}
+ public void resetIfNeeded() throws HornetQException
+ {
+ if(rollbackOnly)
+ {
+ log.warn("resetting session after failure");
+ rollback(false);
+ }
+ }
+
public void start() throws HornetQException
{
checkClosed();
@@ -1193,9 +1204,10 @@
{
checkXA();
+ //we should never throw rollback if we have already prepared
if (rollbackOnly)
{
- throw new XAException(XAException.XA_RBOTHER);
+ log.warn("committing transaction after failover occurred, any non persistent messages may be lost");
}
// Note - don't need to flush acks since the previous end would have
@@ -1211,29 +1223,27 @@
if (response.isError())
{
+ //if we retry and its not there the assume that it was committed
+ if(xaRetry && response.getResponseCode() == XAException.XAER_NOTA)
+ {
+ return;
+ }
throw new XAException(response.getResponseCode());
}
}
catch (HornetQException e)
{
- ClientSessionImpl.log.warn(e.getMessage(), e);
+ ClientSessionImpl.log.warn("failover occured during commit throwing XAException.XA_RETRY");
if (e.getCode() == HornetQException.UNBLOCKED)
{
// Unblocked on failover
+ xaRetry = true;
+ throw new XAException(XAException.XA_RETRY);
+ }
- try
- {
- rollback(false);
- }
- catch (HornetQException e2)
- {
- throw new XAException(XAException.XAER_RMERR);
- }
+ ClientSessionImpl.log.warn(e.getMessage(), e);
- throw new XAException(XAException.XA_RBOTHER);
- }
-
// This should never occur
throw new XAException(XAException.XAER_RMERR);
}
@@ -1365,17 +1375,35 @@
}
else
{
+ xaRetry = false;
return response.getResponseCode();
}
}
catch (HornetQException e)
{
- ClientSessionImpl.log.warn(e.getMessage(), e);
-
if (e.getCode() == HornetQException.UNBLOCKED)
{
// Unblocked on failover
+ try
+ {
+ log.warn("failover occurred during prepare re-trying");
+ SessionXAResponseMessage response = (SessionXAResponseMessage)channel.sendBlocking(packet);
+ if (response.isError())
+ {
+ throw new XAException(response.getResponseCode());
+ }
+ else
+ {
+ xaRetry = false;
+ return response.getResponseCode();
+ }
+ }
+ catch (HornetQException e1)
+ {
+ //ignore and rollback
+ }
+ log.warn("failover occurred during prepare rolling back");
try
{
rollback(false);
@@ -1385,9 +1413,13 @@
throw new XAException(XAException.XAER_RMERR);
}
+ ClientSessionImpl.log.warn(e.getMessage(), e);
+
throw new XAException(XAException.XA_RBOTHER);
}
+ ClientSessionImpl.log.warn(e.getMessage(), e);
+
// This should never occur
throw new XAException(XAException.XAER_RMERR);
}
@@ -1455,11 +1487,22 @@
if (response.isError())
{
+ //if we retry and its not there the assume that it was rolled back
+ if(xaRetry && response.getResponseCode() == XAException.XAER_NOTA)
+ {
+ return;
+ }
throw new XAException(response.getResponseCode());
}
}
catch (HornetQException e)
{
+ if (e.getCode() == HornetQException.UNBLOCKED)
+ {
+ // Unblocked on failover
+ xaRetry = true;
+ throw new XAException(XAException.XA_RETRY);
+ }
// This should never occur
throw new XAException(XAException.XAER_RMERR);
}
@@ -1485,10 +1528,11 @@
public void start(final Xid xid, final int flags) throws XAException
{
checkXA();
+
+ Packet packet = null;
+
try
{
- Packet packet;
-
if (flags == XAResource.TMJOIN)
{
packet = new SessionXAJoinMessage(xid);
@@ -1519,6 +1563,27 @@
}
catch (HornetQException e)
{
+ //we can retry this only because we know for sure that no work would have been done
+ if (e.getCode() == HornetQException.UNBLOCKED)
+ {
+ try
+ {
+ SessionXAResponseMessage response = (SessionXAResponseMessage)channel.sendBlocking(packet);
+
+ if (response.isError())
+ {
+ ClientSessionImpl.log.error("XA operation failed " + response.getMessage() +
+ " code:" +
+ response.getResponseCode());
+ throw new XAException(response.getResponseCode());
+ }
+ }
+ catch (HornetQException e1)
+ {
+ // This should never occur
+ throw new XAException(XAException.XAER_RMERR);
+ }
+ }
// This should never occur
throw new XAException(XAException.XAER_RMERR);
}
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-03-13 02:22:03 UTC (rev 10323)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java 2011-03-14 15:46:46 UTC (rev 10324)
@@ -87,4 +87,7 @@
void setAddress(Message message, SimpleString address);
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-03-13 02:22:03 UTC (rev 10323)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/DelegatingSession.java 2011-03-14 15:46:46 UTC (rev 10324)
@@ -492,6 +492,11 @@
return session.setTransactionTimeout(seconds);
}
+ public void resetIfNeeded() throws HornetQException
+ {
+ session.resetIfNeeded();
+ }
+
public void start() throws HornetQException
{
session.start();
Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java
===================================================================
--- branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2011-03-13 02:22:03 UTC (rev 10323)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2011-03-14 15:46:46 UTC (rev 10324)
@@ -579,7 +579,14 @@
doRollback(considerLastMessageAsDelivered, tx);
- tx = new TransactionImpl(storageManager, timeoutSeconds);
+ if (xa)
+ {
+ tx = null;
+ }
+ else
+ {
+ tx = new TransactionImpl(storageManager, timeoutSeconds);
+ }
}
public void xaCommit(final Xid xid, final boolean onePhase) throws Exception
@@ -878,6 +885,10 @@
throw new HornetQXAException(XAException.XAER_PROTO,
"Cannot prepare transaction, it is suspended " + xid);
}
+ else if(theTx.getState() == Transaction.State.PREPARED)
+ {
+ log.info("ignoring prepare on xid as already called :" + xid);
+ }
else
{
theTx.prepare();
Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/jms/client/HornetQConnection.java
===================================================================
--- branches/Branch_2_2_EAP/src/main/org/hornetq/jms/client/HornetQConnection.java 2011-03-13 02:22:03 UTC (rev 10323)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/jms/client/HornetQConnection.java 2011-03-14 15:46:46 UTC (rev 10324)
@@ -628,7 +628,7 @@
HornetQConnection conn = connectionRef.get();
- if (conn != null)
+ if (conn != null && ! failedOver)
{
try
{
Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/ra/HornetQRAXAResource.java
===================================================================
--- branches/Branch_2_2_EAP/src/main/org/hornetq/ra/HornetQRAXAResource.java 2011-03-13 02:22:03 UTC (rev 10323)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/ra/HornetQRAXAResource.java 2011-03-14 15:46:46 UTC (rev 10324)
@@ -17,6 +17,8 @@
import javax.transaction.xa.XAResource;
import javax.transaction.xa.Xid;
+import org.hornetq.api.core.HornetQException;
+import org.hornetq.core.client.impl.ClientSessionInternal;
import org.hornetq.core.logging.Logger;
/**
@@ -70,8 +72,19 @@
}
managedConnection.lock();
+
+ ClientSessionInternal sessionInternal = (ClientSessionInternal) xaResource;
try
{
+ //this resets any tx stuff, we assume here that the tm and jca layer are well behaved when it comes to this
+ sessionInternal.resetIfNeeded();
+ }
+ catch (HornetQException e)
+ {
+ log.warn("problem resetting HornetQ xa session after failure");
+ }
+ try
+ {
xaResource.start(xid, flags);
}
finally
Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/ra/inflow/HornetQActivation.java
===================================================================
--- branches/Branch_2_2_EAP/src/main/org/hornetq/ra/inflow/HornetQActivation.java 2011-03-13 02:22:03 UTC (rev 10323)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/ra/inflow/HornetQActivation.java 2011-03-14 15:46:46 UTC (rev 10324)
@@ -33,6 +33,7 @@
import org.hornetq.api.core.SimpleString;
import org.hornetq.api.core.client.ClientSession;
import org.hornetq.api.jms.HornetQJMSClient;
+import org.hornetq.core.client.impl.ClientSessionInternal;
import org.hornetq.core.logging.Logger;
import org.hornetq.jms.client.HornetQConnectionFactory;
import org.hornetq.jms.client.HornetQDestination;
@@ -287,7 +288,7 @@
try
{
session = setupSession();
- HornetQMessageHandler handler = new HornetQMessageHandler(this, ra.getTM(), session, i);
+ HornetQMessageHandler handler = new HornetQMessageHandler(this, ra.getTM(), (ClientSessionInternal) session, i);
handler.setup();
session.start();
handlers.add(handler);
Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/ra/inflow/HornetQMessageHandler.java
===================================================================
--- branches/Branch_2_2_EAP/src/main/org/hornetq/ra/inflow/HornetQMessageHandler.java 2011-03-13 02:22:03 UTC (rev 10323)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/ra/inflow/HornetQMessageHandler.java 2011-03-14 15:46:46 UTC (rev 10324)
@@ -28,6 +28,7 @@
import org.hornetq.api.core.client.ClientSession;
import org.hornetq.api.core.client.MessageHandler;
import org.hornetq.api.core.client.ClientSession.QueueQuery;
+import org.hornetq.core.client.impl.ClientSessionInternal;
import org.hornetq.core.logging.Logger;
import org.hornetq.jms.client.HornetQDestination;
import org.hornetq.jms.client.HornetQMessage;
@@ -55,7 +56,7 @@
/**
* The session
*/
- private final ClientSession session;
+ private final ClientSessionInternal session;
private ClientConsumer consumer;
@@ -76,7 +77,7 @@
public HornetQMessageHandler(final HornetQActivation activation,
final TransactionManager tm,
- final ClientSession session,
+ final ClientSessionInternal session,
final int sessionNr)
{
this.activation = activation;
@@ -318,6 +319,17 @@
}
}
}
+ finally
+ {
+ try
+ {
+ session.resetIfNeeded();
+ }
+ catch (HornetQException e)
+ {
+ log.warn("unable to reset session after failure");
+ }
+ }
}
13 years, 2 months
JBoss hornetq SVN: r10323 - branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl.
by do-not-reply＠jboss.org
Author: clebert.suconic(a)jboss.com
Date: 2011-03-12 21:22:03 -0500 (Sat, 12 Mar 2011)
New Revision: 10323
Modified:
branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java
Log:
Resending metadata on failover/reconnection
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-03-11 21:33:07 UTC (rev 10322)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2011-03-13 02:22:03 UTC (rev 10323)
@@ -13,6 +13,7 @@
package org.hornetq.core.client.impl;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
@@ -109,6 +110,8 @@
// Attributes ----------------------------------------------------------------------------
+ private Map<String, String> metadata = new HashMap<String, String>();
+
private final ClientSessionFactoryInternal sessionFactory;
private final String name;
@@ -155,7 +158,7 @@
private final boolean blockOnDurableSend;
private final int minLargeMessageSize;
-
+
private final boolean compressLargeMessages;
private volatile int initialMessagePacketSize;
@@ -184,7 +187,7 @@
private final String groupID;
private volatile boolean inClose;
-
+
private volatile SimpleString defaultAddress;
// Constructors ----------------------------------------------------------------------------
@@ -262,7 +265,7 @@
this.cacheLargeMessageClient = cacheLargeMessageClient;
this.minLargeMessageSize = minLargeMessageSize;
-
+
this.compressLargeMessages = compressLargeMessages;
this.initialMessagePacketSize = initialMessagePacketSize;
@@ -274,7 +277,7 @@
// ClientSession implementation
// -----------------------------------------------------------------
-
+
public void createQueue(final SimpleString address, final SimpleString queueName) throws HornetQException
{
internalCreateQueue(address, queueName, null, false, false);
@@ -649,7 +652,7 @@
{
stop(true);
}
-
+
public void stop(final boolean waitForOnMessage) throws HornetQException
{
checkClosed();
@@ -689,7 +692,7 @@
{
return minLargeMessageSize;
}
-
+
public boolean isCompressLargeMessages()
{
return compressLargeMessages;
@@ -875,197 +878,222 @@
// Needs to be synchronized to prevent issues with occurring concurrently with close()
- public synchronized void handleFailover(final CoreRemotingConnection backupConnection)
+ public void handleFailover(final CoreRemotingConnection backupConnection)
{
- if (closed)
+ synchronized (this)
{
- return;
- }
+ if (closed)
+ {
+ return;
+ }
- boolean resetCreditManager = false;
+ 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);
+ // 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);
- backupConnection.syncIDGeneratorSequence(remotingConnection.getIDGeneratorSequence());
+ backupConnection.syncIDGeneratorSequence(remotingConnection.getIDGeneratorSequence());
- remotingConnection = backupConnection;
+ remotingConnection = backupConnection;
- int lcid = channel.getLastConfirmedCommandID();
+ int lcid = channel.getLastConfirmedCommandID();
- Packet request = new ReattachSessionMessage(name, lcid);
+ Packet request = new ReattachSessionMessage(name, lcid);
- Channel channel1 = backupConnection.getChannel(1, -1);
+ Channel channel1 = backupConnection.getChannel(1, -1);
- ReattachSessionResponseMessage response = (ReattachSessionResponseMessage)channel1.sendBlocking(request);
+ ReattachSessionResponseMessage response = (ReattachSessionResponseMessage)channel1.sendBlocking(request);
- if (response.isReattached())
- {
- // The session was found on the server - we reattached transparently ok
+ if (response.isReattached())
+ {
+ // The session was found on the server - we reattached transparently ok
- channel.replayCommands(response.getLastConfirmedCommandID());
- }
- else
- {
+ channel.replayCommands(response.getLastConfirmedCommandID());
+ }
+ else
+ {
- // The session wasn't found on the server - probably we're failing over onto a backup server where the
- // session won't exist or the target server has been restarted - in this case the session will need to be
- // recreated,
- // and we'll need to recreate any consumers
+ // The session wasn't found on the server - probably we're failing over onto a backup server where the
+ // session won't exist or the target server has been restarted - in this case the session will need to be
+ // recreated,
+ // and we'll need to recreate any consumers
- // It could also be that the server hasn't been restarted, but the session is currently executing close, and
- // that
- // has already been executed on the server, that's why we can't find the session- in this case we *don't*
- // want
- // to recreate the session, we just want to unblock the blocking call
- if (!inClose)
- {
- Packet createRequest = new CreateSessionMessage(name,
- channel.getID(),
- version,
- username,
- password,
- minLargeMessageSize,
- xa,
- autoCommitSends,
- autoCommitAcks,
- preAcknowledge,
- confirmationWindowSize,
- defaultAddress == null ? null
- : defaultAddress.toString());
- boolean retry = false;
- do
+ // It could also be that the server hasn't been restarted, but the session is currently executing close,
+ // and
+ // that
+ // has already been executed on the server, that's why we can't find the session- in this case we *don't*
+ // want
+ // to recreate the session, we just want to unblock the blocking call
+ if (!inClose)
{
- try
+ Packet createRequest = new CreateSessionMessage(name,
+ channel.getID(),
+ version,
+ username,
+ password,
+ minLargeMessageSize,
+ xa,
+ autoCommitSends,
+ autoCommitAcks,
+ preAcknowledge,
+ confirmationWindowSize,
+ defaultAddress == null ? null
+ : defaultAddress.toString());
+ boolean retry = false;
+ do
{
- channel1.sendBlocking(createRequest);
- retry = false;
- }
- catch (HornetQException e)
- {
- // the session was created while its server was starting, retry it:
- if (e.getCode() == HornetQException.SESSION_CREATION_REJECTED)
+ try
{
- ClientSessionImpl.log.warn("Server is starting, retry to create the session " + name);
- retry = true;
- // sleep a little bit to avoid spinning too much
- Thread.sleep(10);
+ channel1.sendBlocking(createRequest);
+ retry = false;
}
- else
+ catch (HornetQException e)
{
- throw e;
+ // the session was created while its server was starting, retry it:
+ if (e.getCode() == HornetQException.SESSION_CREATION_REJECTED)
+ {
+ ClientSessionImpl.log.warn("Server is starting, retry to create the session " + name);
+ retry = true;
+ // sleep a little bit to avoid spinning too much
+ Thread.sleep(10);
+ }
+ else
+ {
+ throw e;
+ }
}
}
- }
- while (retry);
+ while (retry);
- channel.clearCommands();
+ channel.clearCommands();
- for (Map.Entry<Long, ClientConsumerInternal> entry : consumers.entrySet())
- {
- SessionQueueQueryResponseMessage queueInfo = entry.getValue().getQueueInfo();
-
- // We try and recreate any non durable queues, since they probably won't be there unless
- // they are defined in hornetq-configuration.xml
- // This allows e.g. JMS non durable subs and temporary queues to continue to be used after failover
- if (!queueInfo.isDurable())
+ for (Map.Entry<Long, ClientConsumerInternal> entry : consumers.entrySet())
{
- CreateQueueMessage createQueueRequest = new CreateQueueMessage(queueInfo.getAddress(),
- queueInfo.getName(),
- queueInfo.getFilterString(),
- false,
- queueInfo.isTemporary(),
- false);
+ SessionQueueQueryResponseMessage queueInfo = entry.getValue().getQueueInfo();
- sendPacketWithoutLock(createQueueRequest);
- }
+ // We try and recreate any non durable queues, since they probably won't be there unless
+ // they are defined in hornetq-configuration.xml
+ // This allows e.g. JMS non durable subs and temporary queues to continue to be used after failover
+ if (!queueInfo.isDurable())
+ {
+ CreateQueueMessage createQueueRequest = new CreateQueueMessage(queueInfo.getAddress(),
+ queueInfo.getName(),
+ queueInfo.getFilterString(),
+ false,
+ queueInfo.isTemporary(),
+ false);
- SessionCreateConsumerMessage createConsumerRequest = new SessionCreateConsumerMessage(entry.getKey(),
- entry.getValue()
- .getQueueName(),
- entry.getValue()
- .getFilterString(),
- entry.getValue()
- .isBrowseOnly(),
- false);
+ sendPacketWithoutLock(createQueueRequest);
+ }
- sendPacketWithoutLock(createConsumerRequest);
+ SessionCreateConsumerMessage createConsumerRequest = new SessionCreateConsumerMessage(entry.getKey(),
+ entry.getValue()
+ .getQueueName(),
+ entry.getValue()
+ .getFilterString(),
+ entry.getValue()
+ .isBrowseOnly(),
+ false);
- int clientWindowSize = entry.getValue().getClientWindowSize();
+ sendPacketWithoutLock(createConsumerRequest);
- if (clientWindowSize != 0)
- {
- SessionConsumerFlowCreditMessage packet = new SessionConsumerFlowCreditMessage(entry.getKey(),
- clientWindowSize);
+ int clientWindowSize = entry.getValue().getClientWindowSize();
- sendPacketWithoutLock(packet);
+ if (clientWindowSize != 0)
+ {
+ SessionConsumerFlowCreditMessage packet = new SessionConsumerFlowCreditMessage(entry.getKey(),
+ clientWindowSize);
+
+ sendPacketWithoutLock(packet);
+ }
+ else
+ {
+ // https://jira.jboss.org/browse/HORNETQ-522
+ SessionConsumerFlowCreditMessage packet = new SessionConsumerFlowCreditMessage(entry.getKey(),
+ 1);
+ sendPacketWithoutLock(packet);
+ }
}
- else
+
+ if ((!autoCommitAcks || !autoCommitSends) && workDone)
{
- //https://jira.jboss.org/browse/HORNETQ-522
- SessionConsumerFlowCreditMessage packet = new SessionConsumerFlowCreditMessage(entry.getKey(),
- 1);
- sendPacketWithoutLock(packet);
+ // Session is transacted - set for rollback only
+ // FIXME - there is a race condition here - a commit could sneak in before this is set
+ rollbackOnly = true;
}
- }
- if ((!autoCommitAcks || !autoCommitSends) && workDone)
- {
- // Session is transacted - set for rollback only
- // FIXME - there is a race condition here - a commit could sneak in before this is set
- rollbackOnly = true;
- }
-
- // Now start the session if it was already started
- if (started)
- {
- for (ClientConsumerInternal consumer : consumers.values())
+ // Now start the session if it was already started
+ if (started)
{
- consumer.clearAtFailover();
- consumer.start();
- }
+ for (ClientConsumerInternal consumer : consumers.values())
+ {
+ consumer.clearAtFailover();
+ consumer.start();
+ }
- Packet packet = new PacketImpl(PacketImpl.SESS_START);
+ Packet packet = new PacketImpl(PacketImpl.SESS_START);
- packet.setChannelID(channel.getID());
+ packet.setChannelID(channel.getID());
- Connection conn = channel.getConnection().getTransportConnection();
+ Connection conn = channel.getConnection().getTransportConnection();
- HornetQBuffer buffer = packet.encode(channel.getConnection());
+ HornetQBuffer buffer = packet.encode(channel.getConnection());
- conn.write(buffer, false, false);
+ conn.write(buffer, false, false);
+ }
+
+ resetCreditManager = true;
}
- resetCreditManager = true;
+ channel.returnBlocking();
}
- channel.returnBlocking();
+ channel.setTransferring(false);
}
+ catch (Throwable t)
+ {
+ ClientSessionImpl.log.error("Failed to handle failover", t);
+ }
+ finally
+ {
+ channel.unlock();
+ }
- channel.setTransferring(false);
+ if (resetCreditManager)
+ {
+ producerCreditManager.reset();
+
+ // Also need to send more credits for consumers, otherwise the system could hand with the server
+ // not having any credits to send
+ }
}
- catch (Throwable t)
+
+ // Resetting the metadata after failover
+ try
{
- ClientSessionImpl.log.error("Failed to handle failover", t);
+ for (Map.Entry<String, String> entries : metadata.entrySet())
+ {
+ addMetaData(entries.getKey(), entries.getValue());
+ }
}
- finally
+ catch (HornetQException e)
{
- channel.unlock();
- }
- if (resetCreditManager)
- {
- producerCreditManager.reset();
+ log.warn("Error on resending metadata: " + metadata, e);
- // Also need to send more credits for consumers, otherwise the system could hand with the server
- // not having any credits to send
}
}
-
+
+ public void addMetaData(String key, String data) throws HornetQException
+ {
+ metadata.put(key, data);
+ channel.sendBlocking(new SessionAddMetaDataMessage(key, data));
+ }
+
public ClientSessionFactoryInternal getSessionFactory()
{
return sessionFactory;
@@ -1735,8 +1763,6 @@
}
}
-
-
private static class BindingQueryImpl implements BindingQuery
{
@@ -1823,9 +1849,4 @@
}
}
-
- public void addMetaData(String key, String data) throws HornetQException
- {
- channel.sendBlocking(new SessionAddMetaDataMessage(key, data));
- }
}
13 years, 2 months
JBoss hornetq SVN: r10322 - in branches/Branch_2_2_EAP: tests/src/org/hornetq/tests/integration/journal and 1 other directory.
by do-not-reply＠jboss.org
Author: clebert.suconic(a)jboss.com
Date: 2011-03-11 16:33:07 -0500 (Fri, 11 Mar 2011)
New Revision: 10322
Modified:
branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/JournalImpl.java
branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/journal/NIOJournalCompactTest.java
Log:
Adding new test
Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/JournalImpl.java
===================================================================
--- branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/JournalImpl.java 2011-03-11 21:32:40 UTC (rev 10321)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/JournalImpl.java 2011-03-11 21:33:07 UTC (rev 10322)
@@ -1610,6 +1610,7 @@
if (dataFilesToProcess.size() == 0)
{
+ trace("Finishing compacting, nothing to process");
return;
}
@@ -1740,7 +1741,7 @@
if (JournalImpl.trace)
{
- JournalImpl.log.debug("Finished compacting on journal");
+ trace("Finished compacting on journal");
}
if (JournalImpl.TRACE_RECORDS)
Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/journal/NIOJournalCompactTest.java
===================================================================
--- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/journal/NIOJournalCompactTest.java 2011-03-11 21:32:40 UTC (rev 10321)
+++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/journal/NIOJournalCompactTest.java 2011-03-11 21:33:07 UTC (rev 10322)
@@ -17,13 +17,21 @@
import java.io.FilenameFilter;
import java.util.ArrayList;
import java.util.Iterator;
+import java.util.LinkedList;
import java.util.List;
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.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import junit.framework.Assert;
import org.hornetq.api.core.Pair;
-import org.hornetq.core.journal.EncodingSupport;
+import org.hornetq.core.config.Configuration;
+import org.hornetq.core.journal.IOAsyncTask;
import org.hornetq.core.journal.PreparedTransactionInfo;
import org.hornetq.core.journal.RecordInfo;
import org.hornetq.core.journal.SequentialFile;
@@ -36,9 +44,13 @@
import org.hornetq.core.journal.impl.JournalImpl;
import org.hornetq.core.journal.impl.NIOSequentialFileFactory;
import org.hornetq.core.logging.Logger;
+import org.hornetq.core.persistence.impl.journal.JournalStorageManager;
+import org.hornetq.core.persistence.impl.journal.OperationContextImpl;
+import org.hornetq.core.server.impl.ServerMessageImpl;
import org.hornetq.tests.unit.core.journal.impl.JournalImplTestBase;
import org.hornetq.tests.unit.core.journal.impl.fakes.SimpleEncoding;
import org.hornetq.utils.IDGenerator;
+import org.hornetq.utils.OrderedExecutorFactory;
import org.hornetq.utils.SimpleIDGenerator;
/**
@@ -238,41 +250,41 @@
setup(2, 60 * 1024, false);
createJournal();
-
+
startJournal();
-
+
load();
startCompact();
-
+
addTx(1, 2);
-
+
prepare(1, new SimpleEncoding(10, (byte)0));
-
+
finishCompact();
-
+
stopJournal();
-
+
createJournal();
-
+
startJournal();
-
+
loadAndCheck();
-
+
startCompact();
-
+
commit(1);
-
+
finishCompact();
-
+
journal.compact();
-
+
stopJournal();
-
+
createJournal();
startJournal();
-
+
loadAndCheck();
}
@@ -281,37 +293,37 @@
setup(2, 60 * 1024, false);
createJournal();
-
+
startJournal();
-
+
load();
addTx(1, 2);
-
+
prepare(1, new SimpleEncoding(10, (byte)0));
-
+
stopJournal();
-
+
createJournal();
-
+
startJournal();
-
+
loadAndCheck();
-
+
startCompact();
-
+
commit(1);
-
+
finishCompact();
-
+
journal.compact();
-
+
stopJournal();
-
+
createJournal();
startJournal();
-
+
loadAndCheck();
}
@@ -320,29 +332,29 @@
setup(2, 60 * 1024, false);
createJournal();
-
+
startJournal();
-
+
load();
addTx(1, 2, 3);
-
+
prepare(1, new SimpleEncoding(10, (byte)0));
-
+
startCompact();
-
+
commit(1);
-
+
finishCompact();
-
+
journal.compact();
-
+
stopJournal();
-
+
createJournal();
startJournal();
-
+
loadAndCheck();
}
@@ -1446,7 +1458,7 @@
}
long tx1 = idGenerator.generateID();
-
+
journal.forceMoveNextFile();
ArrayList<Long> listToDelete = new ArrayList<Long>();
@@ -1469,7 +1481,7 @@
startCompact();
System.out.println("Committing TX " + tx1);
rollback(tx0);
- for (int i = 0 ; i < 10; i++)
+ for (int i = 0; i < 10; i++)
{
addTx(tx1, ids[i]);
}
@@ -1506,7 +1518,7 @@
}
long tx1 = idGenerator.generateID();
-
+
journal.forceMoveNextFile();
ArrayList<Long> listToDelete = new ArrayList<Long>();
@@ -1529,7 +1541,7 @@
startCompact();
System.out.println("Committing TX " + tx1);
rollback(tx0);
- for (int i = 0 ; i < 10; i++)
+ for (int i = 0; i < 10; i++)
{
addTx(tx1, ids[i]);
}
@@ -1564,11 +1576,11 @@
ids[i] = idGenerator.generateID();
addTx(tx0, ids[i]);
}
-
+
commit(tx0);
startCompact();
- for (int i = 0 ; i < 10; i++)
+ for (int i = 0; i < 10; i++)
{
delete(ids[i]);
}
@@ -1591,22 +1603,22 @@
long tx0 = idGenerator.generateID();
add(idGenerator.generateID());
- long ids[] = new long[]{idGenerator.generateID(), idGenerator.generateID()};
+ long ids[] = new long[] { idGenerator.generateID(), idGenerator.generateID() };
addTx(tx0, ids[0]);
addTx(tx0, ids[1]);
-
+
journal.forceMoveNextFile();
-
+
commit(tx0);
-
+
journal.forceMoveNextFile();
-
+
delete(ids[0]);
delete(ids[1]);
-
+
journal.forceMoveNextFile();
-
+
journal.compact();
stopJournal();
@@ -1700,6 +1712,161 @@
}
+ public void testStressDeletesNoSync() throws Exception
+ {
+ Configuration config = createBasicConfig();
+ config.setJournalFileSize(100 * 1024);
+ System.out.println(config.getJournalDirectory());
+ config.setJournalSyncNonTransactional(false);
+ config.setJournalSyncTransactional(false);
+// config.setJournalBufferTimeout_NIO(2000000000);
+// config.setJournalBufferTimeout_AIO(2000000000);
+ config.setJournalCompactMinFiles(0);
+ config.setJournalCompactPercentage(0);
+
+ final AtomicInteger errors = new AtomicInteger(0);
+
+ final AtomicBoolean running = new AtomicBoolean(true);
+
+ final AtomicLong seqGenerator = new AtomicLong(1);
+
+ final ExecutorService executor = Executors.newCachedThreadPool();
+
+ OrderedExecutorFactory factory = new OrderedExecutorFactory(executor);
+
+ final ExecutorService deleteExecutor = Executors.newCachedThreadPool();
+
+ final JournalStorageManager storage = new JournalStorageManager(config, factory);
+
+ storage.start();
+ storage.loadInternalOnly();
+
+ ((JournalImpl)storage.getMessageJournal()).setAutoReclaim(false);
+ final LinkedList<Long> survivingMsgs = new LinkedList<Long>();
+
+ Runnable producerRunnable = new Runnable()
+ {
+ public void run()
+ {
+ try
+ {
+ while (running.get())
+ {
+ final long[] values = new long[100];
+ long tx = seqGenerator.incrementAndGet();
+
+ OperationContextImpl ctx = new OperationContextImpl(executor);
+ storage.setContext(ctx);
+
+ for (int i = 0; i < 100; i++)
+ {
+ long id = seqGenerator.incrementAndGet();
+ values[i] = id;
+
+ ServerMessageImpl message = new ServerMessageImpl(id, 100);
+
+ message.getBodyBuffer().writeBytes(new byte[1024]);
+
+ storage.storeMessageTransactional(tx, message);
+ }
+ ServerMessageImpl message = new ServerMessageImpl(seqGenerator.incrementAndGet(), 100);
+
+ survivingMsgs.add(message.getMessageID());
+
+ // This one will stay here forever
+ storage.storeMessage(message);
+
+ storage.commit(tx);
+
+ ctx.executeOnCompletion(new IOAsyncTask()
+ {
+ public void onError(int errorCode, String errorMessage)
+ {
+ }
+
+ public void done()
+ {
+ deleteExecutor.execute(new Runnable()
+ {
+ public void run()
+ {
+ try
+ {
+ for (long messageID : values)
+ {
+ storage.deleteMessage(messageID);
+ }
+ }
+ catch (Exception e)
+ {
+ e.printStackTrace();
+ errors.incrementAndGet();
+ }
+
+ }
+ });
+ }
+ });
+
+ }
+ }
+ catch (Throwable e)
+ {
+ e.printStackTrace();
+ errors.incrementAndGet();
+ }
+ }
+ };
+
+ Runnable compressRunnable = new Runnable()
+ {
+ public void run()
+ {
+ try
+ {
+ int i = 0;
+ while (running.get())
+ {
+ Thread.sleep(500);
+ System.out.println("Compacting");
+ ((JournalImpl)storage.getMessageJournal()).compact();
+ ((JournalImpl)storage.getMessageJournal()).checkReclaimStatus();
+ }
+ }
+ catch (Throwable e)
+ {
+ e.printStackTrace();
+ errors.incrementAndGet();
+ }
+
+ }
+ };
+
+ Thread producerThread = new Thread(producerRunnable);
+ producerThread.start();
+
+ Thread compactorThread = new Thread(compressRunnable);
+ compactorThread.start();
+
+ Thread.sleep(10000);
+
+ running.set(false);
+
+ producerThread.join();
+
+ compactorThread.join();
+
+ executor.shutdown();
+
+ assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS));
+
+ deleteExecutor.shutdown();
+
+ assertTrue(deleteExecutor.awaitTermination(30, TimeUnit.SECONDS));
+
+ storage.stop();
+ }
+
@Override
protected void setUp() throws Exception
{
13 years, 2 months
JBoss hornetq SVN: r10321 - branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/stress/journal.
by do-not-reply＠jboss.org
Author: clebert.suconic(a)jboss.com
Date: 2011-03-11 16:32:40 -0500 (Fri, 11 Mar 2011)
New Revision: 10321
Modified:
branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/stress/journal/CompactingStressTest.java
Log:
tweak
Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/stress/journal/CompactingStressTest.java
===================================================================
--- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/stress/journal/CompactingStressTest.java 2011-03-11 14:07:52 UTC (rev 10320)
+++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/stress/journal/CompactingStressTest.java 2011-03-11 21:32:40 UTC (rev 10321)
@@ -71,8 +71,6 @@
if (AsynchronousFileImpl.isLoaded())
{
internalTestCleanup(JournalType.ASYNCIO);
- tearDown();
- setUp();
}
}
@@ -163,7 +161,10 @@
public void testMultiProducerAndCompactAIO() throws Throwable
{
- internalTestMultiProducer(JournalType.ASYNCIO);
+ if (AsynchronousFileImpl.isLoaded())
+ {
+ internalTestMultiProducer(JournalType.ASYNCIO);
+ }
}
public void testMultiProducerAndCompactNIO() throws Throwable
13 years, 2 months
JBoss hornetq SVN: r10320 - branches/Branch_2_2_EAP/docs/quickstart-guide.
by do-not-reply＠jboss.org
Author: clebert.suconic(a)jboss.com
Date: 2011-03-11 09:07:52 -0500 (Fri, 11 Mar 2011)
New Revision: 10320
Modified:
branches/Branch_2_2_EAP/docs/quickstart-guide/
Log:
svn:ignore
Property changes on: branches/Branch_2_2_EAP/docs/quickstart-guide
___________________________________________________________________
Modified: svn:ignore
- build
+ build
target
13 years, 2 months
JBoss hornetq SVN: r10319 - branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration.
by do-not-reply＠jboss.org
Author: clebert.suconic(a)jboss.com
Date: 2011-03-11 09:05:24 -0500 (Fri, 11 Mar 2011)
New Revision: 10319
Modified:
branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/DuplicateDetectionTest.java
Log:
fixing test
Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/DuplicateDetectionTest.java
===================================================================
--- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/DuplicateDetectionTest.java 2011-03-11 12:56:27 UTC (rev 10318)
+++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/DuplicateDetectionTest.java 2011-03-11 14:05:24 UTC (rev 10319)
@@ -582,7 +582,7 @@
}
catch (HornetQException e)
{
- assertEquals(e.getCode(), HornetQException.TRANSACTION_ROLLED_BACK);
+ assertEquals(e.getCode(), HornetQException.DUPLICATE_ID_REJECTED);
session.rollback();
}
@@ -1725,7 +1725,7 @@
}
catch (HornetQException e)
{
- assertEquals(e.getCode(), HornetQException.TRANSACTION_ROLLED_BACK);
+ assertEquals(e.getCode(), HornetQException.DUPLICATE_ID_REJECTED);
session.rollback();
}
@@ -1742,7 +1742,7 @@
}
catch (HornetQException e)
{
- assertEquals(e.getCode(), HornetQException.TRANSACTION_ROLLED_BACK);
+ assertEquals(e.getCode(), HornetQException.DUPLICATE_ID_REJECTED);
session.rollback();
}
13 years, 2 months
JBoss hornetq SVN: r10318 - in branches/Branch_2_2_EAP/src: main/org/hornetq/core/config and 3 other directories.
by do-not-reply＠jboss.org
Author: ataylor
Date: 2011-03-11 07:56:27 -0500 (Fri, 11 Mar 2011)
New Revision: 10318
Modified:
branches/Branch_2_2_EAP/src/config/common/schema/hornetq-configuration.xsd
branches/Branch_2_2_EAP/src/main/org/hornetq/core/config/Configuration.java
branches/Branch_2_2_EAP/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java
branches/Branch_2_2_EAP/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java
branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java
Log:
https://issues.jboss.org/browse/JBPAPP-6049 - added longer delay for failback and also made it configurable
Modified: branches/Branch_2_2_EAP/src/config/common/schema/hornetq-configuration.xsd
===================================================================
--- branches/Branch_2_2_EAP/src/config/common/schema/hornetq-configuration.xsd 2011-03-11 05:11:27 UTC (rev 10317)
+++ branches/Branch_2_2_EAP/src/config/common/schema/hornetq-configuration.xsd 2011-03-11 12:56:27 UTC (rev 10318)
@@ -71,8 +71,10 @@
</xsd:element>
<xsd:element maxOccurs="1" minOccurs="0" name="allow-failback" type="xsd:boolean">
</xsd:element>
- <xsd:element maxOccurs="1" minOccurs="0" name="failover-on-shutdown" type="xsd:boolean">
+ <xsd:element maxOccurs="1" minOccurs="0" name="failback-delay" type="xsd:long">
</xsd:element>
+ <xsd:element maxOccurs="1" minOccurs="0" name="failover-on-shutdown" type="xsd:boolean">
+ </xsd:element>
<xsd:element maxOccurs="1" minOccurs="0" name="shared-store" type="xsd:boolean">
</xsd:element>
<xsd:element maxOccurs="1" minOccurs="0" name="persist-delivery-count-before-delivery" type="xsd:boolean">
Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/config/Configuration.java
===================================================================
--- branches/Branch_2_2_EAP/src/main/org/hornetq/core/config/Configuration.java 2011-03-11 05:11:27 UTC (rev 10317)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/config/Configuration.java 2011-03-11 12:56:27 UTC (rev 10318)
@@ -863,4 +863,14 @@
* @return
*/
List<ConnectorServiceConfiguration> getConnectorServiceConfigurations();
+
+ /*
+ * how long to wait before failback occurs on restart
+ * */
+ long getFailbackDelay();
+
+ /*
+ * set the failback delay
+ * */
+ void setFailbackDelay(long delay);
}
Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java
===================================================================
--- branches/Branch_2_2_EAP/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java 2011-03-11 05:11:27 UTC (rev 10317)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java 2011-03-11 12:56:27 UTC (rev 10318)
@@ -180,6 +180,8 @@
public static final long DEFAULT_MEMORY_MEASURE_INTERVAL = -1; // in milliseconds
+ public static final long DEFAULT_FAILBACK_DELAY = 5000; //in milliseconds
+
public static final String DEFAULT_LOG_DELEGATE_FACTORY_CLASS_NAME = JULLogDelegateFactory.class.getCanonicalName();
// Attributes -----------------------------------------------------------------------------
@@ -334,7 +336,9 @@
private Map<String, Set<Role>> securitySettings = new HashMap<String, Set<Role>>();
protected List<ConnectorServiceConfiguration> connectorServiceConfigurations = new ArrayList<ConnectorServiceConfiguration>();
-
+
+ private long failbackDelay = ConfigurationImpl.DEFAULT_FAILBACK_DELAY;
+
// Public -------------------------------------------------------------------------
public boolean isClustered()
@@ -1357,7 +1361,17 @@
{
return this.connectorServiceConfigurations;
}
-
+
+ public long getFailbackDelay()
+ {
+ return failbackDelay;
+ }
+
+ public void setFailbackDelay(long failbackDelay)
+ {
+ this.failbackDelay = failbackDelay;
+ }
+
public void setConnectorServiceConfigurations(final List<ConnectorServiceConfiguration> configs)
{
this.connectorServiceConfigurations = configs;
Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java
===================================================================
--- branches/Branch_2_2_EAP/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java 2011-03-11 05:11:27 UTC (rev 10317)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java 2011-03-11 12:56:27 UTC (rev 10318)
@@ -169,6 +169,8 @@
config.setAllowAutoFailBack(XMLConfigurationUtil.getBoolean(e, "allow-failback", config.isClustered()));
+ config.setFailbackDelay(XMLConfigurationUtil.getLong(e, "failback-delay", config.getFailbackDelay(), Validators.GT_ZERO));
+
config.setFailoverOnServerShutdown(XMLConfigurationUtil.getBoolean(e,
"failover-on-shutdown",
config.isFailoverOnServerShutdown()));
Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java
===================================================================
--- branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-03-11 05:11:27 UTC (rev 10317)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-03-11 12:56:27 UTC (rev 10318)
@@ -346,7 +346,7 @@
// goes down they failover to us
clusterManager.announceBackup();
//
- Thread.sleep(2000);
+ Thread.sleep(configuration.getFailbackDelay());
}
nodeManager.startLiveNode();
13 years, 2 months
JBoss hornetq SVN: r10317 - in branches/Branch_2_2_EAP: src/main/org/hornetq/core/postoffice/impl and 2 other directories.
by do-not-reply＠jboss.org
Author: clebert.suconic(a)jboss.com
Date: 2011-03-11 00:11:27 -0500 (Fri, 11 Mar 2011)
New Revision: 10317
Modified:
branches/Branch_2_2_EAP/src/main/org/hornetq/api/core/HornetQException.java
branches/Branch_2_2_EAP/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java
branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/PacketImpl.java
branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/AsynchronousFailoverTest.java
branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java
Log:
https://issues.jboss.org/browse/JBPAPP-6080 - DuplicateID changes
Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/api/core/HornetQException.java
===================================================================
--- branches/Branch_2_2_EAP/src/main/org/hornetq/api/core/HornetQException.java 2011-03-10 20:55:26 UTC (rev 10316)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/api/core/HornetQException.java 2011-03-11 05:11:27 UTC (rev 10317)
@@ -128,6 +128,11 @@
* server is starting and has not finish to be initialized)
*/
public static final int SESSION_CREATION_REJECTED = 112;
+
+ /**
+ * A DuplicateID was rejected.
+ */
+ public static final int DUPLICATE_ID_REJECTED = 113;
// Native Error codes ----------------------------------------------
Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java
===================================================================
--- branches/Branch_2_2_EAP/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2011-03-10 20:55:26 UTC (rev 10316)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2011-03-11 05:11:27 UTC (rev 10317)
@@ -1064,7 +1064,7 @@
if (context.getTransaction() != null)
{
- context.getTransaction().markAsRollbackOnly(new HornetQException(HornetQException.TRANSACTION_ROLLED_BACK, warnMessage.toString()));
+ context.getTransaction().markAsRollbackOnly(new HornetQException(HornetQException.DUPLICATE_ID_REJECTED, warnMessage.toString()));
}
return false;
@@ -1107,7 +1107,7 @@
if (context.getTransaction() != null)
{
- context.getTransaction().markAsRollbackOnly(new HornetQException(HornetQException.TRANSACTION_ROLLED_BACK, warnMessage.toString()));
+ context.getTransaction().markAsRollbackOnly(new HornetQException(HornetQException.DUPLICATE_ID_REJECTED, warnMessage.toString()));
}
return false;
Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/PacketImpl.java
===================================================================
--- branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/PacketImpl.java 2011-03-10 20:55:26 UTC (rev 10316)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/protocol/core/impl/PacketImpl.java 2011-03-11 05:11:27 UTC (rev 10317)
@@ -181,17 +181,17 @@
public static final byte REPLICATION_COMPARE_DATA = 102;
public static final byte REPLICATION_SYNC = 103;
-
+
// HA
+ public static final byte SESS_ADD_METADATA = 104;
+
public static final byte CLUSTER_TOPOLOGY = 110;
public static final byte NODE_ANNOUNCE = 111;
public static final byte SUBSCRIBE_TOPOLOGY = 112;
- public static final byte SESS_ADD_METADATA = 104;
-
// Static --------------------------------------------------------
public PacketImpl(final byte type)
Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/AsynchronousFailoverTest.java
===================================================================
--- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/AsynchronousFailoverTest.java 2011-03-10 20:55:26 UTC (rev 10316)
+++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/AsynchronousFailoverTest.java 2011-03-11 05:11:27 UTC (rev 10317)
@@ -412,6 +412,11 @@
}
catch (HornetQException e)
{
+ if (e.getCode() == HornetQException.DUPLICATE_ID_REJECTED)
+ {
+ break;
+ }
+ else
if (e.getCode() == HornetQException.TRANSACTION_ROLLED_BACK || e.getCode() == HornetQException.UNBLOCKED)
{
// OK
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-03-10 20:55:26 UTC (rev 10316)
+++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java 2011-03-11 05:11:27 UTC (rev 10317)
@@ -1977,7 +1977,14 @@
producer.send(message);
}
- session2.commit();
+ try
+ {
+ session2.commit();
+ }
+ catch (HornetQException e)
+ {
+ assertEquals(HornetQException.DUPLICATE_ID_REJECTED, e.getCode());
+ }
ClientConsumer consumer = session2.createConsumer(FailoverTestBase.ADDRESS);
13 years, 2 months