Enhance CheckpointState to support no-op replication

server/src/main/java/org/opensearch/action/bulk/TransportShardBulkAction.java

@@ -89,11 +89,15 @@
 
 import java.io.IOException;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.Executor;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.LongSupplier;
 
+import static org.opensearch.action.support.replication.ReplicationOperation.ReplicationOverridePolicy;
+import static org.opensearch.index.seqno.ReplicationTracker.ReplicationMode;
+
 /**
  * Performs shard-level bulk (index, delete or update) operations
  *
@@ -193,6 +197,14 @@ protected long primaryOperationSize(BulkShardRequest request) {
         return request.ramBytesUsed();
     }
 
+    @Override
+    protected Optional<ReplicationOverridePolicy> getReplicationOverridePolicy(IndexShard indexShard) {
+        if (indexShard.isRemoteTranslogEnabled()) {
+            return Optional.of(new ReplicationOverridePolicy(ReplicationMode.PRIMARY_TERM_VALIDATION));
+        }
+        return super.getReplicationOverridePolicy(indexShard);
+    }
+
     public static void performOnPrimary(
         BulkShardRequest request,
         IndexShard primary,

server/src/main/java/org/opensearch/action/support/replication/ReplicationOperation.java

@@ -35,8 +35,8 @@
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.apache.lucene.store.AlreadyClosedException;
 import org.opensearch.Assertions;
-import org.opensearch.OpenSearchException;
 import org.opensearch.ExceptionsHelper;
+import org.opensearch.OpenSearchException;
 import org.opensearch.action.ActionListener;
 import org.opensearch.action.UnavailableShardsException;
 import org.opensearch.action.support.ActiveShardCount;
@@ -50,8 +50,10 @@
 import org.opensearch.common.io.stream.StreamInput;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.common.util.concurrent.OpenSearchRejectedExecutionException;
+import org.opensearch.index.seqno.ReplicationTracker.ReplicationMode;
 import org.opensearch.index.seqno.SequenceNumbers;
 import org.opensearch.index.shard.ReplicationGroup;
+import org.opensearch.index.shard.ReplicationGroup.ReplicationModeAwareShardRouting;
 import org.opensearch.index.shard.ShardId;
 import org.opensearch.node.NodeClosedException;
 import org.opensearch.rest.RestStatus;
@@ -63,6 +65,8 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
+import java.util.Objects;
+import java.util.Optional;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.LongSupplier;
@@ -99,6 +103,7 @@ public class ReplicationOperation<
     private final TimeValue initialRetryBackoffBound;
     private final TimeValue retryTimeout;
     private final long primaryTerm;
+    private final ReplicationProxy replicationProxy;
 
     // exposed for tests
     private final ActionListener<PrimaryResultT> resultListener;
@@ -117,7 +122,8 @@ public ReplicationOperation(
         String opType,
         long primaryTerm,
         TimeValue initialRetryBackoffBound,
-        TimeValue retryTimeout
+        TimeValue retryTimeout,
+        Optional<ReplicationOverridePolicy> overridePolicy
     ) {
         this.replicasProxy = replicas;
         this.primary = primary;
@@ -129,6 +135,7 @@ public ReplicationOperation(
         this.primaryTerm = primaryTerm;
         this.initialRetryBackoffBound = initialRetryBackoffBound;
         this.retryTimeout = retryTimeout;
+        this.replicationProxy = new ReplicationProxyFactory().create(overridePolicy);
     }
 
     public void execute() throws Exception {
@@ -226,10 +233,158 @@ private void performOnReplicas(
 
         final ShardRouting primaryRouting = primary.routingEntry();
 
-        for (final ShardRouting shard : replicationGroup.getReplicationTargets()) {
-            if (shard.isSameAllocation(primaryRouting) == false) {
-                performOnReplica(shard, replicaRequest, globalCheckpoint, maxSeqNoOfUpdatesOrDeletes, pendingReplicationActions);
+        for (final ReplicationModeAwareShardRouting shardRouting : replicationGroup.getReplicationTargets()) {
+            replicationProxy.performOnReplica(
+                shardRouting,
+                primaryRouting,
+                replicaRequest,
+                globalCheckpoint,
+                maxSeqNoOfUpdatesOrDeletes,
+                pendingReplicationActions
+            );
+        }
+    }
+
+    /**
+     * Factory that return the {@link ReplicationProxy} instance basis the {@link ReplicationModeAwareOverrideProxy}.
+     *
+     * @opensearch.internal
+     */
+    private class ReplicationProxyFactory {
+        ReplicationProxy create(final Optional<ReplicationOverridePolicy> overridePolicy) {
+            if (overridePolicy.isEmpty()) {
+                return new FanoutReplicationProxy();
+            } else {
+                return new ReplicationModeAwareOverrideProxy(overridePolicy.get());
+            }
+        }
+    }
+
+    /**
+     * Used for performing any replication operation on replicas. Depending on the implementation, the replication call
+     * can fanout or stops here.
+     *
+     * @opensearch.internal
+     */
+    private abstract class ReplicationProxy {
+
+        /**
+         * Depending on the actual implementation and the passed {@link ReplicationModeAwareShardRouting}, the replication
+         * mode is determined using which the replication request is performed on the replica or not.
+         *
+         * @param shardRouting replication mode aware ShardRouting
+         * @param primaryRouting primary ShardRouting
+         * @param replicaRequest replication request
+         * @param globalCheckpoint current global checkpoint on primary
+         * @param maxSeqNoOfUpdatesOrDeletes maxSeqNoOfUpdatesOrDeletes
+         * @param pendingReplicationActions pendingReplicationActions
+         */
+        private void performOnReplica(
+            final ReplicationModeAwareShardRouting shardRouting,
+            final ShardRouting primaryRouting,
+            final ReplicaRequest replicaRequest,
+            final long globalCheckpoint,
+            final long maxSeqNoOfUpdatesOrDeletes,
+            final PendingReplicationActions pendingReplicationActions
+        ) {
+            ReplicationMode replicationMode = determineReplicationMode(shardRouting, primaryRouting);
+            // If the replication modes are 1. Logical replication or 2. Primary term validation, we let the call get performed on the
+            // replica shard.
+            if (replicationMode == ReplicationMode.LOGICAL_REPLICATION || replicationMode == ReplicationMode.PRIMARY_TERM_VALIDATION) {
+                ReplicationOperation.this.performOnReplica(
+                    shardRouting.getShardRouting(),
+                    replicaRequest,
+                    globalCheckpoint,
+                    maxSeqNoOfUpdatesOrDeletes,
+                    pendingReplicationActions
+                );
+            }
+        }
+
+        /**
+         * Determines what is the replication mode basis the constructor arguments of the implementation and the current
+         * replication mode aware shard routing.
+         *
+         * @param shardRouting replication mode aware ShardRouting
+         * @param primaryRouting primary ShardRouting
+         * @return the determined replication mode.
+         */
+        abstract ReplicationMode determineReplicationMode(
+            final ReplicationModeAwareShardRouting shardRouting,
+            final ShardRouting primaryRouting
+        );
+    }
+
+    /**
+     * This implementation of {@link ReplicationProxy} fans out the replication request to current shard routing if
+     * it is not the primary and has replication mode as {@link ReplicationMode#LOGICAL_REPLICATION}.
+     *
+     * @opensearch.internal
+     */
+    private class FanoutReplicationProxy extends ReplicationProxy {
+
+        private FanoutReplicationProxy() {
+
+        }
+
+        @Override
+        ReplicationMode determineReplicationMode(ReplicationModeAwareShardRouting shardRouting, ShardRouting primaryRouting) {
+            return shardRouting.getShardRouting().isSameAllocation(primaryRouting) == false
+                ? ReplicationMode.LOGICAL_REPLICATION
+                : ReplicationMode.NONE;
+        }
+    }
+
+    /**
+     * This implementation of {@link ReplicationProxy} fans out the replication request to current shard routing basis
+     * the shard routing's replication mode and replication override policy.
+     *
+     * @opensearch.internal
+     */
+    private class ReplicationModeAwareOverrideProxy extends ReplicationProxy {
+
+        private final ReplicationOverridePolicy overridePolicy;
+
+        private ReplicationModeAwareOverrideProxy(ReplicationOverridePolicy overridePolicy) {
+            assert Objects.nonNull(overridePolicy);
+            this.overridePolicy = overridePolicy;
+        }
+
+        @Override
+        ReplicationMode determineReplicationMode(ReplicationModeAwareShardRouting shardRouting, ShardRouting primaryRouting) {
+            ShardRouting currentRouting = shardRouting.getShardRouting();
+
+            // If the current routing is the primary, then it does not need to be replicated
+            if (currentRouting.isSameAllocation(primaryRouting)) {
+                return ReplicationMode.NONE;
+            }
+
+            // If the current routing's replication mode is not NONE, then we return the original replication mode.
+            if (shardRouting.getReplicationMode() != ReplicationMode.NONE) {
+                return shardRouting.getReplicationMode();
             }
+
+            // If the current routing's replication mode is none, then we check for override and return overridden mode.
+            if (Objects.nonNull(overridePolicy)) {
+                return overridePolicy.overriddenMode;
+            }
+
+            // At the end, return NONE.
+            return ReplicationMode.NONE;
+        }
+    }
+
+    /**
+     * Defines the replication override policy which individual {@link TransportReplicationAction} can implement.
+     *
+     * @opensearch.internal
+     */
+    public static class ReplicationOverridePolicy {
+
+        private final ReplicationMode overriddenMode;
+
+        public ReplicationOverridePolicy(ReplicationMode overriddenMode) {
+            this.overriddenMode = Objects.requireNonNull(overriddenMode);
         }
     }
 

