[hornetq-commits] JBoss hornetq SVN: r8083 - in trunk: docs/user-manual/en and 28 other directories.

Author: timfox Date: 2009-10-13 06:06:28 -0400 (Tue, 13 Oct 2009) New Revision: 8083 Modified: trunk/docs/user-manual/en/client-reconnection.xml trunk/pom.xml trunk/src/config/common/schema/hornetq-jms.xsd trunk/src/main/org/hornetq/core/client/ClientSessionFactory.java trunk/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java trunk/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java trunk/src/main/org/hornetq/core/client/impl/DelegatingSession.java trunk/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java trunk/src/main/org/hornetq/core/remoting/Channel.java trunk/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java trunk/src/main/org/hornetq/core/remoting/impl/wireformat/CreateSessionResponseMessage.java trunk/src/main/org/hornetq/core/remoting/impl/wireformat/ReattachSessionResponseMessage.java trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java trunk/src/main/org/hornetq/jms/client/HornetQConnection.java trunk/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java trunk/src/main/org/hornetq/jms/client/JMSExceptionHelper.java trunk/src/main/org/hornetq/jms/server/JMSServerManager.java trunk/src/main/org/hornetq/jms/server/config/ConnectionFactoryConfiguration.java trunk/src/main/org/hornetq/jms/server/config/impl/ConnectionFactoryConfigurationImpl.java trunk/src/main/org/hornetq/jms/server/impl/JMSServerDeployer.java trunk/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java trunk/src/main/org/hornetq/jms/server/management/JMSServerControl.java trunk/src/main/org/hornetq/jms/server/management/impl/JMSServerControlImpl.java trunk/tests/jms-tests/src/org/hornetq/jms/tests/CTSMiscellaneousTest.java trunk/tests/jms-tests/src/org/hornetq/jms/tests/JMSTestCase.java trunk/tests/jms-tests/src/org/hornetq/jms/tests/tools/container/LocalTestServer.java trunk/tests/src/org/hornetq/tests/integration/client/ConsumerTest.java trunk/tests/src/org/hornetq/tests/integration/client/HeuristicXATest.java trunk/tests/src/org/hornetq/tests/integration/client/HornetQCrashTest.java trunk/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadRandomReattachTestBase.java trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/NettyMultiThreadRandomReattachTest.java trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/RandomReattachTest.java trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/ReattachTest.java trunk/tests/src/org/hornetq/tests/integration/cluster/restart/ClusterRestartTest.java trunk/tests/src/org/hornetq/tests/integration/jms/FloodServerTest.java trunk/tests/src/org/hornetq/tests/integration/jms/ManualReconnectionToSingleServerTest.java trunk/tests/src/org/hornetq/tests/integration/jms/client/PreACKJMSTest.java trunk/tests/src/org/hornetq/tests/integration/jms/client/ResendTest.java trunk/tests/src/org/hornetq/tests/integration/jms/connection/CloseConnectionOnGCTest.java trunk/tests/src/org/hornetq/tests/integration/jms/divert/DivertAndACKClientTest.java trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlTest.java trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlUsingJMSTest.java trunk/tests/src/org/hornetq/tests/integration/server/LVQTest.java trunk/tests/src/org/hornetq/tests/util/JMSTestBase.java Log: various fixes, failover changes, fixed tests etc Modified: trunk/docs/user-manual/en/client-reconnection.xml =================================================================== --- trunk/docs/user-manual/en/client-reconnection.xml 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/docs/user-manual/en/client-reconnection.xml 2009-10-13 10:06:28 UTC (rev 8083) @@ -1,5 +1,4 @@ <?xml version="1.0" encoding="UTF-8"?> - <!-- ============================================================================= --> <!-- Copyright © 2009 Red Hat, Inc. and others. --> <!-- --> @@ -17,14 +16,31 @@ <!-- and agrees not to assert, Section 4d of CC-BY-SA to the fullest extent --> <!-- permitted by applicable law. --> <!-- ============================================================================= --> - <chapter id="client-reconnection"> <title>Client Reconnection</title> - <para>HornetQ clients can be configured to automatically reconnect to the server in the - event that a failure is detected in the connection between the client and the server. If the - client successfully reconnects, and the server still has a record of the clients session - (i.e. the server was not restarted) then the client will transparently re-attach to the - session and it will be able to resume as if nothing had happened.</para> + <para>HornetQ clients can be configured to automatically reconnect to the server in the event + that a failure is detected in the connection between the client and the server. </para> + <para>By default, when a client connection reconnects, HornetQ will automatically recreate any + sessions and consumers on the server. If a particular session is transacted and messages + have already been sent or acknowledged in the current transaction but not committed yet, + then the transaction will be marked as rollback only. This is because HornetQ cannot + guarantee that those messages or acks have really reached the server because of the + connection failure. In this case, any subsequent attempt to commit the transaction will + throw an exception. This exception can be caught and the transaction can be retried.</para> + <para>If you are using the core API, the exception thrown will be instance of HornetQException + with the error code TRANSACTION_ROLLED_BACK. If you are using the JMS API, the exception + will be a javax.jms.TransactionRolledBackException. </para> + <para>For a transacted session if a connection failure occurred during the call to commit(), + it's not possible for the client to determine if the commit was successfully processed on + the server before failure. In this case, if the transaction is retried after reconnection, + be sure to use <link linkend="duplicate-detection">duplicate detection</link> in your messages to prevent them being processed more + than once. </para> + <para>For a non transacted session, after the sessions and consumers have been recreated, + messages or acknowledgements that were in transit at the time of the failure might have been + lost. This could result in lost sent messages or duplicate delivery of messages. If you want + guaranteed once and only once message delivery on failure, you need to use transacted + session with duplicate detection.</para> + <para>Reattach - TODO</para> <para>Client reconnection is also used internally by components such as core bridges to allow them to reconnect to their target servers.</para> <para>Client reconnection is configured using the following parameters:</para> @@ -51,11 +67,22 @@ spaced at equal intervals.</para> </listitem> <listitem> + <para><literal>max-retry-interval</literal>. This optional parameter determines the + maximum retry interval that will be used. When setting <literal + >retry-interval-multiplier</literal> it would otherwise be possible that + subsequent retries exponentially increase to ridiculously large values. By setting + this parameter you can set an upper limit on that value. The default value is + <literal>2000</literal> milliseconds.</para> + </listitem> + <listitem> <para><literal>reconnect-attempts</literal>. This optional parameter determines the total number of reconnect attempts the bridge will make before giving up and shutting down. A value of <literal>-1</literal> signifies an unlimited number of attempts. The default value is <literal>-1</literal>.</para> </listitem> + <listitem> + <para><literal>use-reattach</literal>. </para> + </listitem> </itemizedlist> <para>If you're using JMS, and you're using the JMS Service on the server to load your JMS connection factory instances directly into JNDI, then you can specify these parameters in @@ -69,6 +96,7 @@ &lt;/entries&gt; &lt;retry-interval&gt;1000&lt;/retry-interval&gt; &lt;retry-interval-multiplier&gt;1.5&lt;/retry-interval-multiplier&gt; +&lt;max-retry-interval&gt;60000&lt;/max-retry-interval&gt; &lt;reconnect-attempts&gt;1000&lt;/reconnect-attempts&gt; &lt;/connection-factory&gt; </programlisting> @@ -76,8 +104,8 @@ specify the parameters using the appropriate setter methods on the <literal <para>If you're using the core API and instantiating the <literal>ClientSessionFactory</literal> - instance directly you can also specify the parameters using the appropriate setter methods on the - <literal>ClientSessionFactory</literal> immediately after creating it.</para> + instance directly you can also specify the parameters using the appropriate setter methods + on the <literal>ClientSessionFactory</literal> immediately after creating it.</para> <para>If your client does manage to reconnect but the session is no longer available on the server, for instance if the server has been restarted or it has timed out, then the client won't be able to re-attach, and any <literal>ExceptionListener</literal> or <literal Modified: trunk/pom.xml =================================================================== --- trunk/pom.xml 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/pom.xml 2009-10-13 10:06:28 UTC (rev 8083) @@ -222,7 +222,7 @@ <dependency> <groupId>org.jboss.netty</groupId> <artifactId>netty</artifactId> - <version>3.1.3.GA</version> + <version>3.1.5.GA</version> </dependency> <!--needed to compile the logging jar--> <dependency> Modified: trunk/src/config/common/schema/hornetq-jms.xsd =================================================================== --- trunk/src/config/common/schema/hornetq-jms.xsd 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/config/common/schema/hornetq-jms.xsd 2009-10-13 10:06:28 UTC (rev 8083) @@ -102,9 +102,6 @@ <xsd:element name="reconnect-attempts" type="xsd:int" maxOccurs="1" minOccurs="0"> </xsd:element> - <xsd:element name="use-reattach" type="xsd:boolean" - maxOccurs="1" minOccurs="0"> - </xsd:element> <xsd:element name="failover-on-server-shutdown" type="xsd:boolean" maxOccurs="1" minOccurs="0"> </xsd:element> Modified: trunk/src/main/org/hornetq/core/client/ClientSessionFactory.java =================================================================== --- trunk/src/main/org/hornetq/core/client/ClientSessionFactory.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/core/client/ClientSessionFactory.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -146,10 +146,6 @@ void setReconnectAttempts(int reconnectAttempts); - boolean isUseReattach(); - - void setUseReattach(boolean reattach); - boolean isFailoverOnServerShutdown(); void setFailoverOnServerShutdown(boolean failoverOnServerShutdown); Modified: trunk/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -103,8 +103,6 @@ public static final int DEFAULT_RECONNECT_ATTEMPTS = 0; - public static final boolean DEFAULT_USE_REATTACH = false; - public static final boolean DEFAULT_FAILOVER_ON_SERVER_SHUTDOWN = false; public static final boolean DEFAULT_USE_GLOBAL_POOLS = true; @@ -192,8 +190,6 @@ private int reconnectAttempts; - private boolean useReattach; - private volatile boolean closed; private boolean failoverOnServerShutdown; @@ -290,8 +286,7 @@ retryInterval, retryIntervalMultiplier, maxRetryInterval, - reconnectAttempts, - useReattach, + reconnectAttempts, threadPool, scheduledThreadPool, interceptors); @@ -363,8 +358,6 @@ reconnectAttempts = DEFAULT_RECONNECT_ATTEMPTS; - useReattach = DEFAULT_USE_REATTACH; - failoverOnServerShutdown = DEFAULT_FAILOVER_ON_SERVER_SHUTDOWN; } @@ -666,17 +659,6 @@ this.reconnectAttempts = reconnectAttempts; } - public synchronized boolean isUseReattach() - { - return useReattach; - } - - public synchronized void setUseReattach(boolean reattach) - { - checkWrite(); - this.useReattach = reattach; - } - public synchronized boolean isFailoverOnServerShutdown() { return failoverOnServerShutdown; @@ -938,8 +920,7 @@ retryInterval, retryIntervalMultiplier, maxRetryInterval, - reconnectAttempts, - useReattach, + reconnectAttempts, threadPool, scheduledThreadPool, interceptors); Modified: trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -371,9 +371,8 @@ { return createConsumer(queueName, filterString, consumerWindowSize, consumerMaxRate, browseOnly); } - - public ClientConsumer createConsumer(final SimpleString queueName, - final boolean browseOnly) throws HornetQException + + public ClientConsumer createConsumer(final SimpleString queueName, final boolean browseOnly) throws HornetQException { return createConsumer(queueName, null, consumerWindowSize, consumerMaxRate, browseOnly); } @@ -382,7 +381,7 @@ { return createConsumer(toSimpleString(queueName), toSimpleString(filterString), browseOnly); } - + public ClientConsumer createConsumer(final String queueName, final boolean browseOnly) throws HornetQException { return createConsumer(toSimpleString(queueName), null, browseOnly); @@ -580,7 +579,7 @@ if (!started) { for (ClientConsumerInternal clientConsumerInternal : consumers.values()) - { + { clientConsumerInternal.start(); } @@ -691,7 +690,7 @@ public void removeConsumer(final ClientConsumerInternal consumer) throws HornetQException { - consumers.remove(consumer.getID()); + consumers.remove(consumer.getID()); } public void removeProducer(final ClientProducerInternal producer) @@ -786,17 +785,14 @@ sendAckHandler = handler; } - // Needs to be synchronized to prevent issues with occurring concurrently with close() - - //TODO - need to reenable - public synchronized boolean handleReattach(final RemotingConnection backupConnection) + // Needs to be synchronized to prevent issues with occurring concurrently with close() + + public synchronized void handleFailover(final RemotingConnection backupConnection) { if (closed) { - return true; + return; } - - boolean ok = false; // We lock the channel to prevent any packets to be added to the resend // cache during the failover process @@ -805,7 +801,7 @@ try { channel.transferConnection(backupConnection); - + backupConnection.syncIDGeneratorSequence(remotingConnection.getIDGeneratorSequence()); remotingConnection = backupConnection; @@ -816,115 +812,44 @@ ReattachSessionResponseMessage response = (ReattachSessionResponseMessage)channel1.sendBlocking(request); - if (response.isSessionFound()) - { - channel.replayCommands(response.getLastReceivedCommandID(), channel.getID()); + if (response.isReattached()) + { + // The session was found on the server - we reattached transparently ok - ok = true; + channel.replayCommands(response.getLastReceivedCommandID(), channel.getID()); } else { - if (closedSent) - { - // a session re-attach may fail, if the session close was sent before failover started, hit the server, - // processed, then before the response was received back, failover occurred, re-attach was attempted. in - // this case it's ok - we don't want to call any failure listeners and we don't want to halt the rest of - // the failover process. - // - // however if session re-attach fails and the session was not in a call to close, then we DO want to call - // the session listeners so we return false - // - // Also session reattach will fail if the server is restarted - so the session is lost - ok = true; - } - else - { - log.warn(System.identityHashCode(this) + " Session not found on server when attempting to re-attach"); - } + // 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 + + Packet createRequest = new CreateSessionMessage(name, + channel.getID(), + version, + username, + password, + minLargeMessageSize, + xa, + autoCommitSends, + autoCommitAcks, + preAcknowledge, + producerWindowSize); - channel.returnBlocking(); - } - - } - catch (Throwable t) - { - log.error("Failed to handle failover", t); - } - finally - { - channel.unlock(); - } - - return ok; - } - - public void workDone() - { - workDone = true; - } - - // Needs to be synchronized to prevent issues with occurring concurrently with close() - public synchronized boolean handleFailover(final RemotingConnection backupConnection) - { - if (closed) - { - return true; - } - - boolean ok = false; + channel1.sendBlocking(createRequest); - // Need to stop all consumers outside the lock - for (ClientConsumerInternal consumer : consumers.values()) - { - try - { - consumer.stop(); - } - catch (HornetQException e) - { - log.error("Failed to stop consumer", e); - } - - consumer.clearAtFailover(); - } - - // We lock the channel to prevent any packets being sent during the failover process - channel.lock(); - - try - { - channel.transferConnection(backupConnection); - - remotingConnection = backupConnection; - - Packet request = new CreateSessionMessage(name, - channel.getID(), - version, - username, - password, - minLargeMessageSize, - xa, - autoCommitSends, - autoCommitAcks, - preAcknowledge, - producerWindowSize); - - Channel channel1 = backupConnection.getChannel(1, -1, false); - - CreateSessionResponseMessage response = (CreateSessionResponseMessage)channel1.sendBlocking(request); - - if (response.isCreated()) - { - // Session was created ok - - // Now we need to recreate the consumers - + channel.clearCommands(); + for (Map.Entry<Long, ClientConsumerInternal> entry : consumers.entrySet()) { SessionCreateConsumerMessage createConsumerRequest = new SessionCreateConsumerMessage(entry.getKey(), - entry.getValue().getQueueName(), - entry.getValue().getFilterString(), - entry.getValue().isBrowseOnly(), + entry.getValue() + .getQueueName(), + entry.getValue() + .getFilterString(), + entry.getValue() + .isBrowseOnly(), false); createConsumerRequest.setChannelID(channel.getID()); @@ -936,23 +861,24 @@ createConsumerRequest.encode(buffer); conn.write(buffer, false); - + int clientWindowSize = calcWindowSize(entry.getValue().getClientWindowSize()); - + if (clientWindowSize != 0) { - SessionConsumerFlowCreditMessage packet = new SessionConsumerFlowCreditMessage(entry.getKey(), clientWindowSize); - + SessionConsumerFlowCreditMessage packet = new SessionConsumerFlowCreditMessage(entry.getKey(), + clientWindowSize); + packet.setChannelID(channel.getID()); - + buffer = conn.createBuffer(packet.getRequiredBufferSize()); packet.encode(buffer); - conn.write(buffer, false); + conn.write(buffer, false); } } - + if ((!autoCommitAcks || !autoCommitSends) && workDone) { // Session is transacted - set for rollback only @@ -968,30 +894,23 @@ { consumer.start(); } - + Packet packet = new PacketImpl(PacketImpl.SESS_START); - + packet.setChannelID(channel.getID()); - + Connection conn = channel.getConnection().getTransportConnection(); - + HornetQBuffer buffer = conn.createBuffer(packet.getRequiredBufferSize()); packet.encode(buffer); - conn.write(buffer, false); + conn.write(buffer, false); } - - ok = true; + + channel.returnBlocking(); } - else - { - // This means the server we failed onto is not ready to take new sessions - perhaps it hasn't actually - // failed over - } - // We cause any blocking calls to return - since they won't get responses. - channel.returnBlocking(); } catch (Throwable t) { @@ -1001,10 +920,13 @@ { channel.unlock(); } - - return ok; } + public void workDone() + { + workDone = true; + } + public void returnBlocking() { channel.returnBlocking(); @@ -1033,7 +955,7 @@ public void commit(final Xid xid, final boolean onePhase) throws XAException { checkXA(); - + if (rollbackOnly) { throw new XAException(XAException.XA_RBOTHER); @@ -1049,7 +971,7 @@ SessionXAResponseMessage response = (SessionXAResponseMessage)channel.sendBlocking(packet); workDone = false; - + if (response.isError()) { throw new XAException(response.getResponseCode()); @@ -1065,12 +987,12 @@ public void end(final Xid xid, final int flags) throws XAException { checkXA(); - + if (rollbackOnly) { throw new XAException(XAException.XA_RBOTHER); } - + try { Packet packet; @@ -1391,10 +1313,10 @@ { throw new IllegalArgumentException("Invalid window size " + windowSize); } - + return clientWindowSize; } - + /** * @param queueName * @param filterString @@ -1410,10 +1332,14 @@ final boolean browseOnly) throws HornetQException { checkClosed(); - + long consumerID = idGenerator.generateID(); - SessionCreateConsumerMessage request = new SessionCreateConsumerMessage(consumerID, queueName, filterString, browseOnly, true); + SessionCreateConsumerMessage request = new SessionCreateConsumerMessage(consumerID, + queueName, + filterString, + browseOnly, + true); channel.sendBlocking(request); @@ -1422,7 +1348,7 @@ // The value we send is just a hint int clientWindowSize = calcWindowSize(windowSize); - + ClientConsumerInternal consumer = new ClientConsumerImpl(this, consumerID, queueName, Modified: trunk/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -53,9 +53,9 @@ void handleReceiveContinuation(long consumerID, SessionReceiveContinuationMessage continuation) throws Exception; - boolean handleFailover(RemotingConnection backupConnection); + void handleFailover(RemotingConnection backupConnection); - boolean handleReattach(RemotingConnection backupConnection); + // boolean handleReattach(RemotingConnection backupConnection); RemotingConnection getConnection(); Modified: trunk/src/main/org/hornetq/core/client/impl/DelegatingSession.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/DelegatingSession.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/core/client/impl/DelegatingSession.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -368,16 +368,11 @@ return session.getXAResource(); } - public boolean handleFailover(RemotingConnection backupConnection) + public void handleFailover(RemotingConnection backupConnection) { - return session.handleFailover(backupConnection); + session.handleFailover(backupConnection); } - public boolean handleReattach(RemotingConnection backupConnection) - { - return session.handleReattach(backupConnection); - } - public void handleReceiveContinuation(long consumerID, SessionReceiveContinuationMessage continuation) throws Exception { session.handleReceiveContinuation(consumerID, continuation); Modified: trunk/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -135,8 +135,6 @@ private final List<Interceptor> interceptors; - private final boolean useReattach; - // Static // --------------------------------------------------------------------------------------- @@ -153,8 +151,7 @@ final long retryInterval, final double retryIntervalMultiplier, final long maxRetryInterval, - final int reconnectAttempts, - final boolean useReattach, + final int reconnectAttempts, final ExecutorService threadPool, final ScheduledExecutorService scheduledThreadPool, final List<Interceptor> interceptors) @@ -198,8 +195,6 @@ this.reconnectAttempts = reconnectAttempts; - this.useReattach = useReattach; - this.scheduledThreadPool = scheduledThreadPool; this.threadPool = threadPool; @@ -487,8 +482,6 @@ private void failoverOrReconnect(final Object connectionID, final HornetQException me) { - boolean done = false; - synchronized (failoverLock) { if (connection == null || connection.getID() != connectionID) @@ -610,23 +603,14 @@ backupTransportParams = null; - done = reattachSessions(reconnectAttempts == -1 ? -1 : reconnectAttempts + 1, false); + reconnectSessions(reconnectAttempts == -1 ? -1 : reconnectAttempts + 1); } else { - done = reattachSessions(reconnectAttempts, useReattach); + reconnectSessions(reconnectAttempts); } - - if (done) - { - // Destroy the old connection - - oldConnection.destroy(); - } - else - { - oldConnection.destroy(); - } + + oldConnection.destroy(); } else { @@ -663,7 +647,7 @@ /* * Re-attach sessions all pre-existing sessions to the new remoting connection */ - private boolean reattachSessions(final int reconnectAttempts, final boolean reattach) + private void reconnectSessions(final int reconnectAttempts) { RemotingConnection backupConnection = getConnectionWithRetry(reconnectAttempts); @@ -671,7 +655,7 @@ { log.warn("Failed to connect to server."); - return false; + return; } List<FailureListener> oldListeners = connection.getFailureListeners(); @@ -689,33 +673,11 @@ } backupConnection.setFailureListeners(newListeners); - - boolean ok = true; - - // If all connections got ok, then handle failover + for (ClientSessionInternal session : sessions) { - boolean b; - - if (reattach) - { - b = session.handleReattach(backupConnection); - } - else - { - b = session.handleFailover(backupConnection); - } - - if (!b) - { - // If a session fails to re-attach we doom the lot, but we make sure we try all sessions and don't exit - // early - // or connections might be left lying around - ok = false; - } + session.handleFailover(backupConnection); } - - return ok; } private RemotingConnection getConnectionWithRetry(final int reconnectAttempts) Modified: trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -723,8 +723,6 @@ } } - log.info("redistribute called res is " + res); - return res; } Modified: trunk/src/main/org/hornetq/core/remoting/Channel.java =================================================================== --- trunk/src/main/org/hornetq/core/remoting/Channel.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/core/remoting/Channel.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -60,4 +60,6 @@ void flushConfirmations(); void handlePacket(Packet packet); + + void clearCommands(); } Modified: trunk/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/core/remoting/impl/ChannelImpl.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -75,7 +75,7 @@ private final int confWindowSize; - private final Semaphore sendSemaphore; + private volatile Semaphore sendSemaphore; private int receivedBytes; @@ -420,7 +420,7 @@ if (receivedBytes != 0) { receivedBytes = 0; - + final Packet confirmed = new PacketsConfirmedMessage(lastReceivedCommandID); confirmed.setChannelID(id); @@ -441,6 +441,7 @@ { receivedBytes = 0; + final Packet confirmed = new PacketsConfirmedMessage(lastReceivedCommandID); confirmed.setChannelID(id); @@ -449,6 +450,28 @@ } } } + + public void clearCommands() + { + lastReceivedCommandID = -1; + + firstStoredCommandID = 0; + + resendCache.clear(); + + Semaphore oldSemaphore = sendSemaphore; + + if (oldSemaphore != null) + { + //Reset the semaphore + sendSemaphore = new Semaphore(windowSize, true); + + //Any threads blocking on the send semaphore should be allowed to return - we do this by just giving it + //a lot of permits - note we don't give it Integer.MAX_VALUE since then if if more releases come in that + //could end up with permit count going -ve which would cause subsequent sends to block + oldSemaphore.release(Integer.MAX_VALUE / 2); + } + } public void handlePacket(final Packet packet) { @@ -520,11 +543,12 @@ if (packet == null) { - throw new IllegalStateException(System.identityHashCode(this) + " Can't find packet to clear: " + + log.warn("Can't find packet to clear: " + " last received command id " + lastReceivedCommandID + " first stored command id " + firstStoredCommandID); + return; } if (packet.getType() != PACKETS_CONFIRMED) Modified: trunk/src/main/org/hornetq/core/remoting/impl/wireformat/CreateSessionResponseMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/remoting/impl/wireformat/CreateSessionResponseMessage.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/core/remoting/impl/wireformat/CreateSessionResponseMessage.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -30,18 +30,14 @@ private int serverVersion; - private boolean created; - // Static -------------------------------------------------------- // Constructors -------------------------------------------------- - public CreateSessionResponseMessage(final boolean created, final int serverVersion) + public CreateSessionResponseMessage(final int serverVersion) { super(CREATESESSION_RESP); - this.created = created; - this.serverVersion = serverVersion; } @@ -52,11 +48,6 @@ // Public -------------------------------------------------------- - public boolean isCreated() - { - return created; - } - @Override public boolean isResponse() { @@ -70,21 +61,19 @@ @Override public void encodeBody(final HornetQBuffer buffer) - { - buffer.writeBoolean(created); + { buffer.writeInt(serverVersion); } @Override public void decodeBody(final HornetQBuffer buffer) - { - created = buffer.readBoolean(); + { serverVersion = buffer.readInt(); } public int getRequiredBufferSize() { - return BASIC_PACKET_SIZE + DataConstants.SIZE_BOOLEAN + DataConstants.SIZE_INT; + return BASIC_PACKET_SIZE + DataConstants.SIZE_INT; } @Override @@ -97,7 +86,7 @@ CreateSessionResponseMessage r = (CreateSessionResponseMessage)other; - boolean matches = super.equals(other) && serverVersion == r.serverVersion && created == r.created; + boolean matches = super.equals(other) && serverVersion == r.serverVersion; return matches; } Modified: trunk/src/main/org/hornetq/core/remoting/impl/wireformat/ReattachSessionResponseMessage.java =================================================================== --- trunk/src/main/org/hornetq/core/remoting/impl/wireformat/ReattachSessionResponseMessage.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/core/remoting/impl/wireformat/ReattachSessionResponseMessage.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -31,19 +31,19 @@ private int lastReceivedCommandID; - private boolean sessionFound; + private boolean reattached; // Static -------------------------------------------------------- // Constructors -------------------------------------------------- - public ReattachSessionResponseMessage(final int lastReceivedCommandID, final boolean sessionFound) + public ReattachSessionResponseMessage(final int lastReceivedCommandID, final boolean reattached) { super(REATTACH_SESSION_RESP); this.lastReceivedCommandID = lastReceivedCommandID; - this.sessionFound = sessionFound; + this.reattached = reattached; } public ReattachSessionResponseMessage() @@ -58,9 +58,9 @@ return lastReceivedCommandID; } - public boolean isSessionFound() + public boolean isReattached() { - return sessionFound; + return reattached; } public int getRequiredBufferSize() @@ -68,17 +68,16 @@ return BASIC_PACKET_SIZE + DataConstants.SIZE_INT + DataConstants.SIZE_BOOLEAN; } - public void encodeBody(final HornetQBuffer buffer) { buffer.writeInt(lastReceivedCommandID); - buffer.writeBoolean(sessionFound); + buffer.writeBoolean(reattached); } public void decodeBody(final HornetQBuffer buffer) { lastReceivedCommandID = buffer.readInt(); - sessionFound = buffer.readBoolean(); + reattached = buffer.readBoolean(); } public boolean isResponse() Modified: trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -536,10 +536,13 @@ { // Backup server is not ready to accept connections - return new CreateSessionResponseMessage(false, version.getIncrementingVersion()); + return new CreateSessionResponseMessage(version.getIncrementingVersion()); } - securityStore.authenticate(username, password); + if (securityStore != null) + { + securityStore.authenticate(username, password); + } ServerSession currentSession = sessions.remove(name); @@ -581,7 +584,7 @@ channel.setHandler(handler); - return new CreateSessionResponseMessage(true, version.getIncrementingVersion()); + return new CreateSessionResponseMessage(version.getIncrementingVersion()); } public void removeSession(final String name) throws Exception Modified: trunk/src/main/org/hornetq/jms/client/HornetQConnection.java =================================================================== --- trunk/src/main/org/hornetq/jms/client/HornetQConnection.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/jms/client/HornetQConnection.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -11,7 +11,6 @@ * permissions and limitations under the License. */ - package org.hornetq.jms.client; import java.lang.ref.WeakReference; @@ -244,8 +243,10 @@ public synchronized void close() throws JMSException { + log.info("Closing jms connection"); if (closed) { + log.info("Already closed"); return; } @@ -279,6 +280,7 @@ { if (initialSession != null) { + log.info("closing initial session"); initialSession.close(); } } @@ -541,7 +543,7 @@ try { initialSession = sessionFactory.createSession(username, password, false, false, false, false, 0); - + initialSession.addFailureListener(listener); } catch (HornetQException me) @@ -567,7 +569,7 @@ { return; } - + HornetQConnection conn = connectionRef.get(); if (conn != null) Modified: trunk/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java =================================================================== --- trunk/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -463,16 +463,6 @@ sessionFactory.setReconnectAttempts(reconnectAttempts); } - public synchronized boolean isUseReattach() - { - return sessionFactory.isUseReattach(); - } - - public synchronized void setUseReattach(boolean reattach) - { - sessionFactory.setUseReattach(reattach); - } - public synchronized boolean isFailoverOnServerShutdown() { return sessionFactory.isFailoverOnServerShutdown(); Modified: trunk/src/main/org/hornetq/jms/client/JMSExceptionHelper.java =================================================================== --- trunk/src/main/org/hornetq/jms/client/JMSExceptionHelper.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/jms/client/JMSExceptionHelper.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -78,6 +78,10 @@ je = new javax.jms.IllegalStateException(me.getMessage()); break; + case HornetQException.TRANSACTION_ROLLED_BACK: + je = new javax.jms.TransactionRolledBackException(me.getMessage()); + break; + default: je = new JMSException(me.getMessage()); } Modified: trunk/src/main/org/hornetq/jms/server/JMSServerManager.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/JMSServerManager.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/jms/server/JMSServerManager.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -158,7 +158,7 @@ int dupsOKBatchSize, boolean useGlobalPools, int scheduledThreadPoolMaxSize, - int threadPoolMaxSize, + int threadPoolMaxSize, long retryInterval, double retryIntervalMultiplier, long maxRetryInterval, @@ -191,7 +191,7 @@ long initialWaitTimeout, boolean useGlobalPools, int scheduledThreadPoolMaxSize, - int threadPoolMaxSize, + int threadPoolMaxSize, long retryInterval, double retryIntervalMultiplier, long maxRetryInterval, Modified: trunk/src/main/org/hornetq/jms/server/config/ConnectionFactoryConfiguration.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/config/ConnectionFactoryConfiguration.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/jms/server/config/ConnectionFactoryConfiguration.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -134,7 +134,7 @@ int getThreadPoolMaxSize(); void setThreadPoolMaxSize(int threadPoolMaxSize); - + long getRetryInterval(); void setRetryInterval(long retryInterval); Modified: trunk/src/main/org/hornetq/jms/server/config/impl/ConnectionFactoryConfigurationImpl.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/config/impl/ConnectionFactoryConfigurationImpl.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/jms/server/config/impl/ConnectionFactoryConfigurationImpl.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -89,7 +89,7 @@ public int scheduledThreadPoolMaxSize = ClientSessionFactoryImpl.DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE; public int threadPoolMaxSize = ClientSessionFactoryImpl.DEFAULT_THREAD_POOL_MAX_SIZE; - + public long retryInterval = ClientSessionFactoryImpl.DEFAULT_RETRY_INTERVAL; public double retryIntervalMultiplier = ClientSessionFactoryImpl.DEFAULT_RETRY_INTERVAL_MULTIPLIER; Modified: trunk/src/main/org/hornetq/jms/server/impl/JMSServerDeployer.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/impl/JMSServerDeployer.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/jms/server/impl/JMSServerDeployer.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -253,7 +253,7 @@ discoveryInitialWaitTimeout, useGlobalPools, scheduledThreadPoolMaxSize, - threadPoolMaxSize, + threadPoolMaxSize, retryInterval, retryIntervalMultiplier, maxRetryInterval, @@ -285,7 +285,7 @@ dupsOKBatchSize, useGlobalPools, scheduledThreadPoolMaxSize, - threadPoolMaxSize, + threadPoolMaxSize, retryInterval, retryIntervalMultiplier, maxRetryInterval, Modified: trunk/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -382,7 +382,7 @@ int dupsOKBatchSize, boolean useGlobalPools, int scheduledThreadPoolMaxSize, - int threadPoolMaxSize, + int threadPoolMaxSize, long retryInterval, double retryIntervalMultiplier, long maxRetryInterval, @@ -451,7 +451,7 @@ long initialWaitTimeout, boolean useGlobalPools, int scheduledThreadPoolMaxSize, - int threadPoolMaxSize, + int threadPoolMaxSize, long retryInterval, double retryIntervalMultiplier, long maxRetryInterval, @@ -486,7 +486,7 @@ cf.setDiscoveryInitialWaitTimeout(initialWaitTimeout); cf.setUseGlobalPools(useGlobalPools); cf.setScheduledThreadPoolMaxSize(scheduledThreadPoolMaxSize); - cf.setThreadPoolMaxSize(threadPoolMaxSize); + cf.setThreadPoolMaxSize(threadPoolMaxSize); cf.setRetryInterval(retryInterval); cf.setRetryIntervalMultiplier(retryIntervalMultiplier); cf.setMaxRetryInterval(maxRetryInterval); @@ -766,7 +766,7 @@ config.getInitialWaitTimeout(), config.isUseGlobalPools(), config.getScheduledThreadPoolMaxSize(), - config.getThreadPoolMaxSize(), + config.getThreadPoolMaxSize(), config.getRetryInterval(), config.getRetryIntervalMultiplier(), config.getMaxRetryInterval(), @@ -798,7 +798,7 @@ config.getDupsOKBatchSize(), config.isUseGlobalPools(), config.getScheduledThreadPoolMaxSize(), - config.getThreadPoolMaxSize(), + config.getThreadPoolMaxSize(), config.getRetryInterval(), config.getRetryIntervalMultiplier(), config.getMaxRetryInterval(), Modified: trunk/src/main/org/hornetq/jms/server/management/JMSServerControl.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/management/JMSServerControl.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/jms/server/management/JMSServerControl.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -115,7 +115,7 @@ int dupsOKBatchSize, boolean useGlobalPools, int scheduledThreadPoolMaxSize, - int threadPoolMaxSize, + int threadPoolMaxSize, long retryInterval, double retryIntervalMultiplier, long maxRetryInterval, @@ -148,7 +148,7 @@ @Parameter(name = "dupsOKBatchSize") int dupsOKBatchSize, @Parameter(name = "useGlobalPools") boolean useGlobalPools, @Parameter(name = "scheduledThreadPoolMaxSize") int scheduledThreadPoolMaxSize, - @Parameter(name = "threadPoolMaxSize") int threadPoolMaxSize, + @Parameter(name = "threadPoolMaxSize") int threadPoolMaxSize, @Parameter(name = "retryInterval") long retryInterval, @Parameter(name = "retryIntervalMultiplier") double retryIntervalMultiplier, @Parameter(name = "maxRetryInterval") long maxRetryInterval, @@ -194,7 +194,7 @@ long initialWaitTimeout, boolean useGlobalPools, int scheduledThreadPoolMaxSize, - int threadPoolMaxSize, + int threadPoolMaxSize, long retryInterval, double retryIntervalMultiplier, long maxRetryInterval, @@ -228,7 +228,7 @@ @Parameter(name = "initialWaitTimeout") long initialWaitTimeout, @Parameter(name = "useGlobalPools") boolean useGlobalPools, @Parameter(name = "scheduledThreadPoolMaxSize") int scheduledThreadPoolMaxSize, - @Parameter(name = "threadPoolMaxSize") int threadPoolMaxSize, + @Parameter(name = "threadPoolMaxSize") int threadPoolMaxSize, @Parameter(name = "retryInterval") long retryInterval, @Parameter(name = "retryIntervalMultiplier") double retryIntervalMultiplier, @Parameter(name = "maxRetryInterval") long maxRetryInterval, Modified: trunk/src/main/org/hornetq/jms/server/management/impl/JMSServerControlImpl.java =================================================================== --- trunk/src/main/org/hornetq/jms/server/management/impl/JMSServerControlImpl.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/src/main/org/hornetq/jms/server/management/impl/JMSServerControlImpl.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -242,7 +242,7 @@ final int dupsOKBatchSize, final boolean useGlobalPools, final int scheduledThreadPoolMaxSize, - final int threadPoolMaxSize, + final int threadPoolMaxSize, final long retryInterval, final double retryIntervalMultiplier, final long maxRetryInterval, @@ -279,7 +279,7 @@ dupsOKBatchSize, useGlobalPools, scheduledThreadPoolMaxSize, - threadPoolMaxSize, + threadPoolMaxSize, retryInterval, retryIntervalMultiplier, maxRetryInterval, @@ -315,7 +315,7 @@ final int dupsOKBatchSize, final boolean useGlobalPools, final int scheduledThreadPoolMaxSize, - final int threadPoolMaxSize, + final int threadPoolMaxSize, final long retryInterval, final double retryIntervalMultiplier, final long maxRetryInterval, @@ -354,7 +354,7 @@ dupsOKBatchSize, useGlobalPools, scheduledThreadPoolMaxSize, - threadPoolMaxSize, + threadPoolMaxSize, retryInterval, retryIntervalMultiplier, maxRetryInterval, @@ -412,7 +412,7 @@ final long initialWaitTimeout, final boolean useGlobalPools, final int scheduledThreadPoolMaxSize, - final int threadPoolMaxSize, + final int threadPoolMaxSize, final long retryInterval, final double retryIntervalMultiplier, final long maxRetryInterval, @@ -447,7 +447,7 @@ initialWaitTimeout, useGlobalPools, scheduledThreadPoolMaxSize, - threadPoolMaxSize, + threadPoolMaxSize, retryInterval, retryIntervalMultiplier, maxRetryInterval, @@ -483,7 +483,7 @@ final long initialWaitTimeout, final boolean useGlobalPools, final int scheduledThreadPoolMaxSize, - final int threadPoolMaxSize, + final int threadPoolMaxSize, final long retryInterval, final double retryIntervalMultiplier, final long maxRetryInterval, @@ -518,7 +518,7 @@ initialWaitTimeout, useGlobalPools, scheduledThreadPoolMaxSize, - threadPoolMaxSize, + threadPoolMaxSize, retryInterval, retryIntervalMultiplier, maxRetryInterval, Modified: trunk/tests/jms-tests/src/org/hornetq/jms/tests/CTSMiscellaneousTest.java =================================================================== --- trunk/tests/jms-tests/src/org/hornetq/jms/tests/CTSMiscellaneousTest.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/jms-tests/src/org/hornetq/jms/tests/CTSMiscellaneousTest.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -108,7 +108,7 @@ DEFAULT_ACK_BATCH_SIZE, DEFAULT_USE_GLOBAL_POOLS, DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE, - DEFAULT_THREAD_POOL_MAX_SIZE, + DEFAULT_THREAD_POOL_MAX_SIZE, DEFAULT_RETRY_INTERVAL, DEFAULT_RETRY_INTERVAL_MULTIPLIER, DEFAULT_MAX_RETRY_INTERVAL, Modified: trunk/tests/jms-tests/src/org/hornetq/jms/tests/JMSTestCase.java =================================================================== --- trunk/tests/jms-tests/src/org/hornetq/jms/tests/JMSTestCase.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/jms-tests/src/org/hornetq/jms/tests/JMSTestCase.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -104,7 +104,7 @@ DEFAULT_ACK_BATCH_SIZE, DEFAULT_USE_GLOBAL_POOLS, DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE, - DEFAULT_THREAD_POOL_MAX_SIZE, + DEFAULT_THREAD_POOL_MAX_SIZE, DEFAULT_RETRY_INTERVAL, DEFAULT_RETRY_INTERVAL_MULTIPLIER, DEFAULT_MAX_RETRY_INTERVAL, Modified: trunk/tests/jms-tests/src/org/hornetq/jms/tests/tools/container/LocalTestServer.java =================================================================== --- trunk/tests/jms-tests/src/org/hornetq/jms/tests/tools/container/LocalTestServer.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/jms-tests/src/org/hornetq/jms/tests/tools/container/LocalTestServer.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -333,7 +333,7 @@ dupsOkBatchSize, DEFAULT_USE_GLOBAL_POOLS, DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE, - DEFAULT_THREAD_POOL_MAX_SIZE, + DEFAULT_THREAD_POOL_MAX_SIZE, DEFAULT_RETRY_INTERVAL, DEFAULT_RETRY_INTERVAL_MULTIPLIER, DEFAULT_MAX_RETRY_INTERVAL, Modified: trunk/tests/src/org/hornetq/tests/integration/client/ConsumerTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/ConsumerTest.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/src/org/hornetq/tests/integration/client/ConsumerTest.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -286,6 +286,8 @@ consumer.setMessageHandler(null); consumer.receiveImmediate(); + + session.close(); } public void testNoReceiveWithListener() throws Exception @@ -321,6 +323,8 @@ fail("Wrong exception code"); } } + + session.close(); } // https://jira.jboss.org/jira/browse/HORNETQ-111 Modified: trunk/tests/src/org/hornetq/tests/integration/client/HeuristicXATest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/HeuristicXATest.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/src/org/hornetq/tests/integration/client/HeuristicXATest.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -404,6 +404,8 @@ assertEquals(1, recoveredXids.length); assertEquals(xid, recoveredXids[0]); assertEquals(0, session.recover(XAResource.TMENDRSCAN).length); + + session.close(); } finally { @@ -504,7 +506,8 @@ { assertEquals(0, jmxServer.listHeuristicRolledBackTransactions().length); } - + + session.close(); } finally { Modified: trunk/tests/src/org/hornetq/tests/integration/client/HornetQCrashTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/client/HornetQCrashTest.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/src/org/hornetq/tests/integration/client/HornetQCrashTest.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -75,6 +75,8 @@ Thread.sleep(250); assertFalse(ackReceived); + + session.close(); } public static class AckInterceptor implements Interceptor Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -963,6 +963,7 @@ { public void connectionFailed(HornetQException me) { + log.info("calling listener"); latch.countDown(); } } @@ -1019,6 +1020,8 @@ // Wait to be informed of failure boolean ok = latch.await(1000, TimeUnit.MILLISECONDS); + + log.info("waited for latch"); assertTrue(ok); @@ -1032,6 +1035,8 @@ { assertEquals(XAException.XA_RBOTHER, e.errorCode); } + + //Thread.sleep(30000); session1.close(); Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadRandomReattachTestBase.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadRandomReattachTestBase.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/MultiThreadRandomReattachTestBase.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -1323,8 +1323,7 @@ { final ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); sf.setReconnectAttempts(-1); - sf.setUseReattach(true); - + return sf; } Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/NettyMultiThreadRandomReattachTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/NettyMultiThreadRandomReattachTest.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/NettyMultiThreadRandomReattachTest.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -48,7 +48,6 @@ protected ClientSessionFactoryInternal createSessionFactory() { final ClientSessionFactoryInternal sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.hornetq.integration.transports.netty.NettyConnectorFactory")); - sf.setUseReattach(true); sf.setReconnectAttempts(-1); return sf; } Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/RandomReattachTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/RandomReattachTest.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/RandomReattachTest.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -222,8 +222,6 @@ sf.setReconnectAttempts(-1); - sf.setUseReattach(true); - ClientSession session = sf.createSession(false, false, false); Failer failer = startFailer(1000, session); Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/ReattachTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/ReattachTest.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/reattach/ReattachTest.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -79,7 +79,6 @@ sf.setRetryInterval(retryInterval); sf.setRetryIntervalMultiplier(retryMultiplier); sf.setReconnectAttempts(reconnectAttempts); - sf.setUseReattach(true); ClientSession session = sf.createSession(false, true, true); @@ -157,7 +156,6 @@ sf.setRetryInterval(retryInterval); sf.setRetryIntervalMultiplier(retryMultiplier); sf.setReconnectAttempts(reconnectAttempts); - sf.setUseReattach(true); ClientSession session = sf.createSession(false, true, true); @@ -247,7 +245,6 @@ sf.setRetryInterval(retryInterval); sf.setRetryIntervalMultiplier(retryMultiplier); sf.setReconnectAttempts(reconnectAttempts); - sf.setUseReattach(true); ClientSession session = sf.createSession(false, true, true); @@ -357,7 +354,6 @@ sf.setRetryInterval(retryInterval); sf.setRetryIntervalMultiplier(retryMultiplier); sf.setReconnectAttempts(reconnectAttempts); - sf.setUseReattach(true); ClientSession session = sf.createSession(false, true, true); @@ -434,7 +430,6 @@ sf.setRetryInterval(retryInterval); sf.setRetryIntervalMultiplier(retryMultiplier); sf.setReconnectAttempts(reconnectAttempts); - sf.setUseReattach(true); ClientSession session = sf.createSession(false, true, true); @@ -502,7 +497,6 @@ sf.setRetryInterval(retryInterval); sf.setRetryIntervalMultiplier(retryMultiplier); sf.setReconnectAttempts(reconnectAttempts); - sf.setUseReattach(true); ClientSession session = sf.createSession(false, true, true); @@ -594,7 +588,6 @@ sf.setRetryInterval(retryInterval); sf.setRetryIntervalMultiplier(retryMultiplier); sf.setReconnectAttempts(reconnectAttempts); - sf.setUseReattach(true); ClientSession session = sf.createSession(false, true, true); @@ -673,7 +666,6 @@ sf.setRetryIntervalMultiplier(retryMultiplier); sf.setReconnectAttempts(reconnectAttempts); sf.setMaxRetryInterval(maxRetryInterval); - sf.setUseReattach(true); ClientSession session = sf.createSession(false, true, true); Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/restart/ClusterRestartTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/cluster/restart/ClusterRestartTest.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/src/org/hornetq/tests/integration/cluster/restart/ClusterRestartTest.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -96,7 +96,7 @@ } finally { - //closeAllConsumers(); + closeAllConsumers(); closeAllSessionFactories(); @@ -172,7 +172,7 @@ } finally { - //closeAllConsumers(); + closeAllConsumers(); closeAllSessionFactories(); Modified: trunk/tests/src/org/hornetq/tests/integration/jms/FloodServerTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/FloodServerTest.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/src/org/hornetq/tests/integration/jms/FloodServerTest.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -166,7 +166,7 @@ DEFAULT_ACK_BATCH_SIZE, DEFAULT_USE_GLOBAL_POOLS, DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE, - DEFAULT_THREAD_POOL_MAX_SIZE, + DEFAULT_THREAD_POOL_MAX_SIZE, retryInterval, retryIntervalMultiplier, 1000, Modified: trunk/tests/src/org/hornetq/tests/integration/jms/ManualReconnectionToSingleServerTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/ManualReconnectionToSingleServerTest.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/src/org/hornetq/tests/integration/jms/ManualReconnectionToSingleServerTest.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -177,6 +177,13 @@ //Make sure it doesn't pass by just timing out on blocking send assertTrue(end - start < callTimeout); + + System.gc(); + System.gc(); + System.gc(); + System.gc(); + + Thread.sleep(30000); } @@ -276,7 +283,7 @@ DEFAULT_ACK_BATCH_SIZE, DEFAULT_USE_GLOBAL_POOLS, DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE, - DEFAULT_THREAD_POOL_MAX_SIZE, + DEFAULT_THREAD_POOL_MAX_SIZE, retryInterval, retryIntervalMultiplier, 1000, @@ -287,19 +294,23 @@ protected void disconnect() { + log.info("calling disocnnect"); if (connection == null) { + log.info("connection is null"); return; } try { connection.setExceptionListener(null); + log.info("closing the connection"); connection.close(); connection = null; } catch (Exception e) { + log.info("** got exception"); e.printStackTrace(); } } Modified: trunk/tests/src/org/hornetq/tests/integration/jms/client/PreACKJMSTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/client/PreACKJMSTest.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/src/org/hornetq/tests/integration/jms/client/PreACKJMSTest.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -240,7 +240,7 @@ DEFAULT_ACK_BATCH_SIZE, DEFAULT_USE_GLOBAL_POOLS, DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE, - DEFAULT_THREAD_POOL_MAX_SIZE, + DEFAULT_THREAD_POOL_MAX_SIZE, retryInterval, retryIntervalMultiplier, DEFAULT_MAX_RETRY_INTERVAL, Modified: trunk/tests/src/org/hornetq/tests/integration/jms/client/ResendTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/client/ResendTest.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/src/org/hornetq/tests/integration/jms/client/ResendTest.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -280,7 +280,7 @@ DEFAULT_ACK_BATCH_SIZE, DEFAULT_USE_GLOBAL_POOLS, DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE, - DEFAULT_THREAD_POOL_MAX_SIZE, + DEFAULT_THREAD_POOL_MAX_SIZE, retryInterval, retryIntervalMultiplier, DEFAULT_MAX_RETRY_INTERVAL, Modified: trunk/tests/src/org/hornetq/tests/integration/jms/connection/CloseConnectionOnGCTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/connection/CloseConnectionOnGCTest.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/src/org/hornetq/tests/integration/jms/connection/CloseConnectionOnGCTest.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -17,8 +17,8 @@ import javax.jms.Connection; import javax.jms.Session; -import org.hornetq.core.client.impl.ClientSessionFactoryImpl; import org.hornetq.core.config.TransportConfiguration; +import org.hornetq.core.logging.Logger; import org.hornetq.jms.client.HornetQConnectionFactory; import org.hornetq.tests.util.JMSTestBase; @@ -32,6 +32,8 @@ */ public class CloseConnectionOnGCTest extends JMSTestBase { + private static final Logger log = Logger.getLogger(CloseConnectionOnGCTest.class); + private HornetQConnectionFactory cf; @Override @@ -52,9 +54,24 @@ super.tearDown(); } + public void testFoo() throws Exception + { + for (int i = 0; i < 100; i++) + { + log.info("Iteration " + i); + + testCloseOneConnectionOnGC(); + } + } public void testCloseOneConnectionOnGC() throws Exception { + //Debug - don't remove this until intermittent failure with this test is fixed + int initialConns = server.getRemotingService().getConnections().size(); + + assertEquals(0, initialConns); + + Connection conn = cf.createConnection(); WeakReference<Connection> wr = new WeakReference<Connection>(conn); Modified: trunk/tests/src/org/hornetq/tests/integration/jms/divert/DivertAndACKClientTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/divert/DivertAndACKClientTest.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/src/org/hornetq/tests/integration/jms/divert/DivertAndACKClientTest.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -179,7 +179,7 @@ DEFAULT_ACK_BATCH_SIZE, DEFAULT_USE_GLOBAL_POOLS, DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE, - DEFAULT_THREAD_POOL_MAX_SIZE, + DEFAULT_THREAD_POOL_MAX_SIZE, retryInterval, retryIntervalMultiplier, DEFAULT_MAX_RETRY_INTERVAL, Modified: trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlTest.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlTest.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -59,8 +59,9 @@ // Attributes ---------------------------------------------------- protected InVMContext context; + private HornetQServer server; - + private JMSServerManagerImpl serverManager; // Static -------------------------------------------------------- @@ -128,7 +129,7 @@ checkNoBinding(context, queueJNDIBinding); checkNoResource(ObjectNameBuilder.DEFAULT.getJMSQueueObjectName(queueName)); } - + public void testGetQueueNames() throws Exception { String queueJNDIBinding = randomString(); @@ -185,7 +186,7 @@ checkNoBinding(context, topicJNDIBinding); checkNoResource(ObjectNameBuilder.DEFAULT.getJMSTopicObjectName(topicName)); } - + public void testGetTopicNames() throws Exception { String topicJNDIBinding = randomString(); @@ -320,13 +321,9 @@ 0, TransportConstants.SERVER_ID_PROP_NAME, 1); - - control.createConnectionFactory(cfName, - InVMConnectorFactory.class.getName() + ", " + InVMConnectorFactory.class.getName(), - params, - "", - "", - jndiBindings); + + control.createConnectionFactory(cfName, InVMConnectorFactory.class.getName() + ", " + + InVMConnectorFactory.class.getName(), params, "", "", jndiBindings); } }); } @@ -352,7 +349,7 @@ }); } - // with 1 live and 1 backup + // with 1 live and 1 backup public void testCreateConnectionFactory_4b() throws Exception { doCreateConnectionFactory(new ConnectionFactoryCreator() @@ -431,7 +428,7 @@ clientID, ClientSessionFactoryImpl.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, ClientSessionFactoryImpl.DEFAULT_CONNECTION_TTL, - ClientSessionFactoryImpl.DEFAULT_CALL_TIMEOUT, + ClientSessionFactoryImpl.DEFAULT_CALL_TIMEOUT, ClientSessionFactoryImpl.DEFAULT_CACHE_LARGE_MESSAGE_CLIENT, ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_WINDOW_SIZE, @@ -458,7 +455,7 @@ } }); } - + public void testCreateConnectionFactory_7b() throws Exception { doCreateConnectionFactory(new ConnectionFactoryCreator() @@ -476,7 +473,7 @@ clientID, ClientSessionFactoryImpl.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, ClientSessionFactoryImpl.DEFAULT_CONNECTION_TTL, - ClientSessionFactoryImpl.DEFAULT_CALL_TIMEOUT, + ClientSessionFactoryImpl.DEFAULT_CALL_TIMEOUT, ClientSessionFactoryImpl.DEFAULT_CACHE_LARGE_MESSAGE_CLIENT, ClientSessionFactoryImpl.DEFAULT_MIN_LARGE_MESSAGE_SIZE, ClientSessionFactoryImpl.DEFAULT_CONSUMER_WINDOW_SIZE, @@ -566,7 +563,7 @@ } } - + public void testGetConnectionFactoryNames() throws Exception { String cfBinding = randomString(); @@ -574,14 +571,17 @@ JMSServerControl control = createManagementControl(); assertEquals(0, control.getConnectionFactoryNames().length); - + TransportConfiguration tcLive = new TransportConfiguration(InVMConnectorFactory.class.getName()); - control.createConnectionFactory(cfName, tcLive.getFactoryClassName(), tcLive.getParams(), new String[] {cfBinding}); + control.createConnectionFactory(cfName, + tcLive.getFactoryClassName(), + tcLive.getParams(), + new String[] { cfBinding }); String[] cfNames = control.getConnectionFactoryNames(); assertEquals(1, cfNames.length); assertEquals(cfName, cfNames[0]); - + control.destroyConnectionFactory(cfName); assertEquals(0, control.getConnectionFactoryNames().length); } @@ -612,11 +612,11 @@ protected void tearDown() throws Exception { serverManager.stop(); - + server.stop(); - + serverManager = null; - + server = null; super.tearDown(); Modified: trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlUsingJMSTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlUsingJMSTest.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlUsingJMSTest.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -117,7 +117,7 @@ long initialWaitTimeout, boolean useGlobalPools, int scheduledThreadPoolMaxSize, - int threadPoolMaxSize, + int threadPoolMaxSize, long retryInterval, double retryIntervalMultiplier, long maxRetryInterval, @@ -150,7 +150,7 @@ initialWaitTimeout, useGlobalPools, scheduledThreadPoolMaxSize, - threadPoolMaxSize, + threadPoolMaxSize, retryInterval, retryIntervalMultiplier, maxRetryInterval, @@ -184,7 +184,7 @@ long initialWaitTimeout, boolean useGlobalPools, int scheduledThreadPoolMaxSize, - int threadPoolMaxSize, + int threadPoolMaxSize, long retryInterval, double retryIntervalMultiplier, long maxRetryInterval, @@ -217,7 +217,7 @@ initialWaitTimeout, useGlobalPools, scheduledThreadPoolMaxSize, - threadPoolMaxSize, + threadPoolMaxSize, retryInterval, retryIntervalMultiplier, maxRetryInterval, @@ -399,7 +399,7 @@ int dupsOKBatchSize, boolean useGlobalPools, int scheduledThreadPoolMaxSize, - int threadPoolMaxSize, + int threadPoolMaxSize, long retryInterval, double retryIntervalMultiplier, long maxRetryInterval, @@ -432,7 +432,7 @@ transactionBatchSize, dupsOKBatchSize, useGlobalPools, - scheduledThreadPoolMaxSize, + scheduledThreadPoolMaxSize, threadPoolMaxSize, retryInterval, retryIntervalMultiplier, @@ -468,7 +468,7 @@ int dupsOKBatchSize, boolean useGlobalPools, int scheduledThreadPoolMaxSize, - int threadPoolMaxSize, + int threadPoolMaxSize, long retryInterval, double retryIntervalMultiplier, long maxRetryInterval, @@ -502,7 +502,7 @@ dupsOKBatchSize, useGlobalPools, scheduledThreadPoolMaxSize, - threadPoolMaxSize, + threadPoolMaxSize, retryInterval, retryIntervalMultiplier, maxRetryInterval, Modified: trunk/tests/src/org/hornetq/tests/integration/server/LVQTest.java =================================================================== --- trunk/tests/src/org/hornetq/tests/integration/server/LVQTest.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/src/org/hornetq/tests/integration/server/LVQTest.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -526,6 +526,30 @@ // } } + + if (clientSessionTxReceives != null) + { + try + { + clientSessionTxReceives.close(); + } + catch (HornetQException e1) + { + // + } + } + + if (clientSessionTxSends != null) + { + try + { + clientSessionTxSends.close(); + } + catch (HornetQException e1) + { + // + } + } if (server != null && server.isStarted()) { try Modified: trunk/tests/src/org/hornetq/tests/util/JMSTestBase.java =================================================================== --- trunk/tests/src/org/hornetq/tests/util/JMSTestBase.java 2009-10-13 09:02:45 UTC (rev 8082) +++ trunk/tests/src/org/hornetq/tests/util/JMSTestBase.java 2009-10-13 10:06:28 UTC (rev 8083) @@ -219,7 +219,7 @@ DEFAULT_ACK_BATCH_SIZE, DEFAULT_USE_GLOBAL_POOLS, DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE, - DEFAULT_THREAD_POOL_MAX_SIZE, + DEFAULT_THREAD_POOL_MAX_SIZE, retryInterval, retryIntervalMultiplier, DEFAULT_MAX_RETRY_INTERVAL,