Skip to content

Commit 7b42081

Browse files
dnhatndnhatn
committed
Advance max_seq_no before add operation to Lucene (#38879)
Today when processing an operation on a replica engine (or the following engine), we first add it to Lucene, then add it to translog, then finally marks its seq_no as completed. If a flush occurs after step1, but before step-3, the max_seq_no in the commit's user_data will be smaller than the seq_no of some documents in the Lucene commit.
1 parent b14405d commit 7b42081

File tree

10 files changed

+142
-31
lines changed

10 files changed

+142
-31
lines changed

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

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1004,6 +1004,7 @@ assert config().getIndexSettings().getIndexVersionCreated().before(Version.V_6_0
10041004
}
10051005
}
10061006
}
1007+
markSeqNoAsSeen(index.seqNo());
10071008
return plan;
10081009
}
10091010

@@ -1371,6 +1372,7 @@ assert config().getIndexSettings().getIndexVersionCreated().before(Version.V_6_0
13711372
delete.seqNo(), delete.version());
13721373
}
13731374
}
1375+
markSeqNoAsSeen(delete.seqNo());
13741376
return plan;
13751377
}
13761378

@@ -1525,6 +1527,7 @@ public void maybePruneDeletes() {
15251527
public NoOpResult noOp(final NoOp noOp) {
15261528
NoOpResult noOpResult;
15271529
try (ReleasableLock ignored = readLock.acquire()) {
1530+
markSeqNoAsSeen(noOp.seqNo());
15281531
noOpResult = innerNoOp(noOp);
15291532
} catch (final Exception e) {
15301533
noOpResult = new NoOpResult(getPrimaryTerm(), noOp.seqNo(), e);
@@ -2498,6 +2501,13 @@ public void waitForOpsToComplete(long seqNo) throws InterruptedException {
24982501
localCheckpointTracker.waitForOpsToComplete(seqNo);
24992502
}
25002503

2504+
/**
2505+
* Marks the given seq_no as seen and advances the max_seq_no of this engine to at least that value.
2506+
*/
2507+
protected final void markSeqNoAsSeen(long seqNo) {
2508+
localCheckpointTracker.advanceMaxSeqNo(seqNo);
2509+
}
2510+
25012511
/**
25022512
* Checks if the given operation has been processed in this engine or not.
25032513
* @return true if the given operation was processed; otherwise false.

‎server/src/main/java/org/elasticsearch/index/seqno/LocalCheckpointTracker.java

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,6 +81,15 @@ public synchronized long generateSeqNo() {
8181
return nextSeqNo++;
8282
}
8383

84+
/**
85+
* Marks the provided sequence number as seen and updates the max_seq_no if needed.
86+
*/
87+
public synchronized void advanceMaxSeqNo(long seqNo) {
88+
if (seqNo >= nextSeqNo) {
89+
nextSeqNo = seqNo + 1;
90+
}
91+
}
92+
8493
/**
8594
* Marks the processing of the provided sequence number as completed as updates the checkpoint if possible.
8695
*

‎server/src/test/java/org/elasticsearch/client/transport/TransportClientNodesServiceTests.java

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -215,11 +215,7 @@ public void close() {
215215
transport.endConnectMode();
216216
transportService.stop();
217217
transportClientNodesService.close();
218-
try {
219-
terminate(threadPool);
220-
} catch (InterruptedException e) {
221-
throw new AssertionError(e);
222-
}
218+
terminate(threadPool);
223219
}
224220
}
225221

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

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5581,4 +5581,42 @@ void assertLuceneOperations(InternalEngine engine, long expectedAppends, long ex
55815581
assertThat(message, engine.getNumDocUpdates(), equalTo(expectedUpdates));
55825582
assertThat(message, engine.getNumDocDeletes(), equalTo(expectedDeletes));
55835583
}
5584+
5585+
public void testMaxSeqNoInCommitUserData() throws Exception {
5586+
AtomicBoolean running = new AtomicBoolean(true);
5587+
Thread rollTranslog = new Thread(() -> {
5588+
while (running.get() && engine.getTranslog().currentFileGeneration() < 500) {
5589+
engine.rollTranslogGeneration(); // make adding operations to translog slower
5590+
}
5591+
});
5592+
rollTranslog.start();
5593+
5594+
Thread indexing = new Thread(() -> {
5595+
long seqNo = 0;
5596+
while (running.get() && seqNo <= 1000) {
5597+
try {
5598+
String id = Long.toString(between(1, 50));
5599+
if (randomBoolean()) {
5600+
ParsedDocument doc = testParsedDocument(id, null, testDocumentWithTextField(), SOURCE, null);
5601+
engine.index(replicaIndexForDoc(doc, 1L, seqNo, false));
5602+
} else {
5603+
engine.delete(replicaDeleteForDoc(id, 1L, seqNo, 0L));
5604+
}
5605+
seqNo++;
5606+
} catch (IOException e) {
5607+
throw new AssertionError(e);
5608+
}
5609+
}
5610+
});
5611+
indexing.start();
5612+
5613+
int numCommits = between(5, 20);
5614+
for (int i = 0; i < numCommits; i++) {
5615+
engine.flush(false, true);
5616+
}
5617+
running.set(false);
5618+
indexing.join();
5619+
rollTranslog.join();
5620+
assertMaxSeqNoInCommitUserData(engine);
5621+
}
55845622
}

‎server/src/test/java/org/elasticsearch/transport/TransportActionProxyTests.java

Lines changed: 1 addition & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -68,13 +68,7 @@ public void setUp() throws Exception {
6868
@Override
6969
public void tearDown() throws Exception {
7070
super.tearDown();
71-
IOUtils.close(serviceA, serviceB, serviceC, () -> {
72-
try {
73-
terminate(threadPool);
74-
} catch (InterruptedException e) {
75-
Thread.currentThread().interrupt();
76-
}
77-
});
71+
IOUtils.close(serviceA, serviceB, serviceC, () -> terminate(threadPool));
7872
}
7973

8074
private MockTransportService buildService(final Version version) {

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

Lines changed: 31 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,8 @@
2727
import org.apache.lucene.document.NumericDocValuesField;
2828
import org.apache.lucene.document.StoredField;
2929
import org.apache.lucene.document.TextField;
30+
import org.apache.lucene.index.DirectoryReader;
31+
import org.apache.lucene.index.IndexCommit;
3032
import org.apache.lucene.index.IndexWriter;
3133
import org.apache.lucene.index.IndexWriterConfig;
3234
import org.apache.lucene.index.LeafReader;
@@ -124,6 +126,7 @@
124126
import static org.elasticsearch.index.engine.Engine.Operation.Origin.REPLICA;
125127
import static org.elasticsearch.index.translog.TranslogDeletionPolicies.createTranslogDeletionPolicy;
126128
import static org.hamcrest.Matchers.equalTo;
129+
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
127130
import static org.hamcrest.Matchers.notNullValue;
128131

129132
public abstract class EngineTestCase extends ESTestCase {
@@ -251,18 +254,20 @@ public EngineConfig copy(EngineConfig config, MergePolicy mergePolicy) {
251254
@After
252255
public void tearDown() throws Exception {
253256
super.tearDown();
254-
if (engine != null && engine.isClosed.get() == false) {
255-
engine.getTranslog().getDeletionPolicy().assertNoOpenTranslogRefs();
256-
assertConsistentHistoryBetweenTranslogAndLuceneIndex(engine, createMapperService("test"));
257-
}
258-
if (replicaEngine != null && replicaEngine.isClosed.get() == false) {
259-
replicaEngine.getTranslog().getDeletionPolicy().assertNoOpenTranslogRefs();
260-
assertConsistentHistoryBetweenTranslogAndLuceneIndex(replicaEngine, createMapperService("test"));
257+
try {
258+
if (engine != null && engine.isClosed.get() == false) {
259+
engine.getTranslog().getDeletionPolicy().assertNoOpenTranslogRefs();
260+
assertConsistentHistoryBetweenTranslogAndLuceneIndex(engine, createMapperService("test"));
261+
assertMaxSeqNoInCommitUserData(engine);
262+
}
263+
if (replicaEngine != null && replicaEngine.isClosed.get() == false) {
264+
replicaEngine.getTranslog().getDeletionPolicy().assertNoOpenTranslogRefs();
265+
assertConsistentHistoryBetweenTranslogAndLuceneIndex(replicaEngine, createMapperService("test"));
266+
assertMaxSeqNoInCommitUserData(replicaEngine);
267+
}
268+
} finally {
269+
IOUtils.close(replicaEngine, storeReplica, engine, store, () -> terminate(threadPool));
261270
}
262-
IOUtils.close(
263-
replicaEngine, storeReplica,
264-
engine, store);
265-
terminate(threadPool);
266271
}
267272

268273

@@ -992,6 +997,21 @@ public static void assertConsistentHistoryBetweenTranslogAndLuceneIndex(Engine e
992997
}
993998
}
994999

1000+
/**
1001+
* Asserts that the max_seq_no stored in the commit's user_data is never smaller than seq_no of any document in the commit.
1002+
*/
1003+
public static void assertMaxSeqNoInCommitUserData(Engine engine) throws Exception {
1004+
List<IndexCommit> commits = DirectoryReader.listCommits(engine.store.directory());
1005+
for (IndexCommit commit : commits) {
1006+
try (DirectoryReader reader = DirectoryReader.open(commit)) {
1007+
AtomicLong maxSeqNoFromDocs = new AtomicLong(SequenceNumbers.NO_OPS_PERFORMED);
1008+
Lucene.scanSeqNosInReader(reader, 0, Long.MAX_VALUE, n -> maxSeqNoFromDocs.set(Math.max(n, maxSeqNoFromDocs.get())));
1009+
assertThat(Long.parseLong(commit.getUserData().get(SequenceNumbers.MAX_SEQ_NO)),
1010+
greaterThanOrEqualTo(maxSeqNoFromDocs.get()));
1011+
}
1012+
}
1013+
}
1014+
9951015
public static MapperService createMapperService(String type) throws IOException {
9961016
IndexMetaData indexMetaData = IndexMetaData.builder("test")
9971017
.settings(Settings.builder()

‎test/framework/src/main/java/org/elasticsearch/test/ESTestCase.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -888,7 +888,7 @@ public static boolean terminate(ExecutorService... services) throws InterruptedE
888888
return terminated;
889889
}
890890

891-
public static boolean terminate(ThreadPool threadPool) throws InterruptedException {
891+
public static boolean terminate(ThreadPool threadPool) {
892892
return ThreadPool.terminate(threadPool, 10, TimeUnit.SECONDS);
893893
}
894894

‎test/framework/src/main/java/org/elasticsearch/transport/AbstractSimpleTransportTestCase.java

Lines changed: 1 addition & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -197,13 +197,7 @@ public void tearDown() throws Exception {
197197
assertNoPendingHandshakes(serviceA.getOriginalTransport());
198198
assertNoPendingHandshakes(serviceB.getOriginalTransport());
199199
} finally {
200-
IOUtils.close(serviceA, serviceB, () -> {
201-
try {
202-
terminate(threadPool);
203-
} catch (InterruptedException e) {
204-
Thread.currentThread().interrupt();
205-
}
206-
});
200+
IOUtils.close(serviceA, serviceB, () -> terminate(threadPool));
207201
}
208202
}
209203

‎x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/index/engine/FollowingEngine.java

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -70,6 +70,7 @@ private void preFlight(final Operation operation) {
7070
@Override
7171
protected InternalEngine.IndexingStrategy indexingStrategyForOperation(final Index index) throws IOException {
7272
preFlight(index);
73+
markSeqNoAsSeen(index.seqNo());
7374
// NOTES: refer Engine#getMaxSeqNoOfUpdatesOrDeletes for the explanation of the optimization using sequence numbers.
7475
final long maxSeqNoOfUpdatesOrDeletes = getMaxSeqNoOfUpdatesOrDeletes();
7576
assert maxSeqNoOfUpdatesOrDeletes != SequenceNumbers.UNASSIGNED_SEQ_NO : "max_seq_no_of_updates is not initialized";
@@ -105,6 +106,7 @@ protected InternalEngine.IndexingStrategy indexingStrategyForOperation(final Ind
105106
@Override
106107
protected InternalEngine.DeletionStrategy deletionStrategyForOperation(final Delete delete) throws IOException {
107108
preFlight(delete);
109+
markSeqNoAsSeen(delete.seqNo());
108110
if (delete.origin() == Operation.Origin.PRIMARY && hasBeenProcessedBefore(delete)) {
109111
// See the comment in #indexingStrategyForOperation for the explanation why we can safely skip this operation.
110112
final AlreadyProcessedFollowingEngineException error = new AlreadyProcessedFollowingEngineException(

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

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,8 @@
5858
import java.util.stream.Collectors;
5959

6060
import static org.elasticsearch.index.engine.EngineTestCase.getDocIds;
61+
import static org.elasticsearch.index.engine.EngineTestCase.getTranslog;
62+
import static org.elasticsearch.index.engine.EngineTestCase.getTranslog;
6163
import static org.hamcrest.Matchers.containsString;
6264
import static org.hamcrest.Matchers.equalTo;
6365
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
@@ -640,4 +642,50 @@ public void testProcessOnceOnPrimary() throws Exception {
640642
}
641643
}
642644
}
645+
646+
public void testMaxSeqNoInCommitUserData() throws Exception {
647+
final Settings settings = Settings.builder().put("index.number_of_shards", 1).put("index.number_of_replicas", 0)
648+
.put("index.version.created", Version.CURRENT).put("index.xpack.ccr.following_index", true)
649+
.put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true).build();
650+
final IndexMetaData indexMetaData = IndexMetaData.builder(index.getName()).settings(settings).build();
651+
final IndexSettings indexSettings = new IndexSettings(indexMetaData, settings);
652+
try (Store store = createStore(shardId, indexSettings, newDirectory())) {
653+
final EngineConfig engineConfig = engineConfig(shardId, indexSettings, threadPool, store, logger, xContentRegistry());
654+
try (FollowingEngine engine = createEngine(store, engineConfig)) {
655+
AtomicBoolean running = new AtomicBoolean(true);
656+
Thread rollTranslog = new Thread(() -> {
657+
while (running.get() && getTranslog(engine).currentFileGeneration() < 500) {
658+
engine.rollTranslogGeneration(); // make adding operations to translog slower
659+
}
660+
});
661+
rollTranslog.start();
662+
663+
Thread indexing = new Thread(() -> {
664+
List<Engine.Operation> ops = EngineTestCase.generateSingleDocHistory(
665+
true, VersionType.EXTERNAL, false, 2, 50, 500, "id");
666+
engine.advanceMaxSeqNoOfUpdatesOrDeletes(ops.stream().mapToLong(Engine.Operation::seqNo).max().getAsLong());
667+
for (Engine.Operation op : ops) {
668+
if (running.get() == false) {
669+
return;
670+
}
671+
try {
672+
EngineTestCase.applyOperation(engine, op);
673+
} catch (IOException e) {
674+
throw new AssertionError(e);
675+
}
676+
}
677+
});
678+
indexing.start();
679+
680+
int numCommits = between(5, 20);
681+
for (int i = 0; i < numCommits; i++) {
682+
engine.flush(false, true);
683+
}
684+
running.set(false);
685+
indexing.join();
686+
rollTranslog.join();
687+
EngineTestCase.assertMaxSeqNoInCommitUserData(engine);
688+
}
689+
}
690+
}
643691
}

0 commit comments

Comments
 (0)