Skip to content

Commit 8fefa8a

Browse files
dnhatndnhatn
committed
Update InternalEngine tests on ccr side for #30121
Relates #30121
1 parent db14717 commit 8fefa8a

File tree

3 files changed

+30
-41
lines changed

3 files changed

+30
-41
lines changed

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

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -138,7 +138,6 @@
138138
import java.util.Comparator;
139139
import java.util.HashSet;
140140
import java.util.Iterator;
141-
import java.util.Iterator;
142141
import java.util.LinkedHashMap;
143142
import java.util.List;
144143
import java.util.Map;
@@ -1422,11 +1421,12 @@ public void testConcurrentOutOfOrderDocsOnReplica() throws IOException, Interrup
14221421
}
14231422
// randomly interleave
14241423
final AtomicLong seqNoGenerator = new AtomicLong();
1425-
Function<Engine.Operation, Engine.Operation> seqNoUpdater = operation -> {
1426-
final long newSeqNo = seqNoGenerator.getAndIncrement();
1424+
BiFunction<Engine.Operation, Long, Engine.Operation> seqNoUpdater = (operation, newSeqNo) -> {
14271425
if (operation instanceof Engine.Index) {
14281426
Engine.Index index = (Engine.Index) operation;
1429-
return new Engine.Index(index.uid(), index.parsedDoc(), newSeqNo, index.primaryTerm(), index.version(),
1427+
Document doc = testDocumentWithTextField(index.docs().get(0).get("value"));
1428+
ParsedDocument parsedDocument = testParsedDocument(index.id(), index.routing(), doc, index.source(), null);
1429+
return new Engine.Index(index.uid(), parsedDocument, newSeqNo, index.primaryTerm(), index.version(),
14301430
index.versionType(), index.origin(), index.startTime(), index.getAutoGeneratedIdTimestamp(), index.isRetry());
14311431
} else {
14321432
Engine.Delete delete = (Engine.Delete) operation;
@@ -1439,12 +1439,12 @@ public void testConcurrentOutOfOrderDocsOnReplica() throws IOException, Interrup
14391439
Iterator<Engine.Operation> iter2 = opsDoc2.iterator();
14401440
while (iter1.hasNext() && iter2.hasNext()) {
14411441
final Engine.Operation next = randomBoolean() ? iter1.next() : iter2.next();
1442-
allOps.add(seqNoUpdater.apply(next));
1442+
allOps.add(seqNoUpdater.apply(next, seqNoGenerator.getAndIncrement()));
14431443
}
1444-
iter1.forEachRemaining(o -> allOps.add(seqNoUpdater.apply(o)));
1445-
iter2.forEachRemaining(o -> allOps.add(seqNoUpdater.apply(o)));
1444+
iter1.forEachRemaining(o -> allOps.add(seqNoUpdater.apply(o, seqNoGenerator.getAndIncrement())));
1445+
iter2.forEachRemaining(o -> allOps.add(seqNoUpdater.apply(o, seqNoGenerator.getAndIncrement())));
14461446
// insert some duplicates
1447-
allOps.addAll(randomSubsetOf(allOps));
1447+
randomSubsetOf(allOps).forEach(op -> allOps.add(seqNoUpdater.apply(op, op.seqNo())));
14481448

14491449
shuffle(allOps, random());
14501450
concurrentlyApplyOps(allOps, engine);
@@ -4635,13 +4635,13 @@ public void testTrimUnsafeCommits() throws Exception {
46354635

46364636
public void testLuceneHistoryOnPrimary() throws Exception {
46374637
final List<Engine.Operation> operations = generateSingleDocHistory(false,
4638-
randomFrom(VersionType.INTERNAL, VersionType.EXTERNAL), false, 2, 10, 300);
4638+
randomFrom(VersionType.INTERNAL, VersionType.EXTERNAL), 2, 10, 300, "1");
46394639
assertOperationHistoryInLucene(operations);
46404640
}
46414641

46424642
public void testLuceneHistoryOnReplica() throws Exception {
46434643
final List<Engine.Operation> operations = generateSingleDocHistory(true,
4644-
randomFrom(VersionType.INTERNAL, VersionType.EXTERNAL), false, 2, 10, 300);
4644+
randomFrom(VersionType.INTERNAL, VersionType.EXTERNAL), 2, 10, 300, "2");
46454645
Randomness.shuffle(operations);
46464646
assertOperationHistoryInLucene(operations);
46474647
}

‎test/framework/src/main/java/org/elasticsearch/index/engine/EngineTestCase.java‎

Lines changed: 19 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -579,24 +579,13 @@ protected static void assertVisibleCount(InternalEngine engine, int numDocs, boo
579579
}
580580
}
581581

582-
public static List<Engine.Operation> generateSingleDocHistory(
583-
final boolean forReplica,
584-
final VersionType versionType,
585-
final boolean partialOldPrimary,
586-
final long primaryTerm,
587-
final int minOpCount,
588-
final int maxOpCount) {
582+
public static List<Engine.Operation> generateSingleDocHistory(boolean forReplica, VersionType versionType,
583+
long primaryTerm, int minOpCount, int maxOpCount, String docId) {
589584
final int numOfOps = randomIntBetween(minOpCount, maxOpCount);
590585
final List<Engine.Operation> ops = new ArrayList<>();
591-
final Term id = newUid("1");
592-
final int startWithSeqNo;
593-
if (partialOldPrimary) {
594-
startWithSeqNo = randomBoolean() ? numOfOps - 1 : randomIntBetween(0, numOfOps - 1);
595-
} else {
596-
startWithSeqNo = 0;
597-
}
598-
final int seqNoGap = randomBoolean() ? 1 : 2;
599-
final String valuePrefix = forReplica ? "r_" : "p_";
586+
final Term id = newUid(docId);
587+
final int startWithSeqNo = 0;
588+
final String valuePrefix = (forReplica ? "r_" : "p_" ) + docId + "_";
600589
final boolean incrementTermWhenIntroducingSeqNo = randomBoolean();
601590
for (int i = 0; i < numOfOps; i++) {
602591
final Engine.Operation op;
@@ -618,22 +607,22 @@ public static List<Engine.Operation> generateSingleDocHistory(
618607
throw new UnsupportedOperationException("unknown version type: " + versionType);
619608
}
620609
if (randomBoolean()) {
621-
op = new Engine.Index(id, testParsedDocument("1", null, testDocumentWithTextField(valuePrefix + i), B_1, null),
622-
forReplica && i >= startWithSeqNo ? i * seqNoGap : SequenceNumbers.UNASSIGNED_SEQ_NO,
623-
forReplica && i >= startWithSeqNo && incrementTermWhenIntroducingSeqNo ? primaryTerm + 1 : primaryTerm,
624-
version,
625-
forReplica ? versionType.versionTypeForReplicationAndRecovery() : versionType,
626-
forReplica ? REPLICA : PRIMARY,
627-
System.currentTimeMillis(), -1, false
610+
op = new Engine.Index(id, testParsedDocument(docId, null, testDocumentWithTextField(valuePrefix + i), B_1, null),
611+
forReplica && i >= startWithSeqNo ? i * 2 : SequenceNumbers.UNASSIGNED_SEQ_NO,
612+
forReplica && i >= startWithSeqNo && incrementTermWhenIntroducingSeqNo ? primaryTerm + 1 : primaryTerm,
613+
version,
614+
forReplica ? versionType.versionTypeForReplicationAndRecovery() : versionType,
615+
forReplica ? REPLICA : PRIMARY,
616+
System.currentTimeMillis(), -1, false
628617
);
629618
} else {
630-
op = new Engine.Delete("test", "1", id,
631-
forReplica && i >= startWithSeqNo ? i * seqNoGap : SequenceNumbers.UNASSIGNED_SEQ_NO,
632-
forReplica && i >= startWithSeqNo && incrementTermWhenIntroducingSeqNo ? primaryTerm + 1 : primaryTerm,
633-
version,
634-
forReplica ? versionType.versionTypeForReplicationAndRecovery() : versionType,
635-
forReplica ? REPLICA : PRIMARY,
636-
System.currentTimeMillis());
619+
op = new Engine.Delete("test", docId, id,
620+
forReplica && i >= startWithSeqNo ? i * 2 : SequenceNumbers.UNASSIGNED_SEQ_NO,
621+
forReplica && i >= startWithSeqNo && incrementTermWhenIntroducingSeqNo ? primaryTerm + 1 : primaryTerm,
622+
version,
623+
forReplica ? versionType.versionTypeForReplicationAndRecovery() : versionType,
624+
forReplica ? REPLICA : PRIMARY,
625+
System.currentTimeMillis());
637626
}
638627
ops.add(op);
639628
}

‎x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/index/engine/FollowingEngineTests.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -128,7 +128,7 @@ public void testOutOfOrderDocuments() throws IOException {
128128
try (FollowingEngine followingEngine = createEngine(store, engineConfig)) {
129129
final VersionType versionType =
130130
randomFrom(VersionType.INTERNAL, VersionType.EXTERNAL, VersionType.EXTERNAL_GTE, VersionType.FORCE);
131-
final List<Engine.Operation> ops = EngineTestCase.generateSingleDocHistory(true, versionType, false, 2, 2, 20);
131+
final List<Engine.Operation> ops = EngineTestCase.generateSingleDocHistory(true, versionType, 2, 2, 20, "id");
132132
EngineTestCase.assertOpsOnReplica(ops, followingEngine, true, logger);
133133
}
134134
}

0 commit comments

Comments
 (0)