Cleanup ReplicationEngine:createThreadPool to avoid hardcoding Vcr

linkedin · Feb 23, 2021 · 6a311c6 · 6a311c6
1 parent c64e875
commit 6a311c6
Show file tree

Hide file tree

Showing 3 changed files with 576 additions and 554 deletions.
diff --git a/ambry-cloud/src/main/java/com/github/ambry/cloud/VcrReplicationManager.java b/ambry-cloud/src/main/java/com/github/ambry/cloud/VcrReplicationManager.java
@@ -25,11 +25,7 @@
 import com.github.ambry.config.StoreConfig;
 import com.github.ambry.network.ConnectionPool;
 import com.github.ambry.notification.NotificationSystem;
-import com.github.ambry.replication.FindTokenFactory;
-import com.github.ambry.replication.PartitionInfo;
-import com.github.ambry.replication.RemoteReplicaInfo;
-import com.github.ambry.replication.ReplicationEngine;
-import com.github.ambry.replication.ReplicationException;
+import com.github.ambry.replication.*;
 import com.github.ambry.server.StoreManager;
 import com.github.ambry.store.Store;
 import com.github.ambry.store.StoreKeyConverterFactory;
@@ -271,6 +267,11 @@ public long getRemoteReplicaLagFromLocalInBytes(PartitionId partitionId, String
     return -1;
   }
 
+  @Override
+  protected String getReplicaThreadName(String datacenterToReplicateFrom, int threadIndexWithinPool) {
+    return "Vcr" + ReplicaThread.getCanonicalReplicaThreadName(dataNodeId.getDatacenterName(), datacenterToReplicateFrom, threadIndexWithinPool);
+  }
+
   /**
    * Check if replication is allowed from given datacenter.
    * @param datacenterName datacenter name to check.

diff --git a/ambry-replication/src/main/java/com/github/ambry/replication/ReplicaThread.java b/ambry-replication/src/main/java/com/github/ambry/replication/ReplicaThread.java
@@ -208,6 +208,18 @@ String getName() {
     return threadName;
   }
 
+  /**
+   * Returns the canonical name for a replica thread.
+   * @param datacenterToReplicateTo The datacenter that the thread replicates data to.
+   * @param datacenterToReplicateFrom The datacenter that the thread replicates data from.
+   * @param threadIndexWithinPool The index of the thread within the thread pool.
+   * @return The name of the thread.
+   */
+  public static String getCanonicalReplicaThreadName(String datacenterToReplicateTo, String datacenterToReplicateFrom, int threadIndexWithinPool) {
+    return "ReplicaThread-" + (datacenterToReplicateTo.equals(datacenterToReplicateFrom) ? "Intra-"
+            : "Inter-") + threadIndexWithinPool + "-" + datacenterToReplicateFrom;
+  }
+
   @Override
   public void run() {
     try {