Skip to content

Commit f57eb99

Browse files
jasontedorjasontedor
committed
Add realistic tests for gaps in commit
1 parent 3c37f4b commit f57eb99

File tree

2 files changed

+98
-33
lines changed

2 files changed

+98
-33
lines changed

‎core/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

Lines changed: 7 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -184,12 +184,11 @@ public InternalEngine(EngineConfig engineConfig) throws EngineException {
184184
* that we there might be operations greater than the local checkpoint that will not be replayed. Here we force the local
185185
* checkpoint to the maximum sequence number in the commit (at the potential expense of correctness).
186186
*/
187-
while (seqNoService.getLocalCheckpoint() < seqNoService.getMaxSeqNo()) {
188-
final long next = seqNoService.getLocalCheckpoint() + 1;
189-
seqNoService.markSeqNoAsCompleted(next);
187+
while (seqNoService().getLocalCheckpoint() < seqNoService().getMaxSeqNo()) {
188+
seqNoService().markSeqNoAsCompleted(seqNoService().getLocalCheckpoint() + 1);
190189
}
191190
indexWriter = writer;
192-
translog = openTranslog(engineConfig, writer, seqNoService::getGlobalCheckpoint);
191+
translog = openTranslog(engineConfig, writer, () -> seqNoService().getLocalCheckpoint());
193192
assert translog.getGeneration() != null;
194193
} catch (IOException | TranslogCorruptedException e) {
195194
throw new EngineCreationFailureException(shardId, "failed to create engine", e);
@@ -690,7 +689,7 @@ private IndexResult innerIndex(Index index) throws IOException {
690689
} else {
691690
// no version conflict
692691
if (index.origin() == Operation.Origin.PRIMARY) {
693-
seqNo = seqNoService.generateSeqNo();
692+
seqNo = seqNoService().generateSeqNo();
694693
}
695694

696695
/**
@@ -721,7 +720,7 @@ private IndexResult innerIndex(Index index) throws IOException {
721720
return indexResult;
722721
} finally {
723722
if (seqNo != SequenceNumbersService.UNASSIGNED_SEQ_NO) {
724-
seqNoService.markSeqNoAsCompleted(seqNo);
723+
seqNoService().markSeqNoAsCompleted(seqNo);
725724
}
726725
}
727726

@@ -821,7 +820,7 @@ private DeleteResult innerDelete(Delete delete) throws IOException {
821820
deleteResult = result.get();
822821
} else {
823822
if (delete.origin() == Operation.Origin.PRIMARY) {
824-
seqNo = seqNoService.generateSeqNo();
823+
seqNo = seqNoService().generateSeqNo();
825824
}
826825

827826
updatedVersion = delete.versionType().updateVersion(currentVersion, expectedVersion);
@@ -839,7 +838,7 @@ private DeleteResult innerDelete(Delete delete) throws IOException {
839838
return deleteResult;
840839
} finally {
841840
if (seqNo != SequenceNumbersService.UNASSIGNED_SEQ_NO) {
842-
seqNoService.markSeqNoAsCompleted(deleteResult.getSeqNo());
841+
seqNoService().markSeqNoAsCompleted(deleteResult.getSeqNo());
843842
}
844843
}
845844
}

‎core/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java

Lines changed: 91 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -2931,39 +2931,106 @@ public void testSequenceIDs() throws Exception {
29312931
searchResult.close();
29322932
}
29332933

2934-
public void testSequenceNumberAdvancesToMaxSeqNoOnEngineOpen() throws IOException {
2934+
public void testSequenceNumberAdvancesToMaxSeqOnEngineOpenOnPrimary() throws BrokenBarrierException, InterruptedException, IOException {
29352935
engine.close();
2936-
final long maxSeqNo = randomIntBetween(16, 32);
2937-
final long localCheckpoint =
2938-
rarely() ? SequenceNumbersService.NO_OPS_PERFORMED : randomIntBetween(0, Math.toIntExact(maxSeqNo));
2939-
final long globalCheckpoint =
2940-
localCheckpoint == SequenceNumbersService.NO_OPS_PERFORMED || rarely()
2941-
? SequenceNumbersService.UNASSIGNED_SEQ_NO
2942-
: Math.toIntExact(localCheckpoint);
2943-
Engine initialEngine = null;
2936+
final int docs = randomIntBetween(1, 32);
2937+
InternalEngine initialEngine = null;
29442938
try {
2945-
initialEngine = createEngine(
2946-
defaultSettings,
2947-
store,
2948-
primaryTranslogDir,
2949-
newMergePolicy(),
2950-
null,
2951-
() -> new SequenceNumbersService(shardId, defaultSettings, maxSeqNo, localCheckpoint, globalCheckpoint) {
2939+
final CountDownLatch latch = new CountDownLatch(1);
2940+
final CyclicBarrier barrier = new CyclicBarrier(2);
2941+
final AtomicBoolean skip = new AtomicBoolean();
2942+
final AtomicLong expectedLocalCheckpoint = new AtomicLong(SequenceNumbersService.NO_OPS_PERFORMED);
2943+
final List<Thread> threads = new ArrayList<>();
2944+
final SequenceNumbersService seqNoService =
2945+
new SequenceNumbersService(
2946+
shardId,
2947+
defaultSettings,
2948+
SequenceNumbersService.NO_OPS_PERFORMED,
2949+
SequenceNumbersService.NO_OPS_PERFORMED,
2950+
SequenceNumbersService.UNASSIGNED_SEQ_NO) {
29522951
@Override
29532952
public long generateSeqNo() {
2954-
if (rarely()) {
2955-
// force skipping a sequence number
2956-
super.generateSeqNo();
2953+
final long seqNo = super.generateSeqNo();
2954+
if (skip.get()) {
2955+
try {
2956+
barrier.await();
2957+
latch.await();
2958+
} catch (BrokenBarrierException | InterruptedException e) {
2959+
throw new RuntimeException(e);
2960+
}
2961+
} else {
2962+
if (expectedLocalCheckpoint.get() + 1 == seqNo) {
2963+
expectedLocalCheckpoint.set(seqNo);
2964+
}
29572965
}
2958-
return super.generateSeqNo();
2966+
return seqNo;
29592967
}
2960-
});
2968+
};
2969+
initialEngine = createEngine(defaultSettings, store, primaryTranslogDir, newMergePolicy(), null, () -> seqNoService);
2970+
final InternalEngine finalInitialEngine = initialEngine;
2971+
for (int i = 0; i < docs; i++) {
2972+
final String id = Integer.toString(i);
2973+
final Term uid = newUid(id);
2974+
final ParsedDocument doc = testParsedDocument(id, id, "test", null, testDocumentWithTextField(), SOURCE, null);
2975+
2976+
skip.set(randomBoolean());
2977+
final Thread thread = new Thread(() -> finalInitialEngine.index(new Engine.Index(uid, doc)));
2978+
thread.start();
2979+
if (skip.get()) {
2980+
threads.add(thread);
2981+
barrier.await();
2982+
} else {
2983+
thread.join();
2984+
}
2985+
}
2986+
2987+
assertThat(initialEngine.seqNoService().getLocalCheckpoint(), equalTo(expectedLocalCheckpoint.get()));
2988+
assertThat(initialEngine.seqNoService().getMaxSeqNo(), equalTo((long) (docs - 1)));
2989+
initialEngine.flush(true, true);
2990+
2991+
latch.countDown();
2992+
for (final Thread thread : threads) {
2993+
thread.join();
2994+
}
2995+
} finally {
2996+
IOUtils.close(initialEngine);
2997+
}
2998+
2999+
try (final Engine recoveringEngine =
3000+
new InternalEngine(copy(initialEngine.config(), EngineConfig.OpenMode.OPEN_INDEX_AND_TRANSLOG))) {
3001+
// there might be gaps in the translog; the best that we can say is that the local checkpoint is at least as far the max
3002+
// sequence number
3003+
assertThat(recoveringEngine.seqNoService().getLocalCheckpoint(), greaterThanOrEqualTo((long) (docs - 1)));
3004+
}
3005+
}
3006+
3007+
public void testSequenceNumberAdvancesToMaxSeqNoOnEngineOpenOnReplica() throws IOException {
3008+
final long v = Versions.MATCH_ANY;
3009+
final VersionType t = VersionType.INTERNAL;
3010+
final long ts = IndexRequest.UNSET_AUTO_GENERATED_TIMESTAMP;
3011+
final int docs = randomIntBetween(1, 32);
3012+
InternalEngine initialEngine = null;
3013+
try {
3014+
initialEngine = engine;
3015+
for (int i = 0; i < docs; i++) {
3016+
final String id = Integer.toString(i);
3017+
final Term uid = newUid(id);
3018+
final ParsedDocument doc = testParsedDocument(id, id, "test", null, testDocumentWithTextField(), SOURCE, null);
3019+
// create a gap at sequence number 3 * i + 1
3020+
initialEngine.index(new Engine.Index(uid, doc, 3 * i, 1, v, t, REPLICA, System.nanoTime(), ts, false));
3021+
initialEngine.delete(new Engine.Delete("type", id, uid, 3 * i + 2, 1, v, t, REPLICA, System.nanoTime()));
3022+
}
3023+
3024+
// bake the commit with the local checkpoint stuck at 0 and gaps all along the way up to the max sequence number
3025+
assertThat(initialEngine.seqNoService().getLocalCheckpoint(), equalTo((long) 0));
3026+
assertThat(initialEngine.seqNoService().getMaxSeqNo(), equalTo((long) (3 * (docs - 1) + 2)));
29613027
initialEngine.flush(true, true);
2962-
final int docs = randomIntBetween(1, 32);
3028+
29633029
for (int i = 0; i < docs; i++) {
29643030
final String id = Integer.toString(i);
3031+
final Term uid = newUid(id);
29653032
final ParsedDocument doc = testParsedDocument(id, id, "test", null, testDocumentWithTextField(), SOURCE, null);
2966-
initialEngine.index(new Engine.Index(newUid(id), doc));
3033+
initialEngine.index(new Engine.Index(uid, doc, 3 * i + 1, 1, v, t, REPLICA, System.nanoTime(), ts, false));
29673034
}
29683035
} finally {
29693036
IOUtils.close(initialEngine);
@@ -2973,8 +3040,7 @@ public long generateSeqNo() {
29733040
new InternalEngine(copy(initialEngine.config(), EngineConfig.OpenMode.OPEN_INDEX_AND_TRANSLOG))) {
29743041
// there might be gaps in the translog; the best that we can say is that the local checkpoint is at least as far the max
29753042
// sequence number
2976-
// TODO: tighten this assertion after gaps in the translog are addressed
2977-
assertThat(recoveringEngine.seqNoService().getLocalCheckpoint(), greaterThanOrEqualTo(maxSeqNo));
3043+
assertThat(recoveringEngine.seqNoService().getLocalCheckpoint(), greaterThanOrEqualTo((long) (3 * docs + 2 - 1)));
29783044
}
29793045
}
29803046

0 commit comments

Comments
 (0)