Changes to enable “shouldForward” gating

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

@@ -269,6 +269,7 @@ private void finishRequest() {
                         context.getLocationToSync(),
                         null,
                         context.getPrimary(),
+                        context.getPrimary().shouldForward(),
                         logger
                     )
                 );

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

@@ -151,7 +151,7 @@ public void execute() throws Exception {
     private void handlePrimaryResult(final PrimaryResultT primaryResult) {
         this.primaryResult = primaryResult;
         final ReplicaRequest replicaRequest = primaryResult.replicaRequest();
-        if (replicaRequest != null) {
+        if (replicaRequest != null && primaryResult.shouldForward()) {
             if (logger.isTraceEnabled()) {
                 logger.trace("[{}] op [{}] completed on primary for request [{}]", primary.routingEntry().shardId(), opType, request);
             }
@@ -621,5 +621,7 @@ public interface PrimaryResult<RequestT extends ReplicationRequest<RequestT>> {
          * @param listener calllback that is invoked after post replication actions have completed
          * */
         void runPostReplicationActions(ActionListener<Void> listener);
+
+        boolean shouldForward();
     }
 }

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

@@ -558,12 +558,18 @@ public static class PrimaryResult<ReplicaRequest extends ReplicationRequest<Repl
         protected final ReplicaRequest replicaRequest;
         public final Response finalResponseIfSuccessful;
         public final Exception finalFailure;
+        public final boolean shouldForward;
 
         /**
          * Result of executing a primary operation
          * expects <code>finalResponseIfSuccessful</code> or <code>finalFailure</code> to be not-null
          */
-        public PrimaryResult(ReplicaRequest replicaRequest, Response finalResponseIfSuccessful, Exception finalFailure) {
+        public PrimaryResult(
+            ReplicaRequest replicaRequest,
+            Response finalResponseIfSuccessful,
+            Exception finalFailure,
+            boolean shouldForward
+        ) {
             assert finalFailure != null ^ finalResponseIfSuccessful != null : "either a response or a failure has to be not null, "
                 + "found ["
                 + finalFailure
@@ -573,6 +579,11 @@ public PrimaryResult(ReplicaRequest replicaRequest, Response finalResponseIfSucc
             this.replicaRequest = replicaRequest;
             this.finalResponseIfSuccessful = finalResponseIfSuccessful;
             this.finalFailure = finalFailure;
+            this.shouldForward = shouldForward;
+        }
+
+        public PrimaryResult(ReplicaRequest replicaRequest, Response finalResponseIfSuccessful, Exception finalFailure) {
+            this(replicaRequest, finalResponseIfSuccessful, finalFailure, true);
         }
 
         public PrimaryResult(ReplicaRequest replicaRequest, Response replicationResponse) {
@@ -599,6 +610,11 @@ public void runPostReplicationActions(ActionListener<Void> listener) {
                 listener.onResponse(null);
             }
         }
+
+        @Override
+        public boolean shouldForward() {
+            return shouldForward;
+        }
     }
 
     public static class ReplicaResult {

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

@@ -279,9 +279,10 @@ public WritePrimaryResult(
             @Nullable Location location,
             @Nullable Exception operationFailure,
             IndexShard primary,
+            boolean shouldForward,
             Logger logger
         ) {
-            super(request, finalResponse, operationFailure);
+            super(request, finalResponse, operationFailure, shouldForward);
             this.location = location;
             this.primary = primary;
             this.logger = logger;
@@ -293,6 +294,17 @@ public WritePrimaryResult(
                 + "] failure";
         }
 
+        public WritePrimaryResult(
+            ReplicaRequest request,
+            @Nullable Response finalResponse,
+            @Nullable Location location,
+            @Nullable Exception operationFailure,
+            IndexShard primary,
+            Logger logger
+        ) {
+            this(request, finalResponse, location, operationFailure, primary, true, logger);
+        }
+
         @Override
         public void runPostReplicationActions(ActionListener<Void> listener) {
             if (finalFailure != null) {

server/src/main/java/org/opensearch/index/shard/IndexShard.java

@@ -835,8 +835,7 @@ public Engine.IndexResult applyIndexOperationOnReplica(
         boolean isRetry,
         SourceToParse sourceToParse
     ) throws IOException {
-        Boolean isSegRepEnabled = indexSettings.getValue(IndexSettings.INDEX_SEGMENT_REPLICATION_SETTING);
-        if (isSegRepEnabled != null && isSegRepEnabled) {
+        if (isSegmentReplicationEnabled()) {
             Engine.Index index;
             try {
                 index = parseSourceAndPrepareIndex(
@@ -4096,4 +4095,17 @@ public void verifyShardBeforeIndexClosing() throws IllegalStateException {
     RetentionLeaseSyncer getRetentionLeaseSyncer() {
         return retentionLeaseSyncer;
     }
+
+    /**
+     * Controls whether requests should be forwarded from the
+     * primary to the replica.
+     */
+    public boolean shouldForward() {
+        // Eventually this will also incorporate the presence of pluggable translog
+        return !isSegmentReplicationEnabled();
+    }
+
+    private boolean isSegmentReplicationEnabled() {
+        return indexSettings.getValue(IndexSettings.INDEX_SEGMENT_REPLICATION_SETTING);
+    }
 }

server/src/test/java/org/opensearch/action/admin/indices/close/TransportVerifyShardBeforeCloseActionTests.java

@@ -415,6 +415,14 @@ public void runPostReplicationActions(ActionListener<Void> listener) {
             listener.onResponse(null);
         }
 
+        /**
+         * Default implementation.
+         */
+        @Override
+        public boolean shouldForward() {
+            return true;
+        }
+
         public ReplicationResponse.ShardInfo getShardInfo() {
             return shardInfo.get();
         }

server/src/test/java/org/opensearch/action/support/replication/ReplicationOperationTests.java

@@ -669,6 +669,14 @@ public void runPostReplicationActions(ActionListener<Void> listener) {
                 listener.onResponse(null);
             }
 
+            /**
+             * Default implementation.
+             */
+            @Override
+            public boolean shouldForward() {
+                return true;
+            }
+
             public ShardInfo getShardInfo() {
                 return shardInfo;
             }

test/framework/src/main/java/org/opensearch/index/replication/OpenSearchIndexLevelReplicationTestCase.java

@@ -869,6 +869,14 @@ public void setShardInfo(ReplicationResponse.ShardInfo shardInfo) {
             public void runPostReplicationActions(ActionListener<Void> listener) {
                 listener.onResponse(null);
             }
+
+            /**
+             * Default implementation
+             */
+            @Override
+            public boolean shouldForward() {
+                return true;
+            }
         }
 
     }