JBoss hornetq SVN: r11773 - in branches/Branch_2_2_EAP: tests/src/org/hornetq/tests/integration/jms/server/management and 1 other directory.
by do-not-reply＠jboss.org
Author: gaohoward
Date: 2011-11-28 07:42:57 -0500 (Mon, 28 Nov 2011)
New Revision: 11773
Modified:
branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java
branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/jms/server/management/JMSQueueControlTest.java
Log:
fix test failures on Windows
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-11-28 10:32:30 UTC (rev 11772)
+++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-11-28 12:42:57 UTC (rev 11773)
@@ -317,6 +317,12 @@
public synchronized void start() throws Exception
{
+ if (started)
+ {
+ log.debug("Server already started!");
+ return;
+ }
+
log.debug("Starting server " + this);
OperationContextImpl.clearContext();
Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/jms/server/management/JMSQueueControlTest.java
===================================================================
--- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/jms/server/management/JMSQueueControlTest.java 2011-11-28 10:32:30 UTC (rev 11772)
+++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/integration/jms/server/management/JMSQueueControlTest.java 2011-11-28 12:42:57 UTC (rev 11773)
@@ -1177,7 +1177,6 @@
conf.getAcceptorConfigurations().add(new TransportConfiguration(InVMAcceptorFactory.class.getName()));
conf.setFileDeploymentEnabled(false);
server = HornetQServers.newHornetQServer(conf, mbeanServer, true);
- server.start();
serverManager = new JMSServerManagerImpl(server);
context = new InVMContext();
12 years, 5 months
JBoss hornetq SVN: r11772 - trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover.
by do-not-reply＠jboss.org
Author: borges
Date: 2011-11-28 05:32:30 -0500 (Mon, 28 Nov 2011)
New Revision: 11772
Modified:
trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/ReplicatedDistributionTest.java
Log:
Make test silent: Comment sys.out lines.
Modified: trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/ReplicatedDistributionTest.java
===================================================================
--- trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/ReplicatedDistributionTest.java 2011-11-28 10:32:15 UTC (rev 11771)
+++ trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/ReplicatedDistributionTest.java 2011-11-28 10:32:30 UTC (rev 11772)
@@ -66,7 +66,7 @@
Assert.assertNotNull(msg);
- System.out.println(i + " msg = " + msg);
+ // System.out.println(i + " msg = " + msg);
int received = msg.getIntProperty("key");
@@ -101,7 +101,7 @@
Assert.assertNotNull(msg);
- System.out.println(i + " msg = " + msg);
+ // System.out.println(i + " msg = " + msg);
int received = (Integer)msg.getObjectProperty(new SimpleString("key"));
@@ -131,7 +131,7 @@
Assert.assertNotNull(msg);
- System.out.println(i + " msg = " + msg);
+ // System.out.println(i + " msg = " + msg);
int received = msg.getIntProperty("key");
12 years, 5 months
JBoss hornetq SVN: r11771 - trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover.
by do-not-reply＠jboss.org
Author: borges
Date: 2011-11-28 05:32:15 -0500 (Mon, 28 Nov 2011)
New Revision: 11771
Modified:
trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/FailoverTest.java
Log:
Adjust test for it to make sense in the remote replication case.
Modified: trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/FailoverTest.java
===================================================================
--- trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/FailoverTest.java 2011-11-25 15:01:47 UTC (rev 11770)
+++ trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/FailoverTest.java 2011-11-28 10:32:15 UTC (rev 11771)
@@ -1745,6 +1745,12 @@
// To reload security or other settings that are read during startup
beforeRestart(backupServer);
+ if (!backupServer.getServer().getConfiguration().isSharedStore())
+ {
+ // this test would not make sense in the remote replication use case, without the following
+ backupServer.getServer().getConfiguration().setBackup(false);
+ }
+
backupServer.start();
assertTrue("session failure listener", listener.getLatch().await(5, TimeUnit.SECONDS));
12 years, 5 months
JBoss hornetq SVN: r11770 - trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover.
by do-not-reply＠jboss.org
Author: borges
Date: 2011-11-25 10:01:47 -0500 (Fri, 25 Nov 2011)
New Revision: 11770
Modified:
trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncJournalTest.java
trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncLargeMessageTest.java
Log:
Verify that previous test did clean up.
Modified: trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncJournalTest.java
===================================================================
--- trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncJournalTest.java 2011-11-25 15:01:32 UTC (rev 11769)
+++ trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncJournalTest.java 2011-11-25 15:01:47 UTC (rev 11770)
@@ -86,8 +86,7 @@
finishSyncAndFailover();
JournalImpl backupMsgJournal = getMessageJournalFromServer(backupServer);
- System.out.println("backup journal " + backupMsgJournal);
- System.out.println("live journal " + messageJournal);
+
assertEquals("file sizes must be the same", size, backupMsgJournal.getFileSize());
Set<Long> backupIds = getFileIds(backupMsgJournal);
assertEquals("File IDs must match!", liveIds, backupIds);
Modified: trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncLargeMessageTest.java
===================================================================
--- trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncLargeMessageTest.java 2011-11-25 15:01:32 UTC (rev 11769)
+++ trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncLargeMessageTest.java 2011-11-25 15:01:47 UTC (rev 11770)
@@ -35,9 +35,13 @@
public void testDeleteLargeMessages() throws Exception
{
+ File dir = new File(backupServer.getServer().getConfiguration().getLargeMessagesDirectory());
+ assertEquals("Should not have any large messages... previous test failed to clean up?", 0,
+ getAllMessageFileIds(dir).size());
createProducerSendSomeMessages();
startBackupFinishSyncing();
- File dir = new File(backupServer.getServer().getConfiguration().getLargeMessagesDirectory());
+ // File dir = new
+ // File(backupServer.getServer().getConfiguration().getLargeMessagesDirectory());
receiveMsgsInRange(0, n_msgs / 2);
assertEquals("we really ought to delete these after delivery", n_msgs / 2, getAllMessageFileIds(dir).size());
}
@@ -45,11 +49,15 @@
private Set<Long> getAllMessageFileIds(File dir)
{
Set<Long> idsOnBkp = new HashSet<Long>();
- for (String filename : dir.list())
+ String[] fileList = dir.list();
+ if (fileList != null)
{
- if (filename.endsWith(".msg"))
+ for (String filename : fileList)
{
- idsOnBkp.add(Long.valueOf(filename.split("\\.")[0]));
+ if (filename.endsWith(".msg"))
+ {
+ idsOnBkp.add(Long.valueOf(filename.split("\\.")[0]));
+ }
}
}
return idsOnBkp;
12 years, 5 months
JBoss hornetq SVN: r11769 - in trunk/hornetq-core/src/main/java/org/hornetq/core: server/impl and 1 other directory.
by do-not-reply＠jboss.org
Author: borges
Date: 2011-11-25 10:01:32 -0500 (Fri, 25 Nov 2011)
New Revision: 11769
Modified:
trunk/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java
trunk/hornetq-core/src/main/java/org/hornetq/core/server/impl/HornetQServerImpl.java
Log:
Stop replicationEndPoint before stopping the StorageManager.
Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java
===================================================================
--- trunk/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2011-11-25 15:01:17 UTC (rev 11768)
+++ trunk/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2011-11-25 15:01:32 UTC (rev 11769)
@@ -292,7 +292,6 @@
{
channel.close();
}
- storage.stop();
for (ConcurrentMap<Integer, Page> map : pageIndex.values())
{
Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/server/impl/HornetQServerImpl.java
===================================================================
--- trunk/hornetq-core/src/main/java/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-11-25 15:01:17 UTC (rev 11768)
+++ trunk/hornetq-core/src/main/java/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-11-25 15:01:32 UTC (rev 11769)
@@ -547,6 +547,12 @@
pagingManager.stop();
}
+ if (replicationEndpoint != null)
+ {
+ replicationEndpoint.stop();
+ replicationEndpoint = null;
+ }
+
if (!criticalIOError && storageManager != null)
{
storageManager.stop();
@@ -558,12 +564,6 @@
replicationManager = null;
}
- if (replicationEndpoint != null)
- {
- replicationEndpoint.stop();
- replicationEndpoint = null;
- }
-
if (securityManager != null)
{
securityManager.stop();
12 years, 5 months
JBoss hornetq SVN: r11768 - in trunk/hornetq-core/src/main/java/org/hornetq/core/persistence: impl/nullpm and 1 other directory.
by do-not-reply＠jboss.org
Author: borges
Date: 2011-11-25 10:01:17 -0500 (Fri, 25 Nov 2011)
New Revision: 11768
Modified:
trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/StorageManager.java
trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java
Log:
Delete isReplicated() from StorageManager, as it is only used internally at JournalStorageManager.
Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/StorageManager.java
===================================================================
--- trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/StorageManager.java 2011-11-25 15:01:04 UTC (rev 11767)
+++ trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/StorageManager.java 2011-11-25 15:01:17 UTC (rev 11768)
@@ -87,8 +87,6 @@
void pageWrite(PagedMessage message, int pageNumber);
- boolean isReplicated();
-
void afterCompleteOperations(IOAsyncTask run);
/** Block until the operations are done.
Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java
===================================================================
--- trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2011-11-25 15:01:04 UTC (rev 11767)
+++ trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2011-11-25 15:01:17 UTC (rev 11768)
@@ -317,46 +317,23 @@
{
}
- /* (non-Javadoc)
- * @see org.hornetq.core.persistence.StorageManager#loadInternalOnly()
- */
public JournalLoadInformation[] loadInternalOnly() throws Exception
{
return null;
}
- /* (non-Javadoc)
- * @see org.hornetq.core.persistence.StorageManager#isReplicated()
- */
- public boolean isReplicated()
- {
- return false;
- }
-
- /* (non-Javadoc)
- * @see org.hornetq.core.persistence.StorageManager#completeReplication()
- */
public void completeOperations()
{
}
- /* (non-Javadoc)
- * @see org.hornetq.core.persistence.StorageManager#pageClosed(org.hornetq.utils.SimpleString, int)
- */
public void pageClosed(final SimpleString storeName, final int pageNumber)
{
}
- /* (non-Javadoc)
- * @see org.hornetq.core.persistence.StorageManager#pageDeleted(org.hornetq.utils.SimpleString, int)
- */
public void pageDeleted(final SimpleString storeName, final int pageNumber)
{
}
- /* (non-Javadoc)
- * @see org.hornetq.core.persistence.StorageManager#pageWrite(org.hornetq.core.paging.PagedMessage, int)
- */
public void pageWrite(final PagedMessage message, final int pageNumber)
{
}
@@ -369,9 +346,6 @@
{
}
- /* (non-Javadoc)
- * @see org.hornetq.core.persistence.StorageManager#blockOnReplication(long)
- */
public boolean waitOnOperations(final long timeout) throws Exception
{
return true;
12 years, 5 months
JBoss hornetq SVN: r11767 - in trunk/hornetq-core/src/main/java/org/hornetq/core: replication/impl and 1 other directory.
by do-not-reply＠jboss.org
Author: borges
Date: 2011-11-25 10:01:04 -0500 (Fri, 25 Nov 2011)
New Revision: 11767
Modified:
trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/v11/StompFrameHandlerV11.java
trunk/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicatedJournal.java
Log:
Remove unused imports.
Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/v11/StompFrameHandlerV11.java
===================================================================
--- trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/v11/StompFrameHandlerV11.java 2011-11-25 15:00:50 UTC (rev 11766)
+++ trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/v11/StompFrameHandlerV11.java 2011-11-25 15:01:04 UTC (rev 11767)
@@ -17,10 +17,7 @@
import java.util.concurrent.atomic.AtomicLong;
import org.hornetq.api.core.HornetQBuffer;
-import org.hornetq.api.core.Message;
-import org.hornetq.api.core.SimpleString;
import org.hornetq.core.logging.Logger;
-import org.hornetq.core.message.impl.MessageImpl;
import org.hornetq.core.protocol.stomp.FrameEventListener;
import org.hornetq.core.protocol.stomp.HornetQStompException;
import org.hornetq.core.protocol.stomp.SimpleBytes;
@@ -28,11 +25,7 @@
import org.hornetq.core.protocol.stomp.StompConnection;
import org.hornetq.core.protocol.stomp.StompDecoder;
import org.hornetq.core.protocol.stomp.StompFrame;
-import org.hornetq.core.protocol.stomp.StompSubscription;
-import org.hornetq.core.protocol.stomp.StompUtils;
import org.hornetq.core.protocol.stomp.VersionedStompFrameHandler;
-import org.hornetq.core.server.ServerMessage;
-import org.hornetq.utils.DataConstants;
/**
*
@@ -331,6 +324,7 @@
lastPingTime.set(System.currentTimeMillis());
}
+ @Override
public void run()
{
lastAccepted.set(System.currentTimeMillis());
@@ -434,6 +428,7 @@
}
//all frame except CONNECT are decoded here.
+ @Override
public StompFrame decode(StompDecoder decoder, final HornetQBuffer buffer) throws HornetQStompException
{
int readable = buffer.readableBytes();
Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicatedJournal.java
===================================================================
--- trunk/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicatedJournal.java 2011-11-25 15:00:50 UTC (rev 11766)
+++ trunk/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicatedJournal.java 2011-11-25 15:01:04 UTC (rev 11767)
@@ -27,7 +27,6 @@
import org.hornetq.core.journal.TransactionFailureCallback;
import org.hornetq.core.journal.impl.JournalFile;
import org.hornetq.core.journal.impl.dataformat.ByteArrayEncoding;
-import org.hornetq.core.logging.Logger;
import org.hornetq.core.persistence.OperationContext;
import org.hornetq.core.persistence.impl.journal.JournalStorageManager;
import org.hornetq.core.replication.ReplicationManager;
@@ -42,10 +41,6 @@
public class ReplicatedJournal implements Journal
{
- // Constants -----------------------------------------------------
-
- private static final Logger log = Logger.getLogger(ReplicatedJournal.class);
-
// Attributes ----------------------------------------------------
private static final boolean trace = false;
12 years, 5 months
JBoss hornetq SVN: r11766 - trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover.
by do-not-reply＠jboss.org
Author: borges
Date: 2011-11-25 10:00:50 -0500 (Fri, 25 Nov 2011)
New Revision: 11766
Modified:
trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupAuthenticationTest.java
Log:
Wait for backup to start() and finish sync'ing.
Modified: trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupAuthenticationTest.java
===================================================================
--- trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupAuthenticationTest.java 2011-11-25 15:00:31 UTC (rev 11765)
+++ trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupAuthenticationTest.java 2011-11-25 15:00:50 UTC (rev 11766)
@@ -39,7 +39,7 @@
assertFalse("backup should have stopped", backupServer.isStarted());
backupConfig.setClusterPassword(CLUSTER_PASSWORD);
backupServer.start();
- waitForServer(backupServer.getServer());
+ waitForRemoteBackup(null, 5, true, backupServer.getServer());
}
@Override
12 years, 5 months
JBoss hornetq SVN: r11765 - in trunk/hornetq-core/src/main/java/org/hornetq/core: persistence/impl/journal and 2 other directories.
by do-not-reply＠jboss.org
Author: borges
Date: 2011-11-25 10:00:31 -0500 (Fri, 25 Nov 2011)
New Revision: 11765
Modified:
trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/StorageManager.java
trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java
trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java
trunk/hornetq-core/src/main/java/org/hornetq/core/server/impl/HornetQServerImpl.java
Log:
HORNETQ-720 HORNETQ-768 Reset state at JournalStorageManager in case of errors.
Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/StorageManager.java
===================================================================
--- trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/StorageManager.java 2011-11-25 14:54:32 UTC (rev 11764)
+++ trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/StorageManager.java 2011-11-25 15:00:31 UTC (rev 11765)
@@ -269,4 +269,11 @@
ServerMessage message,
RoutingContext ctx,
RouteContextList listCtx) throws Exception;
+
+ /**
+ * Stops the replication of data from the live to the backup.
+ * <p>
+ * Typical scenario is a broken connection.
+ */
+ void stopReplication();
}
Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java
===================================================================
--- trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2011-11-25 14:54:32 UTC (rev 11764)
+++ trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2011-11-25 15:00:31 UTC (rev 11765)
@@ -194,9 +194,11 @@
}
}
- private Journal messageJournal;
- private Journal bindingsJournal;
- private final SequentialFileFactory largeMessagesFactory;
+ private Journal messageJournal;
+ private Journal bindingsJournal;
+ private final Journal originalMessageJournal;
+ private final Journal originalBindingsJournal;
+ private final SequentialFileFactory largeMessagesFactory;
private volatile boolean started;
@@ -235,18 +237,10 @@
public JournalStorageManager(final Configuration config, final ExecutorFactory executorFactory,
final IOCriticalErrorListener criticalErrorListener)
{
- this(config, executorFactory, null, criticalErrorListener);
- }
-
- public JournalStorageManager(final Configuration config, final ExecutorFactory executorFactory,
- final ReplicationManager replicator, final IOCriticalErrorListener criticalErrorListener)
- {
this.executorFactory = executorFactory;
executor = executorFactory.getExecutor();
- this.replicator = replicator;
-
if (config.getJournalType() != JournalType.NIO && config.getJournalType() != JournalType.ASYNCIO)
{
throw new IllegalArgumentException("Only NIO and AsyncIO are supported journals");
@@ -274,19 +268,13 @@
"bindings",
1);
- if (replicator != null)
- {
- bindingsJournal = new ReplicatedJournal((byte)0, localBindings, replicator);
- }
- else
- {
- bindingsJournal = localBindings;
- }
+ bindingsJournal = localBindings;
+ originalBindingsJournal = localBindings;
- if (journalDir == null)
- {
- throw new NullPointerException("journal-dir is null");
- }
+ if (journalDir == null)
+ {
+ throw new NullPointerException("journal-dir is null");
+ }
createJournalDir = config.isCreateJournalDir();
@@ -332,14 +320,8 @@
config.getJournalType() == JournalType.ASYNCIO ? config.getJournalMaxIO_AIO()
: config.getJournalMaxIO_NIO());
- if (replicator != null)
- {
- messageJournal = new ReplicatedJournal((byte)1, localMessage, replicator);
- }
- else
- {
- messageJournal = localMessage;
- }
+ messageJournal = localMessage;
+ originalMessageJournal = localMessage;
largeMessagesDirectory = config.getLargeMessagesDirectory();
largeMessagesFactory = new NIOSequentialFileFactory(largeMessagesDirectory, false, criticalErrorListener);
@@ -380,8 +362,8 @@
JournalFile[] messageFiles = null;
JournalFile[] bindingsFiles = null;
- final Journal localMessageJournal = messageJournal;
- final Journal localBindingsJournal = bindingsJournal;
+ try
+ {
Map<String, Long> largeMessageFilesToSync;
Map<SimpleString, Collection<Integer>> pageFilesToSync;
@@ -389,15 +371,15 @@
try
{
replicator = replicationManager;
- localMessageJournal.synchronizationLock();
- localBindingsJournal.synchronizationLock();
+ originalMessageJournal.synchronizationLock();
+ originalBindingsJournal.synchronizationLock();
try
{
pagingManager.lock();
try
{
- messageFiles = prepareJournalForCopy(localMessageJournal, JournalContent.MESSAGES, nodeID);
- bindingsFiles = prepareJournalForCopy(localBindingsJournal, JournalContent.BINDINGS, nodeID);
+ messageFiles = prepareJournalForCopy(originalMessageJournal, JournalContent.MESSAGES, nodeID);
+ bindingsFiles = prepareJournalForCopy(originalBindingsJournal, JournalContent.BINDINGS, nodeID);
pageFilesToSync = getPageInformationForSync(pagingManager);
largeMessageFilesToSync = getLargeMessageInformation();
}
@@ -408,11 +390,11 @@
}
finally
{
- localMessageJournal.synchronizationUnlock();
- localBindingsJournal.synchronizationUnlock();
+ originalMessageJournal.synchronizationUnlock();
+ originalBindingsJournal.synchronizationUnlock();
}
- bindingsJournal = new ReplicatedJournal(((byte)0), localBindingsJournal, replicator);
- messageJournal = new ReplicatedJournal((byte)1, localMessageJournal, replicator);
+ bindingsJournal = new ReplicatedJournal(((byte)0), originalBindingsJournal, replicator);
+ messageJournal = new ReplicatedJournal((byte)1, originalMessageJournal, replicator);
}
finally
{
@@ -433,11 +415,36 @@
}
finally
{
+ storageManagerLock.writeLock().unlock();
+ }
+ }
+ catch (Exception e)
+ {
+ stopReplication();
+ throw e;
+ }
+ }
+
+ /**
+ * Stops replication by resetting replication-related fields to their 'unreplicated' state.
+ */
+ @Override
+ public void stopReplication()
+ {
+
+ storageManagerLock.writeLock().lock();
+ try
+ {
+ bindingsJournal = originalBindingsJournal;
+ messageJournal = originalMessageJournal;
+ replicator = null;
+ }
+ finally
+ {
storageManagerLock.writeLock().unlock();
}
}
-
/**
* @param pageFilesToSync
* @throws Exception
@@ -530,10 +537,10 @@
private JournalFile[]
prepareJournalForCopy(Journal journal, JournalContent contentType, String nodeID) throws Exception
{
- journal.forceMoveNextFile();
- JournalFile[] datafiles = journal.getDataFiles();
+ journal.forceMoveNextFile();
+ JournalFile[] datafiles = journal.getDataFiles();
replicator.sendStartSyncMessage(datafiles, contentType, nodeID);
- return datafiles;
+ return datafiles;
}
public void waitOnOperations() throws Exception
@@ -4105,4 +4112,5 @@
}
+
}
\ No newline at end of file
Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java
===================================================================
--- trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2011-11-25 14:54:32 UTC (rev 11764)
+++ trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2011-11-25 15:00:31 UTC (rev 11765)
@@ -618,4 +618,10 @@
{
return false;
}
+
+ @Override
+ public void stopReplication()
+ {
+ // no-op
+ }
}
Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/server/impl/HornetQServerImpl.java
===================================================================
--- trunk/hornetq-core/src/main/java/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-11-25 14:54:32 UTC (rev 11764)
+++ trunk/hornetq-core/src/main/java/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-11-25 15:00:31 UTC (rev 11765)
@@ -88,6 +88,7 @@
import org.hornetq.core.protocol.core.Channel;
import org.hornetq.core.protocol.core.CoreRemotingConnection;
import org.hornetq.core.protocol.core.impl.ChannelImpl.CHANNEL_ID;
+import org.hornetq.core.remoting.FailureListener;
import org.hornetq.core.remoting.server.RemotingService;
import org.hornetq.core.remoting.server.impl.RemotingServiceImpl;
import org.hornetq.core.replication.ReplicationEndpoint;
@@ -227,6 +228,7 @@
private final Object initialiseLock = new Object();
private boolean initialised;
private final Object startUpLock = new Object();
+ private final Object replicationLock = new Object();
/**
* Only applicable to 'remote backup servers'. If this flag is false the backup may not become
@@ -1201,7 +1203,7 @@
{
if (configuration.isPersistenceEnabled())
{
- return new JournalStorageManager(configuration, executorFactory, replicationManager, shutdownOnCriticalIO);
+ return new JournalStorageManager(configuration, executorFactory, shutdownOnCriticalIO);
}
else
{
@@ -2265,30 +2267,60 @@
throw new HornetQException(HornetQException.ALREADY_REPLICATING);
}
- replicationManager = new ReplicationManagerImpl(rc, executorFactory);
- try
+ if (!isStarted())
{
- replicationManager.start();
- storageManager.startReplication(replicationManager, pagingManager, getNodeID().toString(), clusterConnection,
- pair);
+ throw new IllegalStateException();
}
- catch (Exception e)
+
+ synchronized (replicationLock)
{
- /*
- * The reasoning here is that the exception was either caused by (1) the (interaction with)
- * the backup, or (2) by an IO Error at the storage. If (1), we can swallow the exception
- * and ignore the replication request. If (2) the live will crash shortly.
- */
- log.warn("Exception when trying to start replication", e);
- replicationManager = null;
- if (e instanceof HornetQException)
+
+ if (replicationManager != null)
{
- throw (HornetQException)e;
+ throw new HornetQException(HornetQException.ALREADY_REPLICATING);
}
- else
+
+ rc.addFailureListener(new ReplicationFailureListener());
+ replicationManager = new ReplicationManagerImpl(rc, executorFactory);
+
+ try
{
- throw new HornetQException(HornetQException.INTERNAL_ERROR, "Error trying to start replication", e);
+ replicationManager.start();
+ storageManager.startReplication(replicationManager, pagingManager, getNodeID().toString(),
+ clusterConnection, pair);
}
+ catch (Exception e)
+ {
+ /*
+ * The reasoning here is that the exception was either caused by (1) the (interaction
+ * with) the backup, or (2) by an IO Error at the storage. If (1), we can swallow the
+ * exception and ignore the replication request. If (2) the live will crash shortly.
+ */
+ log.warn("Exception when trying to start replication", e);
+
+ try
+ {
+ if (replicationManager != null)
+ replicationManager.stop();
+ }
+ catch (Exception hqe)
+ {
+ log.warn("Exception while trying to close replicationManager", hqe);
+ }
+ finally
+ {
+ replicationManager = null;
+ }
+
+ if (e instanceof HornetQException)
+ {
+ throw (HornetQException)e;
+ }
+ else
+ {
+ throw new HornetQException(HornetQException.INTERNAL_ERROR, "Error trying to start replication", e);
+ }
+ }
}
}
@@ -2309,4 +2341,28 @@
nodeManager.setNodeID(nodeID);
backupUpToDate = true;
}
+
+ private final class ReplicationFailureListener implements FailureListener
+ {
+
+ @Override
+ public void connectionFailed(HornetQException exception, boolean failedOver)
+ {
+ Executors.newSingleThreadExecutor().execute(new Runnable()
+ {
+ public void run()
+ {
+ synchronized (replicationLock)
+
+ {
+ if (replicationManager != null)
+ {
+ storageManager.stopReplication();
+ }
+ }
+ }
+ });
+ }
+ }
+
}
12 years, 5 months
JBoss hornetq SVN: r11764 - branches/Branch_2_2_EAP.
by do-not-reply＠jboss.org
Author: gaohoward
Date: 2011-11-25 09:54:32 -0500 (Fri, 25 Nov 2011)
New Revision: 11764
Modified:
branches/Branch_2_2_EAP/build-maven.xml
Log:
fix windows build error
Modified: branches/Branch_2_2_EAP/build-maven.xml
===================================================================
--- branches/Branch_2_2_EAP/build-maven.xml 2011-11-25 08:29:12 UTC (rev 11763)
+++ branches/Branch_2_2_EAP/build-maven.xml 2011-11-25 14:54:32 UTC (rev 11764)
@@ -17,6 +17,10 @@
<property name="build.dir" value="build"/>
<property name="jars.dir" value="${build.dir}/jars"/>
+ <condition property="maven.executable" value="mvn.bat" else="mvn">
+ <os family="windows"/>
+ </condition>
+
<target name="uploadHornetQBootstrap">
<antcall target="upload">
<param name="artifact.id" value="hornetq-bootstrap"/>
@@ -162,7 +166,7 @@
<target name="upload-local-target">
<!-- install the jar -->
- <exec executable="mvn" dir="./build">
+ <exec executable="${maven.executable}" dir="./build">
<arg value="install:install-file"/>
<!-- uncomment the following line to deploy to the JBoss 5 repository -->
<!-- <arg value="-Dmaven.repo.local=/work/eap-51/maven-repository"/> -->
@@ -173,7 +177,7 @@
<arg value="-Dfile=./jars/${file-name}.jar"/>
</exec>
<!-- install the sources jar -->
- <exec executable="mvn" dir="./build">
+ <exec executable="${maven.executable}" dir="./build">
<arg value="install:install-file"/>
<arg value="-DgroupId=org.hornetq"/>
<!-- uncomment the following line to deploy to the JBoss 5 repository -->
@@ -188,7 +192,7 @@
<target name="upload">
<!-- upload the jar -->
- <exec executable="mvn">
+ <exec executable="${maven.executable}">
<arg value="deploy:deploy-file"/>
<arg value="-e"/>
<arg value="-DgroupId=org.hornetq"/>
@@ -201,7 +205,7 @@
<arg value="-Durl=dav:https://snapshots.jboss.org/maven2"/>
</exec>
<!-- upload the corresponding sources jar -->
- <exec executable="mvn">
+ <exec executable="${maven.executable}">
<arg value="deploy:deploy-file"/>
<arg value="-e"/>
<arg value="-DgroupId=org.hornetq"/>
@@ -283,7 +287,7 @@
<version>${hornetq.version}</version>${line.separator}</project>"/>
<!-- deploy the jar -->
- <exec executable="mvn">
+ <exec executable="${maven.executable}">
<arg value="-e"/>
<arg value="deploy:deploy-file"/>
<arg value="-DpomFile=${temporary.pom}"/>
@@ -296,7 +300,7 @@
<arg value="-Durl=https://repository.jboss.org/nexus/service/local/staging/deploy/maven2/"/>
</exec>
<!-- deploy the sources jar -->
- <exec executable="mvn">
+ <exec executable="${maven.executable}">
<arg value="-e"/>
<arg value="deploy:deploy-file"/>
<arg value="-DpomFile=${temporary.pom}"/>
12 years, 5 months