Added random active shard selector for the getchanges operations

src/main/kotlin/com/amazon/elasticsearch/replication/action/changes/GetChangesRequest.kt

@@ -23,22 +23,18 @@ import org.elasticsearch.common.io.stream.StreamOutput
 import org.elasticsearch.index.shard.ShardId
 import org.elasticsearch.transport.RemoteClusterAwareRequest
 
-class GetChangesRequest : SingleShardRequest<GetChangesRequest>, RemoteClusterAwareRequest {
-
-    val remoteNode: DiscoveryNode
+class GetChangesRequest : SingleShardRequest<GetChangesRequest> {
     val shardId : ShardId
     val fromSeqNo: Long
     val toSeqNo: Long
 
-    constructor(remoteNode: DiscoveryNode, shardId: ShardId, fromSeqNo: Long, toSeqNo: Long) : super(shardId.indexName) {
-        this.remoteNode = remoteNode
+    constructor(shardId: ShardId, fromSeqNo: Long, toSeqNo: Long) : super(shardId.indexName) {
         this.shardId = shardId
         this.fromSeqNo = fromSeqNo
         this.toSeqNo = toSeqNo
     }
 
     constructor(input : StreamInput) : super(input) {
-        this.remoteNode = DiscoveryNode(input)
         this.shardId = ShardId(input)
         this.fromSeqNo = input.readLong()
         this.toSeqNo = input.readVLong()
@@ -50,13 +46,8 @@ class GetChangesRequest : SingleShardRequest<GetChangesRequest>, RemoteClusterAw
 
     override fun writeTo(out: StreamOutput) {
         super.writeTo(out)
-        remoteNode.writeTo(out)
         shardId.writeTo(out)
         out.writeLong(fromSeqNo)
         out.writeVLong(toSeqNo)
     }
-
-    override fun getPreferredTargetNode(): DiscoveryNode {
-        return remoteNode
-    }
 }

src/main/kotlin/com/amazon/elasticsearch/replication/action/changes/TransportGetChangesAction.kt

@@ -107,7 +107,7 @@ class TransportGetChangesAction @Inject constructor(threadPool: ThreadPool, clus
     }
 
     override fun shards(state: ClusterState, request: InternalRequest): ShardsIterator {
-        // TODO: Investigate using any active shards instead of just primary
-        return state.routingTable().shardRoutingTable(request.request().shardId).primaryShardIt()
+        // Random active shards
+        return state.routingTable().shardRoutingTable(request.request().shardId).activeInitializingShardsRandomIt()
     }
 }

src/main/kotlin/com/amazon/elasticsearch/replication/task/shard/ShardReplicationTask.kt

@@ -140,7 +140,6 @@ class ShardReplicationTask(id: Long, type: String, action: String, description:
         log.info("Adding retentionlease at follower sequence number: ${indexShard.lastSyncedGlobalCheckpoint}")
         retentionLeaseHelper.addRetentionLease(remoteShardId, indexShard.lastSyncedGlobalCheckpoint , followerShardId)
 
-        var node = primaryShardNode()
         addListenerToInterruptTask()
 
         // Not really used yet as we only have one get changes action at a time.
@@ -154,7 +153,7 @@ class ShardReplicationTask(id: Long, type: String, action: String, description:
         while (scope.isActive) {
             rateLimiter.acquire()
             try {
-                val changesResponse = getChanges(node, seqNo)
+                val changesResponse = getChanges(seqNo)
                 log.info("Got ${changesResponse.changes.size} changes starting from seqNo: $seqNo")
                 sequencer.send(changesResponse)
                 seqNo = changesResponse.changes.lastOrNull()?.seqNo()?.inc() ?: seqNo
@@ -165,7 +164,6 @@ class ShardReplicationTask(id: Long, type: String, action: String, description:
             } catch (e: NodeNotConnectedException) {
                 log.info("Node not connected. Retrying request using a different node. $e")
                 delay(backOffForNodeDiscovery)
-                node = primaryShardNode()
                 rateLimiter.release()
                 continue
             }
@@ -184,24 +182,9 @@ class ShardReplicationTask(id: Long, type: String, action: String, description:
         sequencer.close()
     }
 
-    private suspend fun primaryShardNode(): DiscoveryNode {
-        val clusterStateRequest = remoteClient.admin().cluster().prepareState()
-            .clear()
-            .setIndices(remoteShardId.indexName)
-            .setRoutingTable(true)
-            .setNodes(true)
-            .setIndicesOptions(IndicesOptions.strictSingleIndexNoExpandForbidClosed())
-            .request()
-        val remoteState = remoteClient.suspending(remoteClient.admin().cluster()::state)(clusterStateRequest).state
-        val shardRouting = remoteState.routingNodes.activePrimary(remoteShardId)
-            ?: throw ShardNotFoundException(remoteShardId, "cluster: $remoteCluster")
-        return remoteState.nodes().get(shardRouting.currentNodeId())
-            ?: throw NoSuchNodeException("remote: $remoteCluster:${shardRouting.currentNodeId()}")
-    }
-
-    private suspend fun getChanges(remoteNode: DiscoveryNode, fromSeqNo: Long): GetChangesResponse {
+    private suspend fun getChanges(fromSeqNo: Long): GetChangesResponse {
         val remoteClient = client.getRemoteClusterClient(remoteCluster)
-        val request = GetChangesRequest(remoteNode, remoteShardId, fromSeqNo, fromSeqNo + batchSize)
+        val request = GetChangesRequest(remoteShardId, fromSeqNo, fromSeqNo + batchSize)
         return remoteClient.suspendExecuteWithRetries(replicationMetadata = replicationMetadata,
                 action = GetChangesAction.INSTANCE, req = request, log = log)
     }