extract just updating the leaderGlobalCheckpoint in a new helper method,

this will make testing easier with out changing leaderGlobalCheckpoint field's visibility

Author: martijnvg

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

@@ -61,8 +61,7 @@ public abstract class ShardFollowNodeTask extends AllocatedPersistentTask {
     private final BiConsumer<TimeValue, Runnable> scheduler;
 
     private volatile long lastRequestedSeqno;
-    // package-protected visibility for testing only:
-    volatile long leaderGlobalCheckpoint;
+    private volatile long leaderGlobalCheckpoint;
 
     private volatile int numConcurrentReads = 0;
     private volatile int numConcurrentWrites = 0;
@@ -182,7 +181,7 @@ void handleReadResponse(long from, long maxRequiredSeqNo, ShardChangesAction.Res
     }
 
     synchronized void innerHandleReadResponse(long from, long maxRequiredSeqNo, ShardChangesAction.Response response) {
-        leaderGlobalCheckpoint = Math.max(leaderGlobalCheckpoint, response.getGlobalCheckpoint());
+        updateLeaderGlobalCheckpoint(response.getGlobalCheckpoint());
         final long newMinRequiredSeqNo;
         if (response.getOperations().length == 0) {
             newMinRequiredSeqNo = from;
@@ -221,6 +220,11 @@ synchronized void innerHandleReadResponse(long from, long maxRequiredSeqNo, Shar
         }
     }
 
+    void updateLeaderGlobalCheckpoint(long newGlobalCheckpoint) {
+        assert Thread.holdsLock(this);
+        leaderGlobalCheckpoint = Math.max(leaderGlobalCheckpoint, newGlobalCheckpoint);
+    }
+
     private void sendBulkShardOperationsRequest(List<Translog.Operation> operations) {
         sendBulkShardOperationsRequest(operations, new AtomicInteger(0));
     }

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

@@ -117,7 +117,9 @@ public void testCoordinateReads_taskCancelled() {
         assertThat(shardChangesRequests.get(0)[1], equalTo(64L));
 
         shardChangesRequests.clear();
-        task.leaderGlobalCheckpoint = 128L;
+        synchronized (task) {
+            task.updateLeaderGlobalCheckpoint(128L);
+        }
         task.markAsCompleted();
         task.coordinateReads();
         assertThat(shardChangesRequests.size(), equalTo(0));
@@ -657,7 +659,9 @@ void startTask(ShardFollowNodeTask task, long leaderGlobalCheckpoint, long follo
         task.start(followerGlobalCheckpoint);
         // Shortcut to just set leaderGlobalCheckpoint, calling for example handleReadResponse() has side effects that
         // complicates testing in isolation.
-        task.leaderGlobalCheckpoint = leaderGlobalCheckpoint;
+        synchronized (task) {
+            task.updateLeaderGlobalCheckpoint(leaderGlobalCheckpoint);
+        }
     }