[hornetq-commits] JBoss hornetq SVN: r11159 - in branches/HORNETQ-720_Replication: hornetq-journal/src/main/java/org/hornetq/core/journal and 3 other directories.
do-not-reply at jboss.org
do-not-reply at jboss.org
Tue Aug 9 06:29:32 EDT 2011
Author: borges
Date: 2011-08-09 06:29:32 -0400 (Tue, 09 Aug 2011)
New Revision: 11159
Modified:
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicatedJournal.java
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java
branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/Journal.java
branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalImpl.java
branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/ReplicatingJournal.java
branches/HORNETQ-720_Replication/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupJournalSyncTest.java
branches/HORNETQ-720_Replication/tests/integration-tests/src/test/java/org/hornetq/tests/integration/replication/ReplicationTest.java
Log:
HORNETQ-720 Reload after sync is done.
Modified: branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicatedJournal.java
===================================================================
--- branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicatedJournal.java 2011-08-09 10:27:51 UTC (rev 11158)
+++ branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicatedJournal.java 2011-08-09 10:29:32 UTC (rev 11159)
@@ -570,6 +570,11 @@
localJournal.lineUpContex(callback);
}
+ @Override
+ public JournalLoadInformation loadSyncOnly() throws Exception
+ {
+ return localJournal.loadSyncOnly();
+ }
// Package protected ---------------------------------------------
Modified: branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java
===================================================================
--- branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2011-08-09 10:27:51 UTC (rev 11158)
+++ branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2011-08-09 10:29:32 UTC (rev 11159)
@@ -81,7 +81,7 @@
private Channel channel;
private Journal[] journals;
- private JournalLoadInformation[] journalLoadInformation;
+ private final JournalLoadInformation[] journalLoadInformation = new JournalLoadInformation[2];
/** Files reserved in each journal for synchronization of existing data from the 'live' server. */
private final Map<JournalContent, Map<Long, JournalFile>> filesReservedForSync =
@@ -244,13 +244,11 @@
for (JournalContent jc : EnumSet.allOf(JournalContent.class))
{
- System.out.println("State? " + journalsHolder.get(jc));
filesReservedForSync.put(jc, new HashMap<Long, JournalFile>());
+ // We only need to load internal structures on the backup...
+ journalLoadInformation[jc.typeByte] = journalsHolder.get(jc).loadSyncOnly();
}
- // We only need to load internal structures on the backup...
- journalLoadInformation = storage.loadInternalOnly();
-
pageManager = new PagingManagerImpl(new PagingStoreFactoryNIO(config.getPagingDirectory(),
config.getJournalBufferSize_NIO(),
server.getScheduledPool(),
@@ -395,7 +393,7 @@
{
for (JournalContent jc : EnumSet.allOf(JournalContent.class))
{
- JournalImpl journal = (JournalImpl)journalsHolder.get(jc);
+ JournalImpl journal = (JournalImpl)journalsHolder.remove(jc);
journal.writeLock();
try
{
@@ -404,10 +402,12 @@
throw new IllegalStateException("Journal should not have any data files at this point");
}
// files should be already in place.
- filesReservedForSync.remove(msg.getJournalContent());
- registerJournal(jc.typeByte, journalsHolder.get(jc));
+ filesReservedForSync.remove(jc);
+ registerJournal(jc.typeByte, journal);
+ journal.loadInternalOnly();
// XXX HORNETQ-720 must reload journals
// XXX HORNETQ-720 must start using real journals
+
}
finally
{
Modified: branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/Journal.java
===================================================================
--- branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/Journal.java 2011-08-09 10:27:51 UTC (rev 11158)
+++ branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/Journal.java 2011-08-09 10:29:32 UTC (rev 11159)
@@ -111,11 +111,18 @@
JournalLoadInformation load(LoaderCallback reloadManager) throws Exception;
- /** Load internal data structures and not expose any data.
- * This is only useful if you're using the journal but not interested on the current data.
- * Useful in situations where the journal is being replicated, copied... etc. */
+ /**
+ * Load internal data structures and not expose any data. This is only useful if you're using the
+ * journal but not interested on the current data. Useful in situations where the journal is
+ * being replicated, copied... etc.
+ */
JournalLoadInformation loadInternalOnly() throws Exception;
+ /**
+ * Load internal data structures, and remain waiting for synchronization to complete.
+ */
+ JournalLoadInformation loadSyncOnly() throws Exception;
+
void lineUpContex(IOCompletion callback);
JournalLoadInformation load(List<RecordInfo> committedRecords,
Modified: branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalImpl.java
===================================================================
--- branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalImpl.java 2011-08-09 10:27:51 UTC (rev 11158)
+++ branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalImpl.java 2011-08-09 10:29:32 UTC (rev 11159)
@@ -1298,35 +1298,39 @@
return fileFactory.getAlignment();
}
- public synchronized JournalLoadInformation loadInternalOnly() throws Exception
+ private static class DummyLoader implements LoaderCallback
{
- LoaderCallback dummyLoader = new LoaderCallback()
+ static final LoaderCallback INSTANCE = new DummyLoader();
+ public void failedTransaction(final long transactionID, final List<RecordInfo> records,
+ final List<RecordInfo> recordsToDelete)
{
+ }
- public void failedTransaction(final long transactionID,
- final List<RecordInfo> records,
- final List<RecordInfo> recordsToDelete)
- {
- }
+ public void updateRecord(final RecordInfo info)
+ {
+ }
- public void updateRecord(final RecordInfo info)
- {
- }
+ public void deleteRecord(final long id)
+ {
+ }
- public void deleteRecord(final long id)
- {
- }
+ public void addRecord(final RecordInfo info)
+ {
+ }
- public void addRecord(final RecordInfo info)
- {
- }
+ public void addPreparedTransaction(final PreparedTransactionInfo preparedTransaction)
+ {
+ }
+ }
- public void addPreparedTransaction(final PreparedTransactionInfo preparedTransaction)
- {
- }
- };
+ public synchronized JournalLoadInformation loadInternalOnly() throws Exception
+ {
+ return load(DummyLoader.INSTANCE, true, false);
+ }
- return this.load(dummyLoader, true, true);
+ public synchronized JournalLoadInformation loadSyncOnly() throws Exception
+ {
+ return load(DummyLoader.INSTANCE, true, true);
}
public JournalLoadInformation load(final List<RecordInfo> committedRecords,
@@ -1746,12 +1750,16 @@
private synchronized JournalLoadInformation load(final LoaderCallback loadManager, boolean fixFailingTransactions,
final boolean replicationSync) throws Exception
{
-
- if (state != JournalState.STARTED)
+ System.out.println("LOAD! " + state + " " + replicationSync);
+ if (state == JournalState.STOPPED || state == JournalState.LOADED)
{
throw new IllegalStateException("Journal " + this + " must be in " + JournalState.STARTED + " state, was " +
state);
}
+ if (state == JournalState.SYNCING && replicationSync)
+ {
+ throw new IllegalStateException("Journal must be state " + JournalState.STARTED);
+ }
checkControlFile();
Modified: branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/ReplicatingJournal.java
===================================================================
--- branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/ReplicatingJournal.java 2011-08-09 10:27:51 UTC (rev 11158)
+++ branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/ReplicatingJournal.java 2011-08-09 10:29:32 UTC (rev 11159)
@@ -223,4 +223,10 @@
{
throw new UnsupportedOperationException();
}
+
+ @Override
+ public JournalLoadInformation loadSyncOnly() throws Exception
+ {
+ throw new UnsupportedOperationException();
+ }
}
Modified: branches/HORNETQ-720_Replication/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupJournalSyncTest.java
===================================================================
--- branches/HORNETQ-720_Replication/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupJournalSyncTest.java 2011-08-09 10:27:51 UTC (rev 11158)
+++ branches/HORNETQ-720_Replication/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupJournalSyncTest.java 2011-08-09 10:29:32 UTC (rev 11159)
@@ -161,10 +161,6 @@
@Override
protected void tearDown() throws Exception
{
- if (handler != null)
- {
- handler.notifyAll();
- }
if (sessionFactory != null)
sessionFactory.close();
if (session != null)
Modified: branches/HORNETQ-720_Replication/tests/integration-tests/src/test/java/org/hornetq/tests/integration/replication/ReplicationTest.java
===================================================================
--- branches/HORNETQ-720_Replication/tests/integration-tests/src/test/java/org/hornetq/tests/integration/replication/ReplicationTest.java 2011-08-09 10:27:51 UTC (rev 11158)
+++ branches/HORNETQ-720_Replication/tests/integration-tests/src/test/java/org/hornetq/tests/integration/replication/ReplicationTest.java 2011-08-09 10:29:32 UTC (rev 11159)
@@ -846,5 +846,16 @@
}
+ /*
+ * (non-Javadoc)
+ * @see org.hornetq.core.journal.Journal#loadSyncOnly()
+ */
+ @Override
+ public JournalLoadInformation loadSyncOnly() throws Exception
+ {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
}
}
More information about the hornetq-commits
mailing list