server/src/main/java/org/opensearch/action/support/replication/TransportReplicationAction.java

@@ -46,6 +46,7 @@
 import org.opensearch.action.support.ChannelActionListener;
 import org.opensearch.action.support.TransportAction;
 import org.opensearch.action.support.TransportActions;
+import org.opensearch.action.support.replication.ReplicationOperation.ReplicationOverridePolicy;
 import org.opensearch.client.transport.NoNodeAvailableException;
 import org.opensearch.cluster.ClusterState;
 import org.opensearch.cluster.ClusterStateObserver;
@@ -70,6 +71,7 @@
 import org.opensearch.common.util.concurrent.AbstractRunnable;
 import org.opensearch.index.IndexNotFoundException;
 import org.opensearch.index.IndexService;
+import org.opensearch.index.seqno.ReplicationTracker.ReplicationMode;
 import org.opensearch.index.seqno.SequenceNumbers;
 import org.opensearch.index.shard.IndexShard;
 import org.opensearch.index.shard.IndexShardClosedException;
@@ -94,6 +96,7 @@
 import java.io.IOException;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
@@ -258,6 +261,15 @@ protected ReplicationOperation.Replicas<ReplicaRequest> newReplicasProxy() {
         return new ReplicasProxy();
     }
 
+    /**
+     * This method is used for defining the {@link ReplicationMode} override policy per {@link TransportReplicationAction}.
+     * @param indexShard index shard used to determining the policy.
+     * @return the override policy.
+     */
+    protected Optional<ReplicationOverridePolicy> getReplicationOverridePolicy(IndexShard indexShard) {
+        return Optional.empty();
+    }
+
     protected abstract Response newResponseInstance(StreamInput in) throws IOException;
 
     /**
@@ -533,7 +545,8 @@ public void handleException(TransportException exp) {
                         actionName,
                         primaryRequest.getPrimaryTerm(),
                         initialRetryBackoffBound,
-                        retryTimeout
+                        retryTimeout,
+                        getReplicationOverridePolicy(primaryShardReference.indexShard)
                     ).execute();
                 }
             } catch (Exception e) {