Menu
▾
▴
[Bigdata-commit] SF.net SVN: bigdata:[7508] branches/BIGDATA_RELEASE_1_3_0
|
From: <tho...@us...> - 2013-11-04 16:42:36
|
Revision: 7508
http://bigdata.svn.sourceforge.net/bigdata/?rev=7508&view=rev
Author: thompsonbry
Date: 2013-11-04 16:42:25 +0000 (Mon, 04 Nov 2013)
Log Message:
-----------
See #764 (RESYNC failure)
The root cause that abort2Phase() discards the live HALog file and does not immediately re-create that HALog file. Given A + B + C, if all three services are at the same commit point and there is an attempt to commit which fails, causing abort2Phase() to be invoked. If C is then shutdown it will fail to RESYNC (per this ticket) since the HALog file is missing on the leader (A).
abort2Phase() called doLocalAbort() which called through to discardWriteSet(), which was calling disableHALog() and deleting the live HALog file. We have modified discardWriteSet() to create the new live HALog file IF the service is joined with the met quorum at the time that discardWriteSet() was invoked. This should fix the problem with abort2Phase() leaving the leader without a live HALog file.
We have also modified getHARootBlocksForWriteSet() to conditionally create the live HALog file if it does not exist. This is the method that is invoked by the RESYNC task. Thus, by ensuring that the live HALog file exists here we can make RESYNC much more robust. We believe that a data race exists between when a service is elected as the leader and when the HALog file would normally be created. It is possible that a 3rd service in a simultaneous restart of {A,B,C} could attempt to obtain the live HALog file before the leader would have otherwise created it. Explicitly creating the live HALog within getHARootBlocksForWriteSet() closes out this possible deadlock.
- conditionalCreateHALog() was moved to HALogNexus.
- AbstractJournal._abort() was made robust to the case were the HAQuorumService was not running.
- HALogWriter: there were some methods that failed to obtain the m_stateLock before examining fields that were protected by that lock. Those methods included: getCommitCounter(), getSequence(), and isHALogOpen().
- CommitCounterUtility: exposed method to format the commitCounter as a 21-digit string with leading zeros.
- HAStatuServlet: modified to disclose the most current historical HALog file (if any) and the live HALog file (if it exists). This was done to help diagnose the problem associated with this ticket.
- HAJournalServer::ResyncTask().doRun() - remove the conditionalCreateHALog() invocation. There was no reason to create the live HALog on the service that was trying to resync. The code comment indicated that this might have been done because of test suite expectations, but the test suite runs fine without this.
- HA CI test suite: Added a number of startABC_restartX tests in an effort to recreate the problem associated with this ticket. These tests did not managed to recreate the problem, but they are being retained because I do not see obvious overlap in the existing test suite with these tests.
- TestHAJournalServerOverride.testStartABC_abort2Phase_restartC() :: Created a unit test that replicates the problem against the revision of the code before this commit. This test can be observed to fail if you disable the HALog create in both HAJournal.getHARootBlocksForWriteSet() and HAJournalServer.discardWriteSet().
Note: If this solution proves to be robust, then we could consider re-enabling the automatic logic to do a disaster rebuild rather than transitioning to an OPERATOR state. However, transitioning to an OPERATOR state is perhaps wise regardless.
HA CI is locally green.
Modified Paths:
--------------
branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/ha/halog/HALogWriter.java
branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/AbstractJournal.java
branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/CommitCounterUtility.java
branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/java/com/bigdata/journal/jini/ha/HAJournal.java
branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/java/com/bigdata/journal/jini/ha/HAJournalServer.java
branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/java/com/bigdata/journal/jini/ha/HALogNexus.java
branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/AbstractHA3JournalServerTestCase.java
branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/HAJournalTest.java
branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHA3JournalServer.java
branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHA3JournalServerWithHALogs.java
branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHAJournalServerOverride.java
branches/BIGDATA_RELEASE_1_3_0/bigdata-sails/src/java/com/bigdata/rdf/sail/BigdataSail.java
branches/BIGDATA_RELEASE_1_3_0/bigdata-sails/src/java/com/bigdata/rdf/sail/webapp/HAStatusServletUtil.java
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/ha/halog/HALogWriter.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/ha/halog/HALogWriter.java 2013-11-01 21:10:45 UTC (rev 7507)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/ha/halog/HALogWriter.java 2013-11-04 16:42:25 UTC (rev 7508)
@@ -37,6 +37,7 @@
import org.apache.log4j.Logger;
+import com.bigdata.ha.msg.HAWriteMessage;
import com.bigdata.ha.msg.IHAWriteMessage;
import com.bigdata.io.FileChannelUtility;
import com.bigdata.io.IReopenChannel;
@@ -51,10 +52,10 @@
/**
* Wrapper class to handle process log creation and output for HA.
- *
- * The process log stores the HAWriteMessages and buffers to support reading and
- * reprocessing as part of the HA synchronization protocol.
- *
+ * <p>
+ * The process log stores the {@link HAWriteMessage} and buffers to support
+ * reading and reprocessing as part of the HA synchronization protocol.
+ * <p>
* The writer encapsulates not only the writing of individual messages but also
* the closing and creation of new files.
*
@@ -128,27 +129,32 @@
/** current write point on the channel. */
private long m_position = headerSize0;
- /**
- * Return the commit counter that is expected for the writes that will be
- * logged (the same commit counter that is on the opening root block).
- */
+ @Override
public long getCommitCounter() {
- assertOpen();
+ final Lock lock = m_stateLock.readLock();
+ lock.lock();
+ try {
+ assertOpen();
+ return m_rootBlock.getCommitCounter();
+ } finally {
+ lock.unlock();
+ }
- return m_rootBlock.getCommitCounter();
-
}
- /**
- * Return the sequence number that is expected for the next write.
- */
+ @Override
public long getSequence() {
- assertOpen();
+ final Lock lock = m_stateLock.readLock();
+ lock.lock();
+ try {
+ assertOpen();
+ return m_nextSequence;
+ } finally {
+ lock.unlock();
+ }
- return m_nextSequence;
-
}
/**
@@ -162,9 +168,16 @@
}
+ @Override
public boolean isHALogOpen() {
-
- return m_state != null && !m_state.isCommitted();
+
+ final Lock lock = m_stateLock.readLock();
+ lock.lock();
+ try {
+ return m_state != null && !m_state.isCommitted();
+ } finally {
+ lock.unlock();
+ }
}
@@ -179,7 +192,6 @@
} finally {
lock.unlock();
}
-
}
/**
@@ -225,6 +237,7 @@
}
+ @Override
public String toString() {
final IRootBlockView tmp = m_rootBlock;
@@ -375,6 +388,7 @@
* The final root block for the write set.
* @throws IOException
*/
+ @Override
public void closeHALog(final IRootBlockView rootBlock)
throws FileNotFoundException, IOException {
@@ -472,6 +486,7 @@
* @throws IOException
* if we can not write on the log.
*/
+ @Override
public void writeOnHALog(final IHAWriteMessage msg, final ByteBuffer data)
throws IOException, IllegalStateException {
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/AbstractJournal.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/AbstractJournal.java 2013-11-01 21:10:45 UTC (rev 7507)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/AbstractJournal.java 2013-11-04 16:42:25 UTC (rev 7508)
@@ -2835,10 +2835,26 @@
* last live HA message).
*/
- final QuorumService<HAGlue> localService = quorum.getClient();
+ QuorumService<HAGlue> localService = null;
+ try {
+
+ localService = quorum.getClient();
+
+ } catch (IllegalStateException ex) {
+
+ /*
+ * Note: Thrown if the QuorumService is not running.
+ */
+
+ // ignore.
+ }
- localService.discardWriteSet();
-
+ if (localService != null) {
+
+ localService.discardWriteSet();
+
+ }
+
}
if (log.isInfoEnabled())
@@ -3854,11 +3870,11 @@
final String msg = "commit: commitTime="
+ cs.commitTime
+ ", latency="
- + TimeUnit.NANOSECONDS.toMillis(elapsedNanos)
- + ", nextOffset="
- + cs.newRootBlock.getNextOffset()
- + ", byteCount="
- + (cs.newRootBlock.getNextOffset() - cs.byteCountBefore);
+ + TimeUnit.NANOSECONDS.toMillis(elapsedNanos);
+// + ", nextOffset="
+// + cs.newRootBlock.getNextOffset()
+// + ", byteCount="
+// + (cs.newRootBlock.getNextOffset() - cs.byteCountBefore);
if (BigdataStatics.debug)
System.err.println(msg);
else if (log.isInfoEnabled())
@@ -7694,6 +7710,7 @@
this.abortMessage = abortMessage;
}
+ @Override
public void run() {
try {
@@ -7731,7 +7748,7 @@
// ALWAYS go through the local abort.
doLocalAbort();
-
+
}
}
@@ -7750,6 +7767,7 @@
* @todo Since these are rare events it may not be worthwhile to setup a
* separate low-level socket service to send/receive the data.
*/
+ @Override
public Future<IHAReadResponse> readFromDisk(
final IHAReadRequest msg) {
@@ -7762,7 +7780,8 @@
final FutureTask<IHAReadResponse> ft = new FutureTask<IHAReadResponse>(
new Callable<IHAReadResponse>() {
-
+
+ @Override
public IHAReadResponse call() throws Exception {
if (haLog.isInfoEnabled())
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/CommitCounterUtility.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/CommitCounterUtility.java 2013-11-01 21:10:45 UTC (rev 7507)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata/src/java/com/bigdata/journal/CommitCounterUtility.java 2013-11-04 16:42:25 UTC (rev 7508)
@@ -68,21 +68,8 @@
* the file name and then partitioning it into groups of THREE (3)
* digits.
*/
- final String basename;
- {
+ final String basename = getCommitCounterStr(commitCounter);
- final StringBuilder sb = new StringBuilder();
-
- final Formatter f = new Formatter(sb);
-
- f.format("%021d", commitCounter);
- f.flush();
- f.close();
-
- basename = sb.toString();
-
- }
-
/*
* Now figure out the recursive directory name.
*/
@@ -105,6 +92,32 @@
}
/**
+ * Format the commit counter with leading zeros such that it will be
+ * lexically ordered in the file system.
+ *
+ * @param commitCounter
+ * The commit counter.
+ *
+ * @return The basename of the file consisting of the foramtted commit
+ * counter with the appropriate leading zeros.
+ */
+ public static String getCommitCounterStr(final long commitCounter) {
+
+ final StringBuilder sb = new StringBuilder(21);
+
+ final Formatter f = new Formatter(sb);
+
+ f.format("%021d", commitCounter);
+ f.flush();
+ f.close();
+
+ final String basename = sb.toString();
+
+ return basename;
+
+ }
+
+ /**
* Parse out the commitCounter from the file name.
*
* @param name
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/java/com/bigdata/journal/jini/ha/HAJournal.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/java/com/bigdata/journal/jini/ha/HAJournal.java 2013-11-01 21:10:45 UTC (rev 7507)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/java/com/bigdata/journal/jini/ha/HAJournal.java 2013-11-04 16:42:25 UTC (rev 7508)
@@ -102,6 +102,7 @@
import com.bigdata.service.proxy.RemoteFuture;
import com.bigdata.service.proxy.RemoteFutureImpl;
import com.bigdata.service.proxy.ThickFuture;
+import com.bigdata.util.StackInfoReport;
/**
* A {@link Journal} that that participates in a write replication pipeline. The
@@ -846,7 +847,44 @@
final Lock logLock = getHALogNexus().getLogLock();
logLock.lock();
try {
-
+
+ /*
+ * Verify that this service is the quorum leader.
+ */
+ final long token = getQuorum().token();
+
+ // Method is only allowed on the quorum leader.
+ getQuorumService().assertLeader(token);
+
+ if (!getHALogNexus().isHALogOpen()) {
+
+ /**
+ * The live HALog should always exist on the leader.
+ * However, the leader is defined by the zookeeper state and
+ * it is possible that the leader has not yet gone through
+ * the code path to create the HALog. Thus, for safety, we
+ * ensure that the live HALog exists here.
+ *
+ * Note: we are holding the logLock.
+ *
+ * Note: This only causes the HALog to be created on the
+ * quorum leader. HAJournalServer.discardWriteSet() handles
+ * this for both the leader and the joined followers.
+ *
+ * @see <a
+ * href="https://sourceforge.net/apps/trac/bigdata/ticket/764"
+ * > RESYNC fails (HA) </a>
+ */
+
+ if (haLog.isInfoEnabled())
+ log.info(
+ "Live HALog does not exist on the quorum leader",
+ new StackInfoReport());
+
+ getHALogNexus().conditionalCreateHALog();
+
+ }
+
// The commit counter of the desired closing root block.
final long commitCounter = msg.getCommitCounter();
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/java/com/bigdata/journal/jini/ha/HAJournalServer.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/java/com/bigdata/journal/jini/ha/HAJournalServer.java 2013-11-01 21:10:45 UTC (rev 7507)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/java/com/bigdata/journal/jini/ha/HAJournalServer.java 2013-11-04 16:42:25 UTC (rev 7508)
@@ -1382,17 +1382,30 @@
journal.getHALogNexus().lastLiveHAWriteMessage = null;
if (journal.getHALogNexus().isHALogOpen()) {
- /*
- * Note: Closing the HALog is necessary for us to be able to
- * re-enter SeekConsensus without violating a pre-condition
- * for that run state.
- */
+ /**
+ * Note: Closing the HALog is necessary for us to be able to
+ * re-enter SeekConsensus without violating a pre-condition
+ * for that run state.
+ *
+ * @see <a
+ * href="https://sourceforge.net/apps/trac/bigdata/ticket/764"
+ * > RESYNC fails (HA) </a>
+ */
try {
journal.getHALogNexus().disableHALog();
} catch (IOException e) {
log.error(e, e);
}
- }
+ final long token = getQuorum().token();
+ if (isJoinedMember(token)) {
+ try {
+ journal.getHALogNexus().createHALog(
+ journal.getRootBlockView());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
} finally {
logLock.unlock();
}
@@ -1767,6 +1780,7 @@
public QuorumMeetTask(final long token, final UUID leaderId) {
this.token = token;
}
+ @Override
public Void call() throws Exception {
journal.setQuorumToken(token);
if (isJoinedMember(token)) {
@@ -2834,9 +2848,6 @@
journal.doLocalAbort();
- // Sets up expectations (maybe just for the test suite?)
- conditionalCreateHALog();
-
/*
* We will do a local commit with each HALog (aka write set)
* that is replicated. This let's us catch up incrementally with
@@ -3393,49 +3404,6 @@
}
- /**
- * Conditionally create the HALog.
- * <p>
- * Refactored out of {@link #pipelineSetup()} since
- * {@link #discardWriteSet()} now removes the current HALog. Therefore,
- * the {@link ResyncTask} needs to call
- * {@link #conditionalCreateHALog()} <em>after</em> it calls
- * {@link AbstractJournal#doLocalAbort()}.
- *
- * @throws FileNotFoundException
- * @throws IOException
- */
- private void conditionalCreateHALog() throws FileNotFoundException,
- IOException {
-
- logLock.lock();
-
- try {
-
- if (!journal.getHALogNexus().isHALogOpen()) {
-
- /*
- * Open the HALogWriter for our current root blocks.
- *
- * Note: We always use the current root block when receiving
- * an HALog file, even for historical writes. This is
- * because the historical log writes occur when we ask the
- * leader to send us a prior commit point in RESYNC.
- */
-
- journal.getHALogNexus().createHALog(
- journal.getRootBlockView());
-
- }
-
- } finally {
-
- logLock.unlock();
-
- }
-
- }
-
@Override
protected void handleReplicatedWrite(final IHASyncRequest req,
final IHAWriteMessage msg, final ByteBuffer data)
@@ -3472,7 +3440,7 @@
logLock.lock();
try {
- conditionalCreateHALog();
+ journal.getHALogNexus().conditionalCreateHALog();
if (haLog.isDebugEnabled())
haLog.debug("msg=" + msg + ", buf=" + data);
@@ -4078,7 +4046,7 @@
try {
- conditionalCreateHALog();
+ journal.getHALogNexus().conditionalCreateHALog();
/*
* Throws IllegalStateException if the message is not
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/java/com/bigdata/journal/jini/ha/HALogNexus.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/java/com/bigdata/journal/jini/ha/HALogNexus.java 2013-11-01 21:10:45 UTC (rev 7507)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/java/com/bigdata/journal/jini/ha/HALogNexus.java 2013-11-04 16:42:25 UTC (rev 7508)
@@ -33,7 +33,6 @@
import java.util.Arrays;
import java.util.Iterator;
import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -1054,6 +1053,43 @@
}
+ /**
+ * Conditionally create the HALog.
+ *
+ * @throws FileNotFoundException
+ * @throws IOException
+ */
+ public void conditionalCreateHALog() throws FileNotFoundException,
+ IOException {
+
+ logLock.lock();
+
+ try {
+
+ if (!isHALogOpen()) {
+
+ /*
+ * Open the HALogWriter for our current root blocks.
+ *
+ * Note: We always use the current root block when receiving an
+ * HALog file, even for historical writes. This is because the
+ * historical log writes occur when we ask the leader to send us
+ * a prior commit point in RESYNC.
+ */
+
+ createHALog(journal.getRootBlockView());
+
+ }
+
+ } finally {
+
+ logLock.unlock();
+
+ }
+
+ }
+
+ @Override
public boolean isHALogOpen() {
logLock.lock();
@@ -1070,6 +1106,7 @@
}
+ @Override
public void closeHALog(final IRootBlockView rootBlock)
throws IOException {
@@ -1116,6 +1153,7 @@
}
+ @Override
public void disableHALog() throws IOException {
logLock.lock();
@@ -1132,6 +1170,7 @@
}
+ @Override
public void writeOnHALog(final IHAWriteMessage msg, final ByteBuffer data)
throws IOException, IllegalStateException {
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/AbstractHA3JournalServerTestCase.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/AbstractHA3JournalServerTestCase.java 2013-11-01 21:10:45 UTC (rev 7507)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/AbstractHA3JournalServerTestCase.java 2013-11-04 16:42:25 UTC (rev 7508)
@@ -606,16 +606,26 @@
}, 5, TimeUnit.SECONDS);
}
-
+
+ /**
+ * Start A then B then C. As each service starts, this method waits for that
+ * service to appear in the pipeline in the proper position.
+ *
+ * @return The ordered array of services <code>[A, B, C]</code>
+ */
protected HAGlue[] startSequenceABC() throws Exception {
- startA();
- awaitPipeline(new HAGlue[] {serverA});
- startB();
- awaitPipeline(new HAGlue[] {serverA, serverB});
- startC();
- awaitPipeline(new HAGlue[] {serverA, serverB, serverC});
-
- return new HAGlue[] {serverA, serverB, serverC};
+
+ startA();
+ awaitPipeline(new HAGlue[] { serverA });
+
+ startB();
+ awaitPipeline(new HAGlue[] { serverA, serverB });
+
+ startC();
+ awaitPipeline(new HAGlue[] { serverA, serverB, serverC });
+
+ return new HAGlue[] { serverA, serverB, serverC };
+
}
/*
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/HAJournalTest.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/HAJournalTest.java 2013-11-01 21:10:45 UTC (rev 7507)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/HAJournalTest.java 2013-11-04 16:42:25 UTC (rev 7508)
@@ -85,11 +85,16 @@
import com.bigdata.ha.msg.IHAWriteSetStateResponse;
import com.bigdata.journal.AbstractJournal;
import com.bigdata.journal.IRootBlockView;
+import com.bigdata.journal.ITx;
import com.bigdata.journal.jini.ha.HAJournalServer.HAQuorumService;
import com.bigdata.journal.jini.ha.HAJournalServer.RunStateEnum;
import com.bigdata.quorum.AsynchronousQuorumCloseException;
import com.bigdata.quorum.Quorum;
import com.bigdata.quorum.zk.ZKQuorumImpl;
+import com.bigdata.rdf.sail.BigdataSail;
+import com.bigdata.rdf.sail.BigdataSailRepository;
+import com.bigdata.rdf.sail.BigdataSailRepositoryConnection;
+import com.bigdata.rdf.store.AbstractTripleStore;
import com.bigdata.service.jini.RemoteDestroyAdmin;
/**
@@ -287,6 +292,12 @@
*/
public RunStateEnum getRunStateEnum() throws IOException;
+ /**
+ * Run a simple update transaction on the quorum leader, but abort()
+ * rather than committing the transaction.
+ */
+ public void simpleTransaction_abort() throws IOException, Exception;
+
}
/**
@@ -1167,6 +1178,65 @@
// @Override
+ @Override
+ public void simpleTransaction_abort() throws IOException, Exception {
+
+ final Quorum<HAGlue, QuorumService<HAGlue>> quorum = getQuorum();
+
+ // Note: Throws IllegalStateException if quorum is not running.
+ final QuorumService<HAGlue> quorumService = quorum.getClient();
+
+ final long token = quorum.token();
+
+ // This service must be the quorum leader.
+ quorumService.assertLeader(token);
+
+ // The default namespace.
+ final String namespace = BigdataSail.Options.DEFAULT_NAMESPACE;
+
+ // resolve the default namespace.
+ final AbstractTripleStore tripleStore = (AbstractTripleStore) getIndexManager()
+ .getResourceLocator().locate(namespace, ITx.UNISOLATED);
+
+ if (tripleStore == null) {
+
+ throw new RuntimeException("Not found: namespace=" + namespace);
+
+ }
+
+ // Wrap with SAIL.
+ final BigdataSail sail = new BigdataSail(tripleStore);
+
+ final BigdataSailRepository repo = new BigdataSailRepository(sail);
+
+ repo.initialize();
+
+ final BigdataSailRepositoryConnection conn = (BigdataSailRepositoryConnection) repo
+ .getUnisolatedConnection();
+
+ try {
+
+// conn.setAutoCommit(false);
+//
+// final ValueFactory f = sail.getValueFactory();
+//
+// conn.add(f.createStatement(
+// f.createURI("http://www.bigdata.com"), RDF.TYPE,
+// RDFS.RESOURCE), null/* contexts */);
+//
+// conn.flush();
+
+ // Fall through.
+
+ } finally {
+
+ // Force abort.
+ conn.rollback();
+
+ }
+
+ }
+
} // class HAGlueTestImpl
private static class MyPrepareMessage implements IHA2PhasePrepareMessage {
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHA3JournalServer.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHA3JournalServer.java 2013-11-01 21:10:45 UTC (rev 7507)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHA3JournalServer.java 2013-11-04 16:42:25 UTC (rev 7508)
@@ -82,7 +82,6 @@
*/
public class TestHA3JournalServer extends AbstractHA3JournalServerTestCase {
-
/**
* {@inheritDoc}
* <p>
@@ -558,6 +557,181 @@
}
/**
+ * Unit test for a situation in which A B and C start. A quorum mets and the
+ * third service resyncs with the met quorum. The quorum then fully meets.
+ * Once the fully met quorum is stable, C is then restarted. This test
+ * exercises a code path that handles the case where C is current, but is
+ * forced into RESYNC in case there are writes in progress on the leader.
+ * <p>
+ * Note: In this version of the test, the HALog files are purged at each
+ * commit of the fully met quorum. Another version of this test exists in
+ * which the HALog files are NOT purged at each commit of a fully met
+ * quorum.
+ */
+ public void testStartABC_restartC() throws Exception {
+
+ final ABC x = new ABC(true/*sequential*/);
+
+ final long token = awaitFullyMetQuorum();
+
+ // Now run several transactions
+ final int NTX = 5;
+ for (int i = 0; i < NTX; i++)
+ simpleTransaction();
+
+ // wait until the commit point is registered on all services.
+ awaitCommitCounter(NTX + 1L, new HAGlue[] { x.serverA, x.serverB,
+ x.serverC });
+
+ /*
+ * The same number of HALog files should exist on all services.
+ *
+ * Note: the restore policy is setup such that we are purging the HALog
+ * files at each commit of a fully met quorum.
+ */
+ awaitLogCount(getHALogDirA(), 1L);
+ awaitLogCount(getHALogDirB(), 1L);
+ awaitLogCount(getHALogDirC(), 1L);
+
+ // shutdown C.
+ shutdownC();
+
+ // wait for C to be gone from zookeeper.
+ awaitPipeline(new HAGlue[] { x.serverA, x.serverB });
+ awaitMembers(new HAGlue[] { x.serverA, x.serverB });
+ awaitJoined(new HAGlue[] { x.serverA, x.serverB });
+
+ // restart C.
+ /*final HAGlue serverC =*/ startC();
+
+ // wait until the quorum fully meets again (on the same token).
+ assertEquals(token, awaitFullyMetQuorum());
+
+ // Verify expected HALog files.
+ awaitLogCount(getHALogDirA(), 1L);
+ awaitLogCount(getHALogDirB(), 1L);
+ awaitLogCount(getHALogDirC(), 1L);
+
+ }
+
+ /**
+ * Unit test for a situation in which A B and C start. A quorum mets and the
+ * third service resyncs with the met quorum. The quorum then fully meets.
+ * Once the fully met quorum is stable, B is then restarted. The pipeline is
+ * reorganized when B is shutdown but the quorum does not break. This test
+ * exercises a code path that handles the case where B is current, but is
+ * forced into RESYNC in case there are writes in progress on the leader.
+ * <p>
+ * Note: In this version of the test, the HALog files are NOT purged at each
+ * commit of the fully met quorum.
+ */
+ public void testStartABC_restartB() throws Exception {
+
+ final ABC x = new ABC(true/*sequential*/);
+
+ final long token = awaitFullyMetQuorum();
+
+ // Now run several transactions
+ final int NTX = 5;
+ for (int i = 0; i < NTX; i++)
+ simpleTransaction();
+
+ // wait until the commit point is registered on all services.
+ awaitCommitCounter(NTX + 1L, new HAGlue[] { x.serverA, x.serverB,
+ x.serverC });
+
+ /*
+ * The same number of HALog files should exist on all services.
+ *
+ * Note: the restore policy is setup such that we are purging the HALog
+ * files at each commit of a fully met quorum.
+ */
+ awaitLogCount(getHALogDirA(), 1L);
+ awaitLogCount(getHALogDirB(), 1L);
+ awaitLogCount(getHALogDirC(), 1L);
+
+ // shutdown B.
+ shutdownB();
+
+ // wait for B to be gone from zookeeper.
+ awaitPipeline(new HAGlue[] { x.serverA, x.serverC });
+ awaitMembers(new HAGlue[] { x.serverA, x.serverC });
+ awaitJoined(new HAGlue[] { x.serverA, x.serverC });
+
+ // restart B.
+ /*final HAGlue serverB =*/ startB();
+
+ // wait until the quorum fully meets again (on the same token).
+ assertEquals(token, awaitFullyMetQuorum());
+
+ // Verify expected HALog files.
+ awaitLogCount(getHALogDirA(), 1L);
+ awaitLogCount(getHALogDirB(), 1L);
+ awaitLogCount(getHALogDirC(), 1L);
+
+ }
+
+ /**
+ * Unit test for a situation in which A B and C start. A quorum mets and the
+ * third service resyncs with the met quorum. The quorum then fully meets.
+ * Once the fully met quorum is stable, A is then restarted. The pipeline is
+ * reorganized when A is shutdown and a new leader is elected. This test
+ * exercises a code path that handles the case where A is current, but is
+ * forced into RESYNC in case there are writes in progress on the leader.
+ * <p>
+ * Note: In this version of the test, the HALog files are NOT purged at each
+ * commit of the fully met quorum.
+ */
+ public void testStartABC_restartA() throws Exception {
+
+ final ABC x = new ABC(true/*sequential*/);
+
+ final long token = awaitFullyMetQuorum();
+
+ // Now run several transactions
+ final int NTX = 5;
+ for (int i = 0; i < NTX; i++)
+ simpleTransaction();
+
+ // wait until the commit point is registered on all services.
+ awaitCommitCounter(NTX + 1L, new HAGlue[] { x.serverA, x.serverB,
+ x.serverC });
+
+ /*
+ * The same number of HALog files should exist on all services.
+ *
+ * Note: the restore policy is setup such that we are purging the HALog
+ * files at each commit of a fully met quorum.
+ */
+ awaitLogCount(getHALogDirA(), 1L);
+ awaitLogCount(getHALogDirB(), 1L);
+ awaitLogCount(getHALogDirC(), 1L);
+
+ // shutdown A.
+ shutdownA();
+
+ // wait for A to be gone from zookeeper.
+// awaitPipeline(new HAGlue[] { x.serverA, x.serverC });
+// awaitMembers(new HAGlue[] { x.serverA, x.serverC });
+// awaitJoined(new HAGlue[] { x.serverA, x.serverC });
+
+ // since the leader failed over, the quorum meets on a new token.
+ final long token2 = awaitNextQuorumMeet(token);
+
+ // restart A.
+ /*final HAGlue serverA =*/ startA();
+
+ // wait until the quorum fully meets again (on the same token).
+ assertEquals(token2, awaitFullyMetQuorum());
+
+ // Verify expected HALog files.
+ awaitLogCount(getHALogDirA(), 1L);
+ awaitLogCount(getHALogDirB(), 1L);
+ awaitLogCount(getHALogDirC(), 1L);
+
+ }
+
+ /**
* Unit test of the ability to go through a simultaneous restart of all
* services once those services are no longer at commit point 0. Two
* services will meet on the lastCommitTime. The third will need to RESYNC
@@ -2405,7 +2579,6 @@
}
-
/**
* Tests shutdown of met quorum, but leader first forces re-organisation concurrent
* with service shutdown
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHA3JournalServerWithHALogs.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHA3JournalServerWithHALogs.java 2013-11-01 21:10:45 UTC (rev 7507)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHA3JournalServerWithHALogs.java 2013-11-04 16:42:25 UTC (rev 7508)
@@ -269,5 +269,178 @@
awaitLogCount(getHALogDirC(), commitCounter2+1);
}
+
+ /**
+ * Unit test for a situation in which A B and C start. A quorum mets and the
+ * third service resyncs with the met quorum. The quorum then fully meets.
+ * Once the fully met quorum is stable, C is then restarted. This test
+ * exercises a code path that handles the case where C is current, but is
+ * forced into RESYNC in case there are writes in progress on the leader.
+ * <p>
+ * Note: In this version of the test, the HALog files are NOT purged at each
+ * commit of the fully met quorum.
+ */
+ public void testStartABC_restartC() throws Exception {
+
+ final ABC x = new ABC(true/*sequential*/);
+
+ final long token = awaitFullyMetQuorum();
+
+ // Now run several transactions
+ final int NTX = 5;
+ for (int i = 0; i < NTX; i++)
+ simpleTransaction();
+ // wait until the commit point is registered on all services.
+ awaitCommitCounter(NTX + 1L, new HAGlue[] { x.serverA, x.serverB,
+ x.serverC });
+
+ /*
+ * The same number of HALog files should exist on all services.
+ *
+ * Note: the restore policy is setup such that we are NOT purging the HALog
+ * files at each commit of a fully met quorum.
+ */
+ awaitLogCount(getHALogDirA(), NTX + 2L);
+ awaitLogCount(getHALogDirB(), NTX + 2L);
+ awaitLogCount(getHALogDirC(), NTX + 2L);
+
+ // shutdown C.
+ shutdownC();
+
+ // wait for C to be gone from zookeeper.
+ awaitPipeline(new HAGlue[] { x.serverA, x.serverB });
+ awaitMembers(new HAGlue[] { x.serverA, x.serverB });
+ awaitJoined(new HAGlue[] { x.serverA, x.serverB });
+
+ // restart C.
+ /*final HAGlue serverC =*/ startC();
+
+ // wait until the quorum fully meets again (on the same token).
+ assertEquals(token, awaitFullyMetQuorum());
+
+ // Verify expected HALog files.
+ awaitLogCount(getHALogDirA(), NTX + 2L);
+ awaitLogCount(getHALogDirB(), NTX + 2L);
+ awaitLogCount(getHALogDirC(), NTX + 2L);
+
+ }
+
+ /**
+ * Unit test for a situation in which A B and C start. A quorum mets and the
+ * third service resyncs with the met quorum. The quorum then fully meets.
+ * Once the fully met quorum is stable, B is then restarted. The pipeline is
+ * reorganized when B is shutdown but the quorum does not break. This test
+ * exercises a code path that handles the case where B is current, but is
+ * forced into RESYNC in case there are writes in progress on the leader.
+ * <p>
+ * Note: In this version of the test, the HALog files are NOT purged at each
+ * commit of the fully met quorum.
+ */
+ public void testStartABC_restartB() throws Exception {
+
+ final ABC x = new ABC(true/*sequential*/);
+
+ final long token = awaitFullyMetQuorum();
+
+ // Now run several transactions
+ final int NTX = 5;
+ for (int i = 0; i < NTX; i++)
+ simpleTransaction();
+
+ // wait until the commit point is registered on all services.
+ awaitCommitCounter(NTX + 1L, new HAGlue[] { x.serverA, x.serverB,
+ x.serverC });
+
+ /*
+ * The same number of HALog files should exist on all services.
+ *
+ * Note: the restore policy is setup such that we are purging the HALog
+ * files at each commit of a fully met quorum.
+ */
+ awaitLogCount(getHALogDirA(), NTX + 2L);
+ awaitLogCount(getHALogDirB(), NTX + 2L);
+ awaitLogCount(getHALogDirC(), NTX + 2L);
+
+ // shutdown B.
+ shutdownB();
+
+ // wait for B to be gone from zookeeper.
+ awaitPipeline(new HAGlue[] { x.serverA, x.serverC });
+ awaitMembers(new HAGlue[] { x.serverA, x.serverC });
+ awaitJoined(new HAGlue[] { x.serverA, x.serverC });
+
+ // restart B.
+ /*final HAGlue serverB =*/ startB();
+
+ // wait until the quorum fully meets again (on the same token).
+ assertEquals(token, awaitFullyMetQuorum());
+
+ // Verify expected HALog files.
+ awaitLogCount(getHALogDirA(), NTX + 2L);
+ awaitLogCount(getHALogDirB(), NTX + 2L);
+ awaitLogCount(getHALogDirC(), NTX + 2L);
+
+ }
+
+ /**
+ * Unit test for a situation in which A B and C start. A quorum mets and the
+ * third service resyncs with the met quorum. The quorum then fully meets.
+ * Once the fully met quorum is stable, A is then restarted. The pipeline is
+ * reorganized when A is shutdown and a new leader is elected. This test
+ * exercises a code path that handles the case where A is current, but is
+ * forced into RESYNC in case there are writes in progress on the leader.
+ * <p>
+ * Note: In this version of the test, the HALog files are NOT purged at each
+ * commit of the fully met quorum.
+ */
+ public void testStartABC_restartA() throws Exception {
+
+ final ABC x = new ABC(true/*sequential*/);
+
+ final long token = awaitFullyMetQuorum();
+
+ // Now run several transactions
+ final int NTX = 5;
+ for (int i = 0; i < NTX; i++)
+ simpleTransaction();
+
+ // wait until the commit point is registered on all services.
+ awaitCommitCounter(NTX + 1L, new HAGlue[] { x.serverA, x.serverB,
+ x.serverC });
+
+ /*
+ * The same number of HALog files should exist on all services.
+ *
+ * Note: the restore policy is setup such that we are NOT purging the HALog
+ * files at each commit of a fully met quorum.
+ */
+ awaitLogCount(getHALogDirA(), NTX + 2L);
+ awaitLogCount(getHALogDirB(), NTX + 2L);
+ awaitLogCount(getHALogDirC(), NTX + 2L);
+
+ // shutdown A.
+ shutdownA();
+
+ // wait for A to be gone from zookeeper.
+// awaitPipeline(new HAGlue[] { x.serverA, x.serverC });
+// awaitMembers(new HAGlue[] { x.serverA, x.serverC });
+// awaitJoined(new HAGlue[] { x.serverA, x.serverC });
+
+ // since the leader failed over, the quorum meets on a new token.
+ final long token2 = awaitNextQuorumMeet(token);
+
+ // restart A.
+ /*final HAGlue serverA =*/ startA();
+
+ // wait until the quorum fully meets again (on the same token).
+ assertEquals(token2, awaitFullyMetQuorum());
+
+ // Verify expected HALog files.
+ awaitLogCount(getHALogDirA(), NTX + 2L);
+ awaitLogCount(getHALogDirB(), NTX + 2L);
+ awaitLogCount(getHALogDirC(), NTX + 2L);
+
+ }
+
}
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHAJournalServerOverride.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHAJournalServerOverride.java 2013-11-01 21:10:45 UTC (rev 7507)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata-jini/src/test/com/bigdata/journal/jini/ha/TestHAJournalServerOverride.java 2013-11-04 16:42:25 UTC (rev 7508)
@@ -550,7 +550,7 @@
// Verify quorum is unchanged.
assertEquals(token, quorum.token());
- // Should be two commit points on {A,C].
+ // Should be two commit points on {A,C}.
awaitCommitCounter(2L, startup.serverA, startup.serverC);
// Just one commit point on B.
@@ -633,6 +633,76 @@
}
/**
+ * Unit test for failure to RESYNC having a root cause that the live HALog
+ * file did not exist on the quorum leader after an abort2Phase() call.
+ *
+ * @see <a href="https://sourceforge.net/apps/trac/bigdata/ticket/764" >
+ * RESYNC fails (HA) </a>
+ */
+ public void testStartABC_abort2Phase_restartC() throws Exception {
+
+ final ABC x = new ABC(true/*sequential*/);
+
+ final long token = awaitFullyMetQuorum();
+
+ // Now run several transactions
+ final int NTX = 5;
+ for (int i = 0; i < NTX; i++)
+ simpleTransaction();
+
+ // wait until the commit point is registered on all services.
+ awaitCommitCounter(NTX + 1L, new HAGlue[] { x.serverA, x.serverB,
+ x.serverC });
+
+ /*
+ * The same number of HALog files should exist on all services.
+ *
+ * Note: the restore policy is setup such that we are purging the HALog
+ * files at each commit of a fully met quorum.
+ */
+ awaitLogCount(getHALogDirA(), 1L);
+ awaitLogCount(getHALogDirB(), 1L);
+ awaitLogCount(getHALogDirC(), 1L);
+
+ // shutdown C.
+ shutdownC();
+
+ // wait for C to be gone from zookeeper.
+ awaitPipeline(new HAGlue[] { x.serverA, x.serverB });
+ awaitMembers(new HAGlue[] { x.serverA, x.serverB });
+ awaitJoined(new HAGlue[] { x.serverA, x.serverB });
+
+ /*
+ * Run a transaction that forces a 2-phase abort.
+ *
+ * Note: Historically, this would cause the live HALog on the leader to
+ * be disabled (aka deleted) without causing that live HALog file to be
+ * recreated. This is ticket #764.
+ */
+ ((HAGlueTest) x.serverA).simpleTransaction_abort();
+
+ /*
+ * Restart C.
+ *
+ * Note: C will go into RESYNC. Since all services are at the same
+ * commit point, C will attempt to replicate the live HALog from the
+ * leader. Once it obtains that HALog, it should figure out that it has
+ * the live HALog and attempt to transition atomically to a joined
+ * service.
+ */
+ /*final HAGlue serverC =*/ startC();
+
+ // wait until the quorum fully meets again (on the same token).
+ assertEquals(token, awaitFullyMetQuorum());
+
+ // Verify expected HALog files.
+ awaitLogCount(getHALogDirA(), 1L);
+ awaitLogCount(getHALogDirB(), 1L);
+ awaitLogCount(getHALogDirC(), 1L);
+
+ }
+
+ /**
* 2 services start, quorum meets then we bounce the zookeeper connection
* for the follower and verify that the quorum meets again.
* <p>
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata-sails/src/java/com/bigdata/rdf/sail/BigdataSail.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata-sails/src/java/com/bigdata/rdf/sail/BigdataSail.java 2013-11-01 21:10:45 UTC (rev 7507)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata-sails/src/java/com/bigdata/rdf/sail/BigdataSail.java 2013-11-04 16:42:25 UTC (rev 7508)
@@ -625,7 +625,7 @@
*
* @see Options
*/
- public BigdataSail(Properties properties) {
+ public BigdataSail(final Properties properties) {
this(createLTS(properties));
Modified: branches/BIGDATA_RELEASE_1_3_0/bigdata-sails/src/java/com/bigdata/rdf/sail/webapp/HAStatusServletUtil.java
===================================================================
--- branches/BIGDATA_RELEASE_1_3_0/bigdata-sails/src/java/com/bigdata/rdf/sail/webapp/HAStatusServletUtil.java 2013-11-01 21:10:45 UTC (rev 7507)
+++ branches/BIGDATA_RELEASE_1_3_0/bigdata-sails/src/java/com/bigdata/rdf/sail/webapp/HAStatusServletUtil.java 2013-11-04 16:42:25 UTC (rev 7508)
@@ -48,6 +48,7 @@
import com.bigdata.ha.msg.HARemoteRebuildRequest;
import com.bigdata.ha.msg.HASnapshotRequest;
import com.bigdata.ha.msg.IHARemoteRebuildRequest;
+import com.bigdata.journal.CommitCounterUtility;
import com.bigdata.journal.IIndexManager;
import com.bigdata.journal.IRootBlockView;
import com.bigdata.journal.RootBlockView;
@@ -326,8 +327,9 @@
int nfiles = 0;
long nbytes = 0L;
final Iterator<IHALogRecord> itr = nexus.getHALogs();
+ IHALogRecord r = null;
while (itr.hasNext()) {
- final IHALogRecord r = itr.next();
+ r = itr.next();
nbytes += r.sizeOnDisk();
nfiles++;
}
@@ -339,13 +341,25 @@
nbytes += currentFile.length();
nfiles++;
}
- final String compressorKey = journal.getProperties().getProperty(
- com.bigdata.journal.Options.HALOG_COMPRESSOR,
- com.bigdata.journal.Options.DEFAULT_HALOG_COMPRESSOR);
- p.text("HALogDir: nfiles=" + nfiles + ", nbytes=" + nbytes
- + ", path=" + nexus.getHALogDir()
- + ", compressorKey=" + compressorKey).node("br")
- .close();
+ final String compressorKey = journal
+ .getProperties()
+ .getProperty(
+ com.bigdata.journal.Options.HALOG_COMPRESSOR,
+ com.bigdata.journal.Options.DEFAULT_HALOG_COMPRESSOR);
+ p.text("HALogDir: nfiles="
+ + nfiles
+ + ", nbytes="
+ + nbytes
+ + ", path="
+ + nexus.getHALogDir()
+ + ", compressorKey="
+ + compressorKey
+ + ", lastHALogClosed="
+ + (r == null ? "N/A" : CommitCounterUtility
+ .getCommitCounterStr(r.getCommitCounter()))
+ + ", liveLog="
+ + (currentFile == null ? "N/A" : currentFile
+ .getName())).node("br").close();
}
if (digests) {
/*
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
