Skip to content

Commit a32705a

Browse files
dnhatndnhatn
committed
Add retention leases replication tests (#38857)
This commit introduces the retention leases to ESIndexLevelReplicationTestCase, then adds some tests verifying that the retention leases replication works correctly in spite of the presence of the primary failover or out of order delivery of retention leases sync requests.
1 parent 1f1deb1 commit a32705a

File tree

8 files changed

+272
-44
lines changed

8 files changed

+272
-44
lines changed

‎server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3130,4 +3130,8 @@ public void advanceMaxSeqNoOfUpdatesOrDeletes(long seqNo) {
31303130
public void verifyShardBeforeIndexClosing() throws IllegalStateException {
31313131
getEngine().verifyEngineBeforeIndexClosing();
31323132
}
3133+
3134+
RetentionLeaseSyncer getRetentionLeaseSyncer() {
3135+
return retentionLeaseSyncer;
3136+
}
31333137
}

‎server/src/test/java/org/elasticsearch/index/replication/RetentionLeasesReplicationTests.java

Lines changed: 151 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,151 @@
1+
/*
2+
* Licensed to Elasticsearch under one or more contributor
3+
* license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright
5+
* ownership. Elasticsearch licenses this file to you under
6+
* the Apache License, Version 2.0 (the "License"); you may
7+
* not use this file except in compliance with the License.
8+
* You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.elasticsearch.index.replication;
21+
22+
import org.elasticsearch.action.ActionListener;
23+
import org.elasticsearch.action.support.PlainActionFuture;
24+
import org.elasticsearch.action.support.replication.ReplicationResponse;
25+
import org.elasticsearch.cluster.metadata.IndexMetaData;
26+
import org.elasticsearch.common.Randomness;
27+
import org.elasticsearch.common.settings.Settings;
28+
import org.elasticsearch.index.IndexSettings;
29+
import org.elasticsearch.index.seqno.RetentionLease;
30+
import org.elasticsearch.index.seqno.RetentionLeaseSyncAction;
31+
import org.elasticsearch.index.seqno.RetentionLeases;
32+
import org.elasticsearch.index.shard.IndexShard;
33+
import org.elasticsearch.index.shard.ShardId;
34+
35+
import java.util.ArrayList;
36+
import java.util.List;
37+
import java.util.concurrent.CountDownLatch;
38+
39+
import static org.hamcrest.Matchers.containsInAnyOrder;
40+
import static org.hamcrest.Matchers.equalTo;
41+
import static org.hamcrest.Matchers.hasSize;
42+
43+
public class RetentionLeasesReplicationTests extends ESIndexLevelReplicationTestCase {
44+
45+
public void testSimpleSyncRetentionLeases() throws Exception {
46+
Settings settings = Settings.builder().put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true).build();
47+
try (ReplicationGroup group = createGroup(between(0, 2), settings)) {
48+
group.startAll();
49+
List<RetentionLease> leases = new ArrayList<>();
50+
int iterations = between(1, 100);
51+
CountDownLatch latch = new CountDownLatch(iterations);
52+
for (int i = 0; i < iterations; i++) {
53+
if (leases.isEmpty() == false && rarely()) {
54+
RetentionLease leaseToRemove = randomFrom(leases);
55+
leases.remove(leaseToRemove);
56+
group.removeRetentionLease(leaseToRemove.id(), ActionListener.wrap(latch::countDown));
57+
} else {
58+
RetentionLease newLease = group.addRetentionLease(Integer.toString(i), randomNonNegativeLong(), "test-" + i,
59+
ActionListener.wrap(latch::countDown));
60+
leases.add(newLease);
61+
}
62+
}
63+
RetentionLeases leasesOnPrimary = group.getPrimary().getRetentionLeases();
64+
assertThat(leasesOnPrimary.version(), equalTo((long) iterations));
65+
assertThat(leasesOnPrimary.primaryTerm(), equalTo(group.getPrimary().getOperationPrimaryTerm()));
66+
assertThat(leasesOnPrimary.leases(), containsInAnyOrder(leases.toArray(new RetentionLease[0])));
67+
latch.await();
68+
for (IndexShard replica : group.getReplicas()) {
69+
assertThat(replica.getRetentionLeases(), equalTo(leasesOnPrimary));
70+
}
71+
}
72+
}
73+
74+
public void testOutOfOrderRetentionLeasesRequests() throws Exception {
75+
Settings settings = Settings.builder().put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true).build();
76+
int numberOfReplicas = between(1, 2);
77+
IndexMetaData indexMetaData = buildIndexMetaData(numberOfReplicas, settings, indexMapping);
78+
try (ReplicationGroup group = new ReplicationGroup(indexMetaData) {
79+
@Override
80+
protected void syncRetentionLeases(ShardId shardId, RetentionLeases leases, ActionListener<ReplicationResponse> listener) {
81+
listener.onResponse(new SyncRetentionLeasesResponse(new RetentionLeaseSyncAction.Request(shardId, leases)));
82+
}
83+
}) {
84+
group.startAll();
85+
int numLeases = between(1, 10);
86+
List<RetentionLeaseSyncAction.Request> requests = new ArrayList<>();
87+
for (int i = 0; i < numLeases; i++) {
88+
PlainActionFuture<ReplicationResponse> future = new PlainActionFuture<>();
89+
group.addRetentionLease(Integer.toString(i), randomNonNegativeLong(), "test-" + i, future);
90+
requests.add(((SyncRetentionLeasesResponse) future.actionGet()).syncRequest);
91+
}
92+
RetentionLeases leasesOnPrimary = group.getPrimary().getRetentionLeases();
93+
for (IndexShard replica : group.getReplicas()) {
94+
Randomness.shuffle(requests);
95+
requests.forEach(request -> group.executeRetentionLeasesSyncRequestOnReplica(request, replica));
96+
assertThat(replica.getRetentionLeases(), equalTo(leasesOnPrimary));
97+
}
98+
}
99+
}
100+
101+
public void testSyncRetentionLeasesWithPrimaryPromotion() throws Exception {
102+
Settings settings = Settings.builder().put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true).build();
103+
int numberOfReplicas = between(2, 4);
104+
IndexMetaData indexMetaData = buildIndexMetaData(numberOfReplicas, settings, indexMapping);
105+
try (ReplicationGroup group = new ReplicationGroup(indexMetaData) {
106+
@Override
107+
protected void syncRetentionLeases(ShardId shardId, RetentionLeases leases, ActionListener<ReplicationResponse> listener) {
108+
listener.onResponse(new SyncRetentionLeasesResponse(new RetentionLeaseSyncAction.Request(shardId, leases)));
109+
}
110+
}) {
111+
group.startAll();
112+
int numLeases = between(1, 100);
113+
IndexShard newPrimary = randomFrom(group.getReplicas());
114+
RetentionLeases latestRetentionLeasesOnNewPrimary = RetentionLeases.EMPTY;
115+
for (int i = 0; i < numLeases; i++) {
116+
PlainActionFuture<ReplicationResponse> addLeaseFuture = new PlainActionFuture<>();
117+
group.addRetentionLease(Integer.toString(i), randomNonNegativeLong(), "test-" + i, addLeaseFuture);
118+
RetentionLeaseSyncAction.Request request = ((SyncRetentionLeasesResponse) addLeaseFuture.actionGet()).syncRequest;
119+
for (IndexShard replica : randomSubsetOf(group.getReplicas())) {
120+
group.executeRetentionLeasesSyncRequestOnReplica(request, replica);
121+
if (newPrimary == replica) {
122+
latestRetentionLeasesOnNewPrimary = request.getRetentionLeases();
123+
}
124+
}
125+
}
126+
group.promoteReplicaToPrimary(newPrimary).get();
127+
// we need to make changes to retention leases to sync it to replicas
128+
// since we don't sync retention leases when promoting a new primary.
129+
PlainActionFuture<ReplicationResponse> newLeaseFuture = new PlainActionFuture<>();
130+
group.addRetentionLease("new-lease-after-promotion", randomNonNegativeLong(), "test", newLeaseFuture);
131+
RetentionLeases leasesOnPrimary = group.getPrimary().getRetentionLeases();
132+
assertThat(leasesOnPrimary.primaryTerm(), equalTo(group.getPrimary().getOperationPrimaryTerm()));
133+
assertThat(leasesOnPrimary.version(), equalTo(latestRetentionLeasesOnNewPrimary.version() + 1L));
134+
assertThat(leasesOnPrimary.leases(), hasSize(latestRetentionLeasesOnNewPrimary.leases().size() + 1));
135+
RetentionLeaseSyncAction.Request request = ((SyncRetentionLeasesResponse) newLeaseFuture.actionGet()).syncRequest;
136+
for (IndexShard replica : group.getReplicas()) {
137+
group.executeRetentionLeasesSyncRequestOnReplica(request, replica);
138+
}
139+
for (IndexShard replica : group.getReplicas()) {
140+
assertThat(replica.getRetentionLeases(), equalTo(leasesOnPrimary));
141+
}
142+
}
143+
}
144+
145+
static final class SyncRetentionLeasesResponse extends ReplicationResponse {
146+
final RetentionLeaseSyncAction.Request syncRequest;
147+
SyncRetentionLeasesResponse(RetentionLeaseSyncAction.Request syncRequest) {
148+
this.syncRequest = syncRequest;
149+
}
150+
}
151+
}

‎server/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java

Lines changed: 24 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -102,6 +102,7 @@
102102
import org.elasticsearch.index.mapper.SourceToParse;
103103
import org.elasticsearch.index.mapper.Uid;
104104
import org.elasticsearch.index.mapper.VersionFieldMapper;
105+
import org.elasticsearch.index.seqno.RetentionLeaseSyncer;
105106
import org.elasticsearch.index.seqno.RetentionLeases;
106107
import org.elasticsearch.index.seqno.SeqNoStats;
107108
import org.elasticsearch.index.seqno.SequenceNumbers;
@@ -1055,8 +1056,8 @@ public void testGlobalCheckpointSync() throws IOException {
10551056
final IndexMetaData.Builder indexMetadata =
10561057
IndexMetaData.builder(shardRouting.getIndexName()).settings(settings).primaryTerm(0, 1);
10571058
final AtomicBoolean synced = new AtomicBoolean();
1058-
final IndexShard primaryShard =
1059-
newShard(shardRouting, indexMetadata.build(), null, new InternalEngineFactory(), () -> synced.set(true));
1059+
final IndexShard primaryShard = newShard(
1060+
shardRouting, indexMetadata.build(), null, new InternalEngineFactory(), () -> synced.set(true), RetentionLeaseSyncer.EMPTY);
10601061
// add a replica
10611062
recoverShardFromStore(primaryShard);
10621063
final IndexShard replicaShard = newShard(shardId, false);
@@ -1471,9 +1472,8 @@ public String[] listAll() throws IOException {
14711472
};
14721473

14731474
try (Store store = createStore(shardId, new IndexSettings(metaData, Settings.EMPTY), directory)) {
1474-
IndexShard shard = newShard(shardRouting, shardPath, metaData, i -> store,
1475-
null, new InternalEngineFactory(), () -> {
1476-
}, EMPTY_EVENT_LISTENER);
1475+
IndexShard shard = newShard(shardRouting, shardPath, metaData, i -> store, null, new InternalEngineFactory(),
1476+
() -> { }, RetentionLeaseSyncer.EMPTY, EMPTY_EVENT_LISTENER);
14771477
AtomicBoolean failureCallbackTriggered = new AtomicBoolean(false);
14781478
shard.addShardFailureCallback((ig)->failureCallbackTriggered.set(true));
14791479

@@ -2131,6 +2131,7 @@ public void testRecoverFromStoreRemoveStaleOperations() throws Exception {
21312131
null,
21322132
shard.getEngineFactory(),
21332133
shard.getGlobalCheckpointSyncer(),
2134+
shard.getRetentionLeaseSyncer(),
21342135
EMPTY_EVENT_LISTENER);
21352136
DiscoveryNode localNode = new DiscoveryNode("foo", buildNewFakeTransportAddress(), emptyMap(), emptySet(), Version.CURRENT);
21362137
newShard.markAsRecovering("store", new RecoveryState(newShard.routingEntry(), localNode, null));
@@ -2250,6 +2251,7 @@ public IndexSearcher wrap(IndexSearcher searcher) throws EngineException {
22502251
wrapper,
22512252
new InternalEngineFactory(),
22522253
() -> {},
2254+
RetentionLeaseSyncer.EMPTY,
22532255
EMPTY_EVENT_LISTENER);
22542256

22552257
recoverShardFromStore(newShard);
@@ -2403,6 +2405,7 @@ public IndexSearcher wrap(IndexSearcher searcher) throws EngineException {
24032405
wrapper,
24042406
new InternalEngineFactory(),
24052407
() -> {},
2408+
RetentionLeaseSyncer.EMPTY,
24062409
EMPTY_EVENT_LISTENER);
24072410

24082411
recoverShardFromStore(newShard);
@@ -2946,9 +2949,8 @@ public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IO
29462949
.put(IndexSettings.INDEX_CHECK_ON_STARTUP.getKey(), randomFrom("true", "checksum")))
29472950
.build();
29482951

2949-
IndexShard corruptedShard = newShard(shardRouting, shardPath, indexMetaData,
2950-
null, null, indexShard.engineFactory,
2951-
indexShard.getGlobalCheckpointSyncer(), EMPTY_EVENT_LISTENER);
2952+
IndexShard corruptedShard = newShard(shardRouting, shardPath, indexMetaData, null, null, indexShard.engineFactory,
2953+
indexShard.getGlobalCheckpointSyncer(), indexShard.getRetentionLeaseSyncer(), EMPTY_EVENT_LISTENER);
29522954

29532955
final IndexShardRecoveryException indexShardRecoveryException =
29542956
expectThrows(IndexShardRecoveryException.class, () -> newStartedShard(p -> corruptedShard, true));
@@ -2991,9 +2993,8 @@ public void testShardDoesNotStartIfCorruptedMarkerIsPresent() throws Exception {
29912993
}
29922994

29932995
// try to start shard on corrupted files
2994-
final IndexShard corruptedShard = newShard(shardRouting, shardPath, indexMetaData,
2995-
null, null, indexShard.engineFactory,
2996-
indexShard.getGlobalCheckpointSyncer(), EMPTY_EVENT_LISTENER);
2996+
final IndexShard corruptedShard = newShard(shardRouting, shardPath, indexMetaData, null, null, indexShard.engineFactory,
2997+
indexShard.getGlobalCheckpointSyncer(), indexShard.getRetentionLeaseSyncer(), EMPTY_EVENT_LISTENER);
29972998

29982999
final IndexShardRecoveryException exception1 = expectThrows(IndexShardRecoveryException.class,
29993000
() -> newStartedShard(p -> corruptedShard, true));
@@ -3014,9 +3015,8 @@ public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IO
30143015
assertThat("store has to be marked as corrupted", corruptedMarkerCount.get(), equalTo(1));
30153016

30163017
// try to start another time shard on corrupted files
3017-
final IndexShard corruptedShard2 = newShard(shardRouting, shardPath, indexMetaData,
3018-
null, null, indexShard.engineFactory,
3019-
indexShard.getGlobalCheckpointSyncer(), EMPTY_EVENT_LISTENER);
3018+
final IndexShard corruptedShard2 = newShard(shardRouting, shardPath, indexMetaData, null, null, indexShard.engineFactory,
3019+
indexShard.getGlobalCheckpointSyncer(), indexShard.getRetentionLeaseSyncer(), EMPTY_EVENT_LISTENER);
30203020

30213021
final IndexShardRecoveryException exception2 = expectThrows(IndexShardRecoveryException.class,
30223022
() -> newStartedShard(p -> corruptedShard2, true));
@@ -3054,9 +3054,8 @@ public void testReadSnapshotAndCheckIndexConcurrently() throws Exception {
30543054
.put(indexShard.indexSettings.getSettings())
30553055
.put(IndexSettings.INDEX_CHECK_ON_STARTUP.getKey(), randomFrom("false", "true", "checksum")))
30563056
.build();
3057-
final IndexShard newShard = newShard(shardRouting, indexShard.shardPath(), indexMetaData,
3058-
null, null, indexShard.engineFactory,
3059-
indexShard.getGlobalCheckpointSyncer(), EMPTY_EVENT_LISTENER);
3057+
final IndexShard newShard = newShard(shardRouting, indexShard.shardPath(), indexMetaData, null, null, indexShard.engineFactory,
3058+
indexShard.getGlobalCheckpointSyncer(), indexShard.getRetentionLeaseSyncer(), EMPTY_EVENT_LISTENER);
30603059

30613060
Store.MetadataSnapshot storeFileMetaDatas = newShard.snapshotStoreMetadata();
30623061
assertTrue("at least 2 files, commit and data: " + storeFileMetaDatas.toString(), storeFileMetaDatas.size() > 1);
@@ -3436,15 +3435,14 @@ public void testFlushOnInactive() throws Exception {
34363435
ShardPath shardPath = new ShardPath(false, nodePath.resolve(shardId), nodePath.resolve(shardId), shardId);
34373436
AtomicBoolean markedInactive = new AtomicBoolean();
34383437
AtomicReference<IndexShard> primaryRef = new AtomicReference<>();
3439-
IndexShard primary = newShard(shardRouting, shardPath, metaData, null, null,
3440-
new InternalEngineFactory(), () -> {
3441-
}, new IndexEventListener() {
3442-
@Override
3443-
public void onShardInactive(IndexShard indexShard) {
3444-
markedInactive.set(true);
3445-
primaryRef.get().flush(new FlushRequest());
3446-
}
3447-
});
3438+
IndexShard primary = newShard(shardRouting, shardPath, metaData, null, null, new InternalEngineFactory(), () -> { },
3439+
RetentionLeaseSyncer.EMPTY, new IndexEventListener() {
3440+
@Override
3441+
public void onShardInactive(IndexShard indexShard) {
3442+
markedInactive.set(true);
3443+
primaryRef.get().flush(new FlushRequest());
3444+
}
3445+
});
34483446
primaryRef.set(primary);
34493447
recoverShardFromStore(primary);
34503448
for (int i = 0; i < 3; i++) {

‎server/src/test/java/org/elasticsearch/index/shard/RemoveCorruptedShardDataCommandTests.java

Lines changed: 5 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@
4040
import org.elasticsearch.index.MergePolicyConfig;
4141
import org.elasticsearch.index.engine.EngineException;
4242
import org.elasticsearch.index.engine.InternalEngineFactory;
43+
import org.elasticsearch.index.seqno.RetentionLeaseSyncer;
4344
import org.elasticsearch.index.store.Store;
4445
import org.elasticsearch.index.translog.TestTranslog;
4546
import org.elasticsearch.index.translog.TranslogCorruptedException;
@@ -107,11 +108,8 @@ public void setup() throws IOException {
107108
.putMapping("_doc", "{ \"properties\": {} }");
108109
indexMetaData = metaData.build();
109110

110-
indexShard = newStartedShard(p ->
111-
newShard(routing, shardPath, indexMetaData, null, null,
112-
new InternalEngineFactory(), () -> {
113-
}, EMPTY_EVENT_LISTENER),
114-
true);
111+
indexShard = newStartedShard(p -> newShard(routing, shardPath, indexMetaData, null, null,
112+
new InternalEngineFactory(), () -> { }, RetentionLeaseSyncer.EMPTY, EMPTY_EVENT_LISTENER), true);
115113

116114
translogPath = shardPath.resolveTranslog();
117115
indexPath = shardPath.resolveIndex();
@@ -371,8 +369,8 @@ private IndexShard reopenIndexShard(boolean corrupted) throws IOException {
371369
return new Store(shardId, indexSettings, baseDirectoryWrapper, new DummyShardLock(shardId));
372370
};
373371

374-
return newShard(shardRouting, shardPath, metaData, storeProvider, null,
375-
indexShard.engineFactory, indexShard.getGlobalCheckpointSyncer(), EMPTY_EVENT_LISTENER);
372+
return newShard(shardRouting, shardPath, metaData, storeProvider, null, indexShard.engineFactory,
373+
indexShard.getGlobalCheckpointSyncer(), indexShard.getRetentionLeaseSyncer(), EMPTY_EVENT_LISTENER);
376374
}
377375

378376
private int indexDocs(IndexShard indexShard, boolean flushLast) throws IOException {

‎server/src/test/java/org/elasticsearch/repositories/blobstore/BlobStoreRepositoryRestoreTests.java

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
import org.elasticsearch.env.Environment;
3232
import org.elasticsearch.env.TestEnvironment;
3333
import org.elasticsearch.index.engine.InternalEngineFactory;
34+
import org.elasticsearch.index.seqno.RetentionLeaseSyncer;
3435
import org.elasticsearch.index.shard.IndexShard;
3536
import org.elasticsearch.index.shard.IndexShardState;
3637
import org.elasticsearch.index.shard.IndexShardTestCase;
@@ -109,6 +110,7 @@ public void testRestoreSnapshotWithExistingFiles() throws IOException {
109110
null,
110111
new InternalEngineFactory(),
111112
() -> {},
113+
RetentionLeaseSyncer.EMPTY,
112114
EMPTY_EVENT_LISTENER);
113115

114116
// restore the shard

0 commit comments

Comments
 (0)