Skip to content

Commit 09313fd

Browse files
srikanthpadakantisrikanthpadakanti
authored
Merge branch 'main' into 20200-closed-index-reroute-fix
Signed-off-by: Srikanth Padakanti <srikanth29.9@gmail.com>
2 parents 8d41b62 + 8134f12 commit 09313fd

11 files changed

Lines changed: 162 additions & 35 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,12 +10,14 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
1010
- Add security policy to allow `accessUnixDomainSocket` in `transport-grpc` module ([#20463](https://github.com/opensearch-project/OpenSearch/pull/20463))
1111

1212
### Changed
13+
- Move Randomness from server to libs/common ([#20570](https://github.com/opensearch-project/OpenSearch/pull/20570))
1314

1415
### Fixed
1516
- Fix flaky test failures in ShardsLimitAllocationDeciderIT ([#20375](https://github.com/opensearch-project/OpenSearch/pull/20375))
1617
- Prevent criteria update for context aware indices ([#20250](https://github.com/opensearch-project/OpenSearch/pull/20250))
1718
- Update EncryptedBlobContainer to adhere limits while listing blobs in specific sort order if wrapped blob container supports ([#20514](https://github.com/opensearch-project/OpenSearch/pull/20514))
1819
- Shard routings for closed index are allocated again without opening the index ([#20558](https://github.com/opensearch-project/OpenSearch/pull/20558))
20+
- [segment replication] Fix segment replication infinite retry due to stale metadata checkpoint ([#20551](https://github.com/opensearch-project/OpenSearch/pull/20551))
1921
- Changing opensearch.cgroups.hierarchy.override causes java.lang.SecurityException exception ([#20565](https://github.com/opensearch-project/OpenSearch/pull/20565))
2022

2123
### Dependencies

‎distribution/tools/fips-demo-installer-cli/build.gradle‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ apply plugin: 'opensearch.build'
1212

1313
dependencies {
1414
api project(":libs:opensearch-cli")
15-
api project(":server")
15+
api project(":libs:opensearch-common")
1616
api project(':distribution:tools:java-version-checker')
1717
api "info.picocli:picocli:${versions.picocli}"
1818
api "org.bouncycastle:bc-fips:${versions.bouncycastle_jce}"

server/src/main/java/org/opensearch/common/Randomness.java renamed to libs/common/src/main/java/org/opensearch/common/Randomness.java

Lines changed: 0 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -32,9 +32,6 @@
3232

3333
package org.opensearch.common;
3434

35-
import org.opensearch.common.settings.Setting;
36-
import org.opensearch.common.settings.Settings;
37-
3835
import java.lang.reflect.Method;
3936
import java.security.GeneralSecurityException;
4037
import java.security.NoSuchAlgorithmException;
@@ -82,22 +79,6 @@ public final class Randomness {
8279

8380
private Randomness() {}
8481

85-
/**
86-
* Provides a reproducible source of randomness seeded by a long
87-
* seed in the settings with the key setting.
88-
*
89-
* @param settings the settings containing the seed
90-
* @param setting the setting to access the seed
91-
* @return a reproducible source of randomness
92-
*/
93-
public static Random get(Settings settings, Setting<Long> setting) {
94-
if (setting.exists(settings)) {
95-
return new Random(setting.get(settings));
96-
} else {
97-
return get();
98-
}
99-
}
100-
10182
/**
10283
* Provides a source of randomness that is reproducible when
10384
* running under the OpenSearch test suite, and otherwise

‎server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationIT.java‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,8 @@
6363
import org.opensearch.common.settings.Settings;
6464
import org.opensearch.common.unit.TimeValue;
6565
import org.opensearch.common.util.set.Sets;
66+
import org.opensearch.core.common.breaker.CircuitBreaker;
67+
import org.opensearch.core.common.breaker.CircuitBreakingException;
6668
import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
6769
import org.opensearch.core.concurrency.OpenSearchRejectedExecutionException;
6870
import org.opensearch.core.index.shard.ShardId;
@@ -77,6 +79,7 @@
7779
import org.opensearch.index.engine.EngineConfig;
7880
import org.opensearch.index.engine.NRTReplicationReaderManager;
7981
import org.opensearch.index.shard.IndexShard;
82+
import org.opensearch.index.store.StoreFileMetadata;
8083
import org.opensearch.indices.recovery.FileChunkRequest;
8184
import org.opensearch.indices.replication.checkpoint.PublishCheckpointAction;
8285
import org.opensearch.indices.replication.common.ReplicationType;
@@ -140,6 +143,87 @@ private static String indexOrAlias() {
140143
return randomBoolean() ? INDEX_NAME : "alias";
141144
}
142145

146+
public void testLocalSegmentReplicationWithException() throws Exception {
147+
// this test stubs transport calls specific to node-node replication.
148+
assumeFalse(
149+
"Skipping the test as its not compatible with segment replication with remote store.",
150+
segmentReplicationWithRemoteEnabled()
151+
);
152+
153+
final String primaryNode = internalCluster().startDataOnlyNode();
154+
createIndex(INDEX_NAME);
155+
ensureYellowAndNoInitializingShards(INDEX_NAME);
156+
final String replicaNode = internalCluster().startDataOnlyNode();
157+
ensureGreen(INDEX_NAME);
158+
159+
MockTransportService primaryTransportService = ((MockTransportService) internalCluster().getInstance(
160+
TransportService.class,
161+
primaryNode
162+
));
163+
164+
AtomicBoolean mockException = new AtomicBoolean(true);
165+
CountDownLatch latch1 = new CountDownLatch(1);
166+
CountDownLatch latch2 = new CountDownLatch(1);
167+
168+
primaryTransportService.addRequestHandlingBehavior(
169+
SegmentReplicationSourceService.Actions.GET_SEGMENT_FILES,
170+
(handler, request, channel, task) -> {
171+
logger.info(
172+
"replicationId {}, get segment files {}",
173+
((GetSegmentFilesRequest) request).getReplicationId(),
174+
((GetSegmentFilesRequest) request).getFilesToFetch().stream().map(StoreFileMetadata::name).collect(Collectors.toList())
175+
);
176+
if (mockException.get()) {
177+
mockException.set(false);
178+
latch1.countDown();
179+
latch2.await();
180+
throw new CircuitBreakingException("mock circuit break exception", CircuitBreaker.Durability.TRANSIENT);
181+
} else {
182+
handler.messageReceived(request, channel, task);
183+
}
184+
}
185+
);
186+
187+
// generate _0.si
188+
client().prepareIndex(INDEX_NAME)
189+
.setId(String.valueOf(1))
190+
.setSource("foo", "bar")
191+
.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
192+
.get();
193+
194+
latch1.await();
195+
196+
MockTransportService replicaTransportService = ((MockTransportService) internalCluster().getInstance(
197+
TransportService.class,
198+
replicaNode
199+
));
200+
replicaTransportService.addRequestHandlingBehavior(
201+
PublishCheckpointAction.ACTION_NAME + TransportReplicationAction.REPLICA_ACTION_SUFFIX,
202+
(handler, request, channel, task) -> {
203+
logger.info("replica receive publish checkpoint request");
204+
handler.messageReceived(request, channel, task);
205+
latch2.countDown();
206+
}
207+
);
208+
209+
// generate _1.si
210+
client().prepareIndex(INDEX_NAME)
211+
.setId(String.valueOf(2))
212+
.setSource("foo2", "bar2")
213+
.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
214+
.get();
215+
216+
waitForSearchableDocs(2, primaryNode, replicaNode);
217+
218+
client().prepareIndex(INDEX_NAME)
219+
.setId(String.valueOf(3))
220+
.setSource("foo3", "bar3")
221+
.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
222+
.get();
223+
224+
waitForSearchableDocs(3, primaryNode, replicaNode);
225+
}
226+
143227
public void testAcquireLastIndexCommit() throws Exception {
144228
final String primaryNode = internalCluster().startDataOnlyNode();
145229
createIndex(INDEX_NAME);

‎server/src/main/java/org/opensearch/env/NodeEnvironment.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -507,7 +507,7 @@ private static NodeMetadata loadNodeMetadata(Settings settings, Logger logger, N
507507
}
508508

509509
public static String generateNodeId(Settings settings) {
510-
Random random = Randomness.get(settings, NODE_ID_SEED_SETTING);
510+
Random random = NODE_ID_SEED_SETTING.exists(settings) ? new Random(NODE_ID_SEED_SETTING.get(settings)) : Randomness.get();
511511
return UUIDs.randomBase64UUID(random);
512512
}
513513

‎server/src/main/java/org/opensearch/indices/replication/AbstractSegmentReplicationTarget.java‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -47,17 +47,20 @@ public abstract class AbstractSegmentReplicationTarget extends ReplicationTarget
4747
protected final SegmentReplicationSource source;
4848
protected final SegmentReplicationState state;
4949
protected final MultiFileWriter multiFileWriter;
50+
protected final boolean isRetry;
5051

5152
public AbstractSegmentReplicationTarget(
5253
String name,
5354
IndexShard indexShard,
5455
ReplicationCheckpoint checkpoint,
5556
SegmentReplicationSource source,
57+
boolean isRetry,
5658
ReplicationListener listener
5759
) {
5860
super(name, indexShard, new ReplicationLuceneIndex(), listener);
5961
this.checkpoint = checkpoint;
6062
this.source = source;
63+
this.isRetry = isRetry;
6164
this.state = new SegmentReplicationState(
6265
indexShard.routingEntry(),
6366
stateIndex,
@@ -168,7 +171,10 @@ public void startReplication(ActionListener<Void> listener, BiConsumer<Replicati
168171
// from before a restart, and should accept the primary's current state even if it appears older.
169172
// See: https://github.com/opensearch-project/OpenSearch/issues/19234
170173
boolean isRecovering = indexShard.routingEntry().initializing() || indexShard.routingEntry().relocating();
171-
if (indexShard.indexSettings().isSegRepLocalEnabled() && checkpoint.isAheadOf(getMetadataCheckpoint) && !isRecovering) {
174+
if (indexShard.indexSettings().isSegRepLocalEnabled()
175+
&& checkpoint.isAheadOf(getMetadataCheckpoint)
176+
&& false == isRecovering
177+
&& false == isRetry) {
172178
// Fixes https://github.com/opensearch-project/OpenSearch/issues/18490
173179
listener.onFailure(
174180
new ReplicationFailedException(

‎server/src/main/java/org/opensearch/indices/replication/MergedSegmentReplicationTarget.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ public MergedSegmentReplicationTarget(
3333
SegmentReplicationSource source,
3434
ReplicationListener listener
3535
) {
36-
super("merged_segment_replication_target", indexShard, checkpoint, source, listener);
36+
super("merged_segment_replication_target", indexShard, checkpoint, source, false, listener);
3737
}
3838

3939
@Override

‎server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTarget.java‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,17 @@ public SegmentReplicationTarget(
4242
SegmentReplicationSource source,
4343
ReplicationListener listener
4444
) {
45-
super("replication_target", indexShard, checkpoint, source, listener);
45+
this(indexShard, checkpoint, source, false, listener);
46+
}
47+
48+
public SegmentReplicationTarget(
49+
IndexShard indexShard,
50+
ReplicationCheckpoint checkpoint,
51+
SegmentReplicationSource source,
52+
boolean isRetry,
53+
ReplicationListener listener
54+
) {
55+
super("replication_target", indexShard, checkpoint, source, isRetry, listener);
4656
}
4757

4858
@Override
@@ -128,6 +138,6 @@ protected void finalizeReplication(CheckpointInfoResponse checkpointInfoResponse
128138

129139
@Override
130140
public SegmentReplicationTarget retryCopy() {
131-
return new SegmentReplicationTarget(indexShard, checkpoint, source, listener);
141+
return new SegmentReplicationTarget(indexShard, checkpoint, source, isRetry, listener);
132142
}
133143
}

‎server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTargetService.java‎

Lines changed: 43 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -295,7 +295,23 @@ public SegmentReplicationTarget get(ShardId shardId) {
295295
* @param receivedCheckpoint received checkpoint that is checked for processing
296296
* @param replicaShard replica shard on which checkpoint is received
297297
*/
298-
public synchronized void onNewCheckpoint(final ReplicationCheckpoint receivedCheckpoint, final IndexShard replicaShard) {
298+
public void onNewCheckpoint(final ReplicationCheckpoint receivedCheckpoint, final IndexShard replicaShard) {
299+
onNewCheckpoint(receivedCheckpoint, replicaShard, false);
300+
}
301+
302+
/**
303+
* Invoked when a new checkpoint is received from a primary shard.
304+
* It checks if a new checkpoint should be processed or not and starts replication if needed.
305+
*
306+
* @param receivedCheckpoint received checkpoint that is checked for processing
307+
* @param replicaShard replica shard on which checkpoint is received
308+
* @param isRetry is it a retry after failure
309+
*/
310+
public synchronized void onNewCheckpoint(
311+
final ReplicationCheckpoint receivedCheckpoint,
312+
final IndexShard replicaShard,
313+
boolean isRetry
314+
) {
299315
logger.debug(() -> new ParameterizedMessage("Replica received new replication checkpoint from primary [{}]", receivedCheckpoint));
300316
// if the shard is in any state
301317
if (replicaShard.state().equals(IndexShardState.CLOSED)) {
@@ -332,7 +348,7 @@ public synchronized void onNewCheckpoint(final ReplicationCheckpoint receivedChe
332348
}
333349
final Thread thread = Thread.currentThread();
334350
if (replicaShard.shouldProcessCheckpoint(receivedCheckpoint)) {
335-
startReplication(replicaShard, receivedCheckpoint, new SegmentReplicationListener() {
351+
startReplication(replicaShard, receivedCheckpoint, isRetry, new SegmentReplicationListener() {
336352
@Override
337353
public void onReplicationDone(SegmentReplicationState state) {
338354
logger.debug(
@@ -366,7 +382,7 @@ public void onReplicationFailure(
366382
if (sendShardFailure == true) {
367383
failShard(e, replicaShard);
368384
} else {
369-
processLatestReceivedCheckpoint(replicaShard, thread);
385+
processLatestReceivedCheckpoint(replicaShard, thread, true);
370386
}
371387
}
372388
});
@@ -481,6 +497,11 @@ private DiscoveryNode getPrimaryNode(ShardRouting primaryShard) {
481497

482498
// visible to tests
483499
protected boolean processLatestReceivedCheckpoint(IndexShard replicaShard, Thread thread) {
500+
return processLatestReceivedCheckpoint(replicaShard, thread, false);
501+
}
502+
503+
// visible to tests
504+
protected boolean processLatestReceivedCheckpoint(IndexShard replicaShard, Thread thread, boolean isRetry) {
484505
final ReplicationCheckpoint latestPublishedCheckpoint = replicator.getPrimaryCheckpoint(replicaShard.shardId());
485506
if (latestPublishedCheckpoint != null) {
486507
logger.trace(
@@ -494,7 +515,7 @@ protected boolean processLatestReceivedCheckpoint(IndexShard replicaShard, Threa
494515
// if we retry ensure the shard is not in the process of being closed.
495516
// it will be removed from indexService's collection before the shard is actually marked as closed.
496517
if (indicesService.getShardOrNull(replicaShard.shardId()) != null) {
497-
onNewCheckpoint(replicator.getPrimaryCheckpoint(replicaShard.shardId()), replicaShard);
518+
onNewCheckpoint(replicator.getPrimaryCheckpoint(replicaShard.shardId()), replicaShard, isRetry);
498519
}
499520
};
500521
// Checks if we are using same thread and forks if necessary.
@@ -525,7 +546,24 @@ public SegmentReplicationTarget startReplication(
525546
final ReplicationCheckpoint checkpoint,
526547
final SegmentReplicationListener listener
527548
) {
528-
return replicator.startReplication(indexShard, checkpoint, sourceFactory.get(indexShard), listener);
549+
return startReplication(indexShard, checkpoint, false, listener);
550+
}
551+
552+
/**
553+
* Start a round of replication and sync to at least the given checkpoint.
554+
* @param indexShard - {@link IndexShard} replica shard
555+
* @param checkpoint - {@link ReplicationCheckpoint} checkpoint to sync to
556+
* @param isRetry - is it a retry after failure
557+
* @param listener - {@link ReplicationListener}
558+
* @return {@link SegmentReplicationTarget} target event orchestrating the event.
559+
*/
560+
public SegmentReplicationTarget startReplication(
561+
final IndexShard indexShard,
562+
final ReplicationCheckpoint checkpoint,
563+
final boolean isRetry,
564+
final SegmentReplicationListener listener
565+
) {
566+
return replicator.startReplication(indexShard, checkpoint, sourceFactory.get(indexShard), isRetry, listener);
529567
}
530568

531569
// pkg-private for integration tests

‎server/src/main/java/org/opensearch/indices/replication/SegmentReplicator.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,7 @@ public void startReplication(IndexShard shard) {
8080
shard,
8181
shard.getLatestReplicationCheckpoint(),
8282
sourceFactory.get().get(shard),
83+
false,
8384
new SegmentReplicationTargetService.SegmentReplicationListener() {
8485
@Override
8586
public void onReplicationDone(SegmentReplicationState state) {
@@ -105,16 +106,19 @@ void setSourceFactory(SegmentReplicationSourceFactory sourceFactory) {
105106
* Start a round of replication and sync to at least the given checkpoint.
106107
* @param indexShard - {@link IndexShard} replica shard
107108
* @param checkpoint - {@link ReplicationCheckpoint} checkpoint to sync to
109+
* @param source - {@link SegmentReplicationSource} segment replication source
110+
* @param isRetry - is it a retry after failure
108111
* @param listener - {@link ReplicationListener}
109112
* @return {@link SegmentReplicationTarget} target event orchestrating the event.
110113
*/
111114
SegmentReplicationTarget startReplication(
112115
final IndexShard indexShard,
113116
final ReplicationCheckpoint checkpoint,
114117
final SegmentReplicationSource source,
118+
final boolean isRetry,
115119
final SegmentReplicationTargetService.SegmentReplicationListener listener
116120
) {
117-
final SegmentReplicationTarget target = new SegmentReplicationTarget(indexShard, checkpoint, source, listener);
121+
final SegmentReplicationTarget target = new SegmentReplicationTarget(indexShard, checkpoint, source, isRetry, listener);
118122
startReplication(target, indexShard.getRecoverySettings().activityTimeout());
119123
return target;
120124
}

0 commit comments

Comments
 (0)