The pending merge here is the cascade merge that a successful forced merge registers
* (forceMerge(1, false) followed by close()). Latches in a Directory wrapper order the three
* forced merges; an InfoStream hook only holds the successful merge thread for the few
* instructions between registering the cascade merge and asking the scheduler to start it. The
* one injected fault is an IOException from createOutput of one merge (a full disk); no fault
* fires after it.
*/
public class TestCloseAfterTragicMergeWithPendingMerge extends LuceneTestCase {
private static void await(CountDownLatch latch) {
try {
if (latch.await(60, TimeUnit.SECONDS) == false) {
throw new AssertionError("test latch timed out");
}
} catch (InterruptedException e) {
throw new AssertionError(e);
}
}
private static void addSegment(IndexWriter w, int numDocs, String prefix) throws IOException {
for (int i = 0; i < numDocs; i++) {
Document doc = new Document();
doc.add(new StringField("id", prefix + i, Field.Store.YES));
w.addDocument(doc);
}
w.commit();
}
/** The bug: one merge fails while the cascade merge is pending; close() never returns. */
public void testCloseReturnsAfterMergeFailureWithPendingMerge() throws Exception {
runScenario(true);
}
/** Control: the same schedule without the fault; close() commits and returns. */
public void testControlSameScheduleWithoutFault() throws Exception {
runScenario(false);
}
private void runScenario(boolean failC) throws Exception {
// Forced merges of LogDocMergePolicy(mergeFactor=2) over six segments of 1,1,2,2,4,4 docs:
// A=[s1,s2] (2 docs), B=[s3,s4] (4 docs), C=[s5,s6] (8 docs). When B finishes after A, the
// forced-merge cascade registers X=[A',B'] (6 docs).
final AtomicBoolean armed = new AtomicBoolean(false);
final CountDownLatch releaseA = new CountDownLatch(1);
final CountDownLatch releaseB = new CountDownLatch(1);
final CountDownLatch releaseC = new CountDownLatch(1);
final AtomicInteger cFaults = new AtomicInteger();
Directory dir =
new FilterDirectory(new ByteBuffersDirectory()) {
@Override
public IndexOutput createOutput(String name, IOContext context) throws IOException {
if (armed.get()
&& context.context() == IOContext.Context.MERGE
&& context.mergeInfo() != null) {
int docs = context.mergeInfo().totalMaxDoc();
if (docs == 2) {
await(releaseA);
} else if (docs == 4) {
await(releaseB);
} else if (docs == 8) {
await(releaseC);
if (failC && cFaults.getAndIncrement() == 0) {
throw new IOException("simulated: no space left on device");
}
}
}
return super.createOutput(name, context);
}
};
final CountDownLatch closeWaitingForMerges = new CountDownLatch(1);
final CountDownLatch firstMergeDone = new CountDownLatch(1);
final CountDownLatch secondMergeDoneHeld = new CountDownLatch(1);
final CountDownLatch releaseSecondMergeThread = new CountDownLatch(1);
final AtomicInteger mergedCount = new AtomicInteger();
InfoStream infoStream =
new InfoStream() {
@Override
public void message(String component, String message) {
if (armed.get() == false || "IW".equals(component) == false) {
return;
}
if (message.equals("waitForMerges")) {
closeWaitingForMerges.countDown();
} else if (message.startsWith("merged new segment")) {
// printed by IndexWriter.merge after the merge committed, outside of any lock
int n = mergedCount.incrementAndGet();
if (n == 1) {
firstMergeDone.countDown();
} else if (n == 2) {
secondMergeDoneHeld.countDown();
await(releaseSecondMergeThread);
}
}
}
@Override
public boolean isEnabled(String component) {
return "IW".equals(component);
}
@Override
public void close() {}
};
LogDocMergePolicy mp = new LogDocMergePolicy();
mp.setMergeFactor(2);
mp.setMinMergeDocs(1);
ConcurrentMergeScheduler cms = new ConcurrentMergeScheduler();
cms.setMaxMergesAndThreads(4, 4);
// only keeps the expected merge failure from failing the test as an uncaught exception
cms.setSuppressExceptions();
IndexWriterConfig iwc =
new IndexWriterConfig()
.setMergePolicy(NoMergePolicy.INSTANCE)
.setMergeScheduler(cms)
.setInfoStream(infoStream)
.setCommitOnClose(true)
.setMaxFullFlushMergeWaitMillis(0);
IndexWriter w = new IndexWriter(dir, iwc);
addSegment(w, 1, "a");
addSegment(w, 1, "b");
addSegment(w, 2, "c");
addSegment(w, 2, "d");
addSegment(w, 4, "e");
addSegment(w, 4, "f");
assertEquals(6, w.cloneSegmentInfos().size());
w.getConfig().setMergePolicy(mp);
armed.set(true);
// starts A, B and C in the background; all three wait in createOutput
w.forceMerge(1, false);
final AtomicReference