Adding condition to check if fork of thread is necessary.

Signed-off-by: Rishikesh1159 <rishireddy1159@gmail.com>
opensearch-project · Jul 20, 2022 · e0ec134 · e0ec134
1 parent d9b2e71
commit e0ec134
Showing 1 changed file with 9 additions and 2 deletions.
diff --git a/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTargetService.java b/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTargetService.java
@@ -113,15 +113,22 @@ public synchronized void onNewCheckpoint(final ReplicationCheckpoint receivedChe
             );
             return;
         }
+        final Thread thread = Thread.currentThread();
         if (replicaShard.shouldProcessCheckpoint(receivedCheckpoint)) {
             startReplication(receivedCheckpoint, replicaShard, new SegmentReplicationListener() {
                 @Override
                 public void onReplicationDone(SegmentReplicationState state) {
                     // if we received a checkpoint during the copy event that is ahead of this
                     // try and process it.
                     if (latestReceivedCheckpoint.get(replicaShard.shardId()).isAheadOf(replicaShard.getLatestReplicationCheckpoint())) {
-                        threadPool.generic()
-                            .execute(() -> onNewCheckpoint(latestReceivedCheckpoint.get(replicaShard.shardId()), replicaShard));
+                        Runnable runnable = () -> onNewCheckpoint(latestReceivedCheckpoint.get(replicaShard.shardId()), replicaShard);
+                        // Checks if we are using same thread and forks if necessary.
+                        if (thread == Thread.currentThread()){
+                            threadPool.generic().execute(runnable);
+                        }
+                        else{
+                            runnable.run();
+                        }
                     }
                 }