Skip to content

Commit d458ada

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 a3d2310 commit d458ada

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
@@ -1020,6 +1020,7 @@ assert config().getIndexSettings().getIndexVersionCreated().before(Version.V_6_0
10201020
}
10211021
}
10221022
}
1023+
markSeqNoAsSeen(index.seqNo());
10231024
return plan;
10241025
}
10251026

@@ -1387,6 +1388,7 @@ assert config().getIndexSettings().getIndexVersionCreated().before(Version.V_6_0
13871388
delete.seqNo(), delete.version());
13881389
}
13891390
}
1391+
markSeqNoAsSeen(delete.seqNo());
13901392
return plan;
13911393
}
13921394

@@ -1541,6 +1543,7 @@ public void maybePruneDeletes() {
15411543
public NoOpResult noOp(final NoOp noOp) {
15421544
NoOpResult noOpResult;
15431545
try (ReleasableLock ignored = readLock.acquire()) {
1546+
markSeqNoAsSeen(noOp.seqNo());
15441547
noOpResult = innerNoOp(noOp);
15451548
} catch (final Exception e) {
15461549
noOpResult = new NoOpResult(getPrimaryTerm(), noOp.seqNo(), e);
@@ -2520,6 +2523,13 @@ public void waitForOpsToComplete(long seqNo) throws InterruptedException {
25202523
localCheckpointTracker.waitForOpsToComplete(seqNo);
25212524
}
25222525

2526+
/**
2527+
* Marks the given seq_no as seen and advances the max_seq_no of this engine to at least that value.
2528+
*/
2529+
protected final void markSeqNoAsSeen(long seqNo) {
2530+
localCheckpointTracker.advanceMaxSeqNo(seqNo);
2531+
}
2532+
25232533
/**
25242534
* Checks if the given operation has been processed in this engine or not.
25252535
* @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
@@ -5655,4 +5655,42 @@ void assertLuceneOperations(InternalEngine engine, long expectedAppends, long ex
56555655
assertThat(message, engine.getNumDocUpdates(), equalTo(expectedUpdates));
56565656
assertThat(message, engine.getNumDocDeletes(), equalTo(expectedDeletes));
56575657
}
5658+
5659+
public void testMaxSeqNoInCommitUserData() throws Exception {
5660+
AtomicBoolean running = new AtomicBoolean(true);
5661+
Thread rollTranslog = new Thread(() -> {
5662+
while (running.get() && engine.getTranslog().currentFileGeneration() < 500) {
5663+
engine.rollTranslogGeneration(); // make adding operations to translog slower
5664+
}
5665+
});
5666+
rollTranslog.start();
5667+
5668+
Thread indexing = new Thread(() -> {
5669+
long seqNo = 0;
5670+
while (running.get() && seqNo <= 1000) {
5671+
try {
5672+
String id = Long.toString(between(1, 50));
5673+
if (randomBoolean()) {
5674+
ParsedDocument doc = testParsedDocument(id, null, testDocumentWithTextField(), SOURCE, null);
5675+
engine.index(replicaIndexForDoc(doc, 1L, seqNo, false));
5676+
} else {
5677+
engine.delete(replicaDeleteForDoc(id, 1L, seqNo, 0L));
5678+
}
5679+
seqNo++;
5680+
} catch (IOException e) {
5681+
throw new AssertionError(e);
5682+
}
5683+
}
5684+
});
5685+
indexing.start();
5686+
5687+
int numCommits = between(5, 20);
5688+
for (int i = 0; i < numCommits; i++) {
5689+
engine.flush(false, true);
5690+
}
5691+
running.set(false);
5692+
indexing.join();
5693+
rollTranslog.join();
5694+
assertMaxSeqNoInCommitUserData(engine);
5695+
}
56585696
}

‎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;
@@ -126,6 +128,7 @@
126128
import static org.elasticsearch.index.engine.Engine.Operation.Origin.REPLICA;
127129
import static org.elasticsearch.index.translog.TranslogDeletionPolicies.createTranslogDeletionPolicy;
128130
import static org.hamcrest.Matchers.equalTo;
131+
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
129132
import static org.hamcrest.Matchers.notNullValue;
130133

131134
public abstract class EngineTestCase extends ESTestCase {
@@ -254,18 +257,20 @@ public EngineConfig copy(EngineConfig config, MergePolicy mergePolicy) {
254257
@After
255258
public void tearDown() throws Exception {
256259
super.tearDown();
257-
if (engine != null && engine.isClosed.get() == false) {
258-
engine.getTranslog().getDeletionPolicy().assertNoOpenTranslogRefs();
259-
assertConsistentHistoryBetweenTranslogAndLuceneIndex(engine, createMapperService("test"));
260-
}
261-
if (replicaEngine != null && replicaEngine.isClosed.get() == false) {
262-
replicaEngine.getTranslog().getDeletionPolicy().assertNoOpenTranslogRefs();
263-
assertConsistentHistoryBetweenTranslogAndLuceneIndex(replicaEngine, createMapperService("test"));
260+
try {
261+
if (engine != null && engine.isClosed.get() == false) {
262+
engine.getTranslog().getDeletionPolicy().assertNoOpenTranslogRefs();
263+
assertConsistentHistoryBetweenTranslogAndLuceneIndex(engine, createMapperService("test"));
264+
assertMaxSeqNoInCommitUserData(engine);
265+
}
266+
if (replicaEngine != null && replicaEngine.isClosed.get() == false) {
267+
replicaEngine.getTranslog().getDeletionPolicy().assertNoOpenTranslogRefs();
268+
assertConsistentHistoryBetweenTranslogAndLuceneIndex(replicaEngine, createMapperService("test"));
269+
assertMaxSeqNoInCommitUserData(replicaEngine);
270+
}
271+
} finally {
272+
IOUtils.close(replicaEngine, storeReplica, engine, store, () -> terminate(threadPool));
264273
}
265-
IOUtils.close(
266-
replicaEngine, storeReplica,
267-
engine, store);
268-
terminate(threadPool);
269274
}
270275

271276

@@ -1079,6 +1084,21 @@ public static void assertConsistentHistoryBetweenTranslogAndLuceneIndex(Engine e
10791084
}
10801085
}
10811086

1087+
/**
1088+
* 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.
1089+
*/
1090+
public static void assertMaxSeqNoInCommitUserData(Engine engine) throws Exception {
1091+
List<IndexCommit> commits = DirectoryReader.listCommits(engine.store.directory());
1092+
for (IndexCommit commit : commits) {
1093+
try (DirectoryReader reader = DirectoryReader.open(commit)) {
1094+
AtomicLong maxSeqNoFromDocs = new AtomicLong(SequenceNumbers.NO_OPS_PERFORMED);
1095+
Lucene.scanSeqNosInReader(reader, 0, Long.MAX_VALUE, n -> maxSeqNoFromDocs.set(Math.max(n, maxSeqNoFromDocs.get())));
1096+
assertThat(Long.parseLong(commit.getUserData().get(SequenceNumbers.MAX_SEQ_NO)),
1097+
greaterThanOrEqualTo(maxSeqNoFromDocs.get()));
1098+
}
1099+
}
1100+
}
1101+
10821102
public static MapperService createMapperService(String type) throws IOException {
10831103
IndexMetaData indexMetaData = IndexMetaData.builder("test")
10841104
.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
@@ -68,6 +68,7 @@ private void preFlight(final Operation operation) {
6868
@Override
6969
protected InternalEngine.IndexingStrategy indexingStrategyForOperation(final Index index) throws IOException {
7070
preFlight(index);
71+
markSeqNoAsSeen(index.seqNo());
7172
// NOTES: refer Engine#getMaxSeqNoOfUpdatesOrDeletes for the explanation of the optimization using sequence numbers.
7273
final long maxSeqNoOfUpdatesOrDeletes = getMaxSeqNoOfUpdatesOrDeletes();
7374
assert maxSeqNoOfUpdatesOrDeletes != SequenceNumbers.UNASSIGNED_SEQ_NO : "max_seq_no_of_updates is not initialized";
@@ -103,6 +104,7 @@ protected InternalEngine.IndexingStrategy indexingStrategyForOperation(final Ind
103104
@Override
104105
protected InternalEngine.DeletionStrategy deletionStrategyForOperation(final Delete delete) throws IOException {
105106
preFlight(delete);
107+
markSeqNoAsSeen(delete.seqNo());
106108
if (delete.origin() == Operation.Origin.PRIMARY && hasBeenProcessedBefore(delete)) {
107109
// See the comment in #indexingStrategyForOperation for the explanation why we can safely skip this operation.
108110
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
@@ -59,6 +59,8 @@
5959
import java.util.stream.Collectors;
6060

6161
import static org.elasticsearch.index.engine.EngineTestCase.getDocIds;
62+
import static org.elasticsearch.index.engine.EngineTestCase.getTranslog;
63+
import static org.elasticsearch.index.engine.EngineTestCase.getTranslog;
6264
import static org.hamcrest.Matchers.containsString;
6365
import static org.hamcrest.Matchers.equalTo;
6466
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
@@ -660,4 +662,50 @@ public void testVerifyShardBeforeIndexClosingIsNoOp() throws IOException {
660662
}
661663
});
662664
}
665+
666+
public void testMaxSeqNoInCommitUserData() throws Exception {
667+
final Settings settings = Settings.builder().put("index.number_of_shards", 1).put("index.number_of_replicas", 0)
668+
.put("index.version.created", Version.CURRENT).put("index.xpack.ccr.following_index", true)
669+
.put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true).build();
670+
final IndexMetaData indexMetaData = IndexMetaData.builder(index.getName()).settings(settings).build();
671+
final IndexSettings indexSettings = new IndexSettings(indexMetaData, settings);
672+
try (Store store = createStore(shardId, indexSettings, newDirectory())) {
673+
final EngineConfig engineConfig = engineConfig(shardId, indexSettings, threadPool, store, logger, xContentRegistry());
674+
try (FollowingEngine engine = createEngine(store, engineConfig)) {
675+
AtomicBoolean running = new AtomicBoolean(true);
676+
Thread rollTranslog = new Thread(() -> {
677+
while (running.get() && getTranslog(engine).currentFileGeneration() < 500) {
678+
engine.rollTranslogGeneration(); // make adding operations to translog slower
679+
}
680+
});
681+
rollTranslog.start();
682+
683+
Thread indexing = new Thread(() -> {
684+
List<Engine.Operation> ops = EngineTestCase.generateSingleDocHistory(
685+
true, VersionType.EXTERNAL, false, 2, 50, 500, "id");
686+
engine.advanceMaxSeqNoOfUpdatesOrDeletes(ops.stream().mapToLong(Engine.Operation::seqNo).max().getAsLong());
687+
for (Engine.Operation op : ops) {
688+
if (running.get() == false) {
689+
return;
690+
}
691+
try {
692+
EngineTestCase.applyOperation(engine, op);
693+
} catch (IOException e) {
694+
throw new AssertionError(e);
695+
}
696+
}
697+
});
698+
indexing.start();
699+
700+
int numCommits = between(5, 20);
701+
for (int i = 0; i < numCommits; i++) {
702+
engine.flush(false, true);
703+
}
704+
running.set(false);
705+
indexing.join();
706+
rollTranslog.join();
707+
EngineTestCase.assertMaxSeqNoInCommitUserData(engine);
708+
}
709+
}
710+
}
663711
}

0 commit comments

Comments
 (0)