CCR: Expose the operation primary term

Relates #32442

Author: dnhatn

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

@@ -377,6 +377,11 @@ public long getPendingPrimaryTerm() {
         return this.pendingPrimaryTerm;
     }
 
+    /** Returns the primary term that is currently being used to assign to operations */
+    public long getOperationPrimaryTerm() {
+        return this.operationPrimaryTerm;
+    }
+
     /**
      * Returns the latest cluster routing entry received with this shard.
      */

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/bulk/TransportBulkShardOperationsAction.java

@@ -78,7 +78,7 @@ public static WritePrimaryResult<BulkShardOperationsRequest, BulkShardOperations
                             index.type(),
                             index.id(),
                             index.seqNo(),
-                            primary.getPrimaryTerm(),
+                            primary.getOperationPrimaryTerm(),
                             index.version(),
                             index.versionType(),
                             BytesReference.toBytes(index.source()),
@@ -93,13 +93,13 @@ public static WritePrimaryResult<BulkShardOperationsRequest, BulkShardOperations
                             delete.id(),
                             delete.uid(),
                             delete.seqNo(),
-                            primary.getPrimaryTerm(),
+                            primary.getOperationPrimaryTerm(),
                             delete.version(),
                             delete.versionType());
                     break;
                 case NO_OP:
                     final Translog.NoOp noOp = (Translog.NoOp) operation;
-                    operationWithPrimaryTerm = new Translog.NoOp(noOp.seqNo(), primary.getPrimaryTerm(), noOp.reason());
+                    operationWithPrimaryTerm = new Translog.NoOp(noOp.seqNo(), primary.getOperationPrimaryTerm(), noOp.reason());
                     break;
                 default:
                     throw new IllegalStateException("unexpected operation type [" + operation.opType() + "]");

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/bulk/BulkShardOperationsTests.java

@@ -66,12 +66,12 @@ public void testPrimaryTermFromFollower() throws IOException {
             assertThat(snapshot.totalOperations(), equalTo(operations.size()));
             Translog.Operation operation;
             while ((operation = snapshot.next()) != null) {
-                assertThat(operation.primaryTerm(), equalTo(followerPrimary.getPrimaryTerm()));
+                assertThat(operation.primaryTerm(), equalTo(followerPrimary.getOperationPrimaryTerm()));
             }
         }
 
         for (final Translog.Operation operation : result.replicaRequest().getOperations()) {
-            assertThat(operation.primaryTerm(), equalTo(followerPrimary.getPrimaryTerm()));
+            assertThat(operation.primaryTerm(), equalTo(followerPrimary.getOperationPrimaryTerm()));
         }
 
         closeShards(followerPrimary);

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/index/engine/FollowEngineIndexShardTests.java

@@ -60,7 +60,7 @@ public void testDoNotFillGaps() throws Exception {
                 true,
                 ShardRoutingState.STARTED,
                 replicaRouting.allocationId());
-        indexShard.updateShardState(primaryRouting, indexShard.getPrimaryTerm() + 1, (shard, listener) -> {},
+        indexShard.updateShardState(primaryRouting, indexShard.getOperationPrimaryTerm() + 1, (shard, listener) -> {},
             0L, Collections.singleton(primaryRouting.allocationId().getId()),
             new IndexShardRoutingTable.Builder(primaryRouting.shardId()).addShard(primaryRouting).build(), Collections.emptySet());