From 0984b60658367fb63f0d07d79ce32acf743abc0c Mon Sep 17 00:00:00 2001 From: Suraj Singh Date: Thu, 27 Jul 2023 15:22:01 -0700 Subject: [PATCH 1/2] [Segment Replication] Use deterministic mechanism to have concurrent invocation of segment replication Signed-off-by: Suraj Singh --- .../SegmentReplicationTargetServiceTests.java | 44 ++++++++++++++++++- 1 file changed, 43 insertions(+), 1 deletion(-) diff --git a/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java b/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java index 4643615d45d7e..e4b3ca7b60358 100644 --- a/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java +++ b/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java @@ -32,6 +32,7 @@ import org.opensearch.indices.recovery.ForceSyncRequest; import org.opensearch.indices.recovery.RecoverySettings; import org.opensearch.indices.replication.checkpoint.ReplicationCheckpoint; +import org.opensearch.indices.replication.common.CopyState; import org.opensearch.indices.replication.common.ReplicationCollection; import org.opensearch.indices.replication.common.ReplicationFailedException; import org.opensearch.indices.replication.common.ReplicationLuceneIndex; @@ -49,6 +50,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import static org.junit.Assert.assertEquals; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.atLeastOnce; @@ -242,7 +244,46 @@ public void testAlreadyOnNewCheckpoint() { } public void testShardAlreadyReplicating() { - sut.startReplication(replicaShard, mock(SegmentReplicationTargetService.SegmentReplicationListener.class)); + CountDownLatch blockGetCheckpointMetadata = new CountDownLatch(1); + SegmentReplicationSource source = new TestReplicationSource() { + @Override + public void getCheckpointMetadata( + long replicationId, + ReplicationCheckpoint checkpoint, + ActionListener listener + ) { + try { + blockGetCheckpointMetadata.await(); + final CopyState copyState = new CopyState( + ReplicationCheckpoint.empty(primaryShard.shardId(), primaryShard.getLatestReplicationCheckpoint().getCodec()), + primaryShard + ); + listener.onResponse( + new CheckpointInfoResponse(copyState.getCheckpoint(), copyState.getMetadataMap(), copyState.getInfosBytes()) + ); + } catch (InterruptedException | IOException e) { + throw new RuntimeException(e); + } + } + + @Override + public void getSegmentFiles( + long replicationId, + ReplicationCheckpoint checkpoint, + List filesToFetch, + IndexShard indexShard, + ActionListener listener + ) { + listener.onResponse(new GetSegmentFilesResponse(Collections.emptyList())); + } + }; + final SegmentReplicationTarget target = spy( + new SegmentReplicationTarget(replicaShard, source, mock(SegmentReplicationTargetService.SegmentReplicationListener.class)) + ); + // Start first round of segment replication. + sut.startReplication(target); + + // Start second round of segment replication, this should fail to start as first round is still in-progress sut.startReplication(replicaShard, new SegmentReplicationTargetService.SegmentReplicationListener() { @Override public void onReplicationDone(SegmentReplicationState state) { @@ -255,6 +296,7 @@ public void onReplicationFailure(SegmentReplicationState state, ReplicationFaile assertFalse(sendShardFailure); } }); + blockGetCheckpointMetadata.countDown(); } public void testOnNewCheckpointFromNewPrimaryCancelOngoingReplication() throws InterruptedException { From 21f3036d4288f78ae85a75d10fade7078eeea5af Mon Sep 17 00:00:00 2001 From: Suraj Singh Date: Thu, 27 Jul 2023 18:11:36 -0700 Subject: [PATCH 2/2] Clean up Signed-off-by: Suraj Singh --- .../SegmentReplicationTargetServiceTests.java | 13 +++---------- 1 file changed, 3 insertions(+), 10 deletions(-) diff --git a/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java b/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java index e4b3ca7b60358..94e57f4a0d3e4 100644 --- a/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java +++ b/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java @@ -72,10 +72,7 @@ public class SegmentReplicationTargetServiceTests extends IndexShardTestCase { private IndexShard replicaShard; private IndexShard primaryShard; private ReplicationCheckpoint checkpoint; - private SegmentReplicationSource replicationSource; private SegmentReplicationTargetService sut; - - private ReplicationCheckpoint initialCheckpoint; private ReplicationCheckpoint aheadCheckpoint; private ReplicationCheckpoint newPrimaryCheckpoint; @@ -85,11 +82,10 @@ public class SegmentReplicationTargetServiceTests extends IndexShardTestCase { private DiscoveryNode localNode; private IndicesService indicesService; - private ClusterService clusterService; private SegmentReplicationState state; - private static long TRANSPORT_TIMEOUT = 30000;// 30sec + private static final long TRANSPORT_TIMEOUT = 30000;// 30sec @Override public void setUp() throws Exception { @@ -109,9 +105,6 @@ public void setUp() throws Exception { 0L, replicaShard.getLatestReplicationCheckpoint().getCodec() ); - SegmentReplicationSourceFactory replicationSourceFactory = mock(SegmentReplicationSourceFactory.class); - replicationSource = mock(SegmentReplicationSource.class); - when(replicationSourceFactory.get(replicaShard)).thenReturn(replicationSource); testThreadPool = new TestThreadPool("test", Settings.EMPTY); localNode = new DiscoveryNode("local", buildNewFakeTransportAddress(), Version.CURRENT); @@ -128,7 +121,7 @@ public void setUp() throws Exception { transportService.acceptIncomingRequests(); indicesService = mock(IndicesService.class); - clusterService = mock(ClusterService.class); + ClusterService clusterService = mock(ClusterService.class); ClusterState clusterState = mock(ClusterState.class); RoutingTable mockRoutingTable = mock(RoutingTable.class); when(clusterService.state()).thenReturn(clusterState); @@ -137,7 +130,7 @@ public void setUp() throws Exception { when(clusterState.nodes()).thenReturn(DiscoveryNodes.builder().add(localNode).build()); sut = prepareForReplication(primaryShard, replicaShard, transportService, indicesService, clusterService); - initialCheckpoint = replicaShard.getLatestReplicationCheckpoint(); + ReplicationCheckpoint initialCheckpoint = replicaShard.getLatestReplicationCheckpoint(); aheadCheckpoint = new ReplicationCheckpoint( initialCheckpoint.getShardId(), initialCheckpoint.getPrimaryTerm(),