Cleanup ReplicationEngine:createThreadPool to avoid hardcoding Vcr

linkedin · Feb 23, 2021 · 2739d6a · 2739d6a
1 parent c64e875
commit 2739d6a
Show file tree

Hide file tree

Showing 3 changed files with 29 additions and 8 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 {

diff --git a/ambry-replication/src/main/java/com/github/ambry/replication/ReplicationEngine.java b/ambry-replication/src/main/java/com/github/ambry/replication/ReplicationEngine.java
@@ -343,9 +343,7 @@ private List<ReplicaThread> createThreadPool(String datacenter, int numberOfThre
     ResponseHandler responseHandler = new ResponseHandler(clusterMap);
     for (int i = 0; i < numberOfThreads; i++) {
       boolean replicatingOverSsl = sslEnabledDatacenters.contains(datacenter);
-      String threadIdentity =
-          (startThread ? "Vcr" : "") + "ReplicaThread-" + (dataNodeId.getDatacenterName().equals(datacenter) ? "Intra-"
-              : "Inter-") + i + "-" + datacenter;
+      String threadIdentity = getReplicaThreadName(datacenter, i);
       try {
         StoreKeyConverter threadSpecificKeyConverter = storeKeyConverterFactory.getStoreKeyConverter();
         Transformer threadSpecificTransformer =
@@ -370,6 +368,16 @@ private List<ReplicaThread> createThreadPool(String datacenter, int numberOfThre
     return replicaThreads;
   }
 
+  /**
+   * Chooses a name for a new replica thread.
+   * @param datacenterToReplicateFrom The datacenter that we replicate from in this thread.
+   * @param threadIndexWithinPool The index of the thread within the thread pool.
+   * @return The name of the thread.
+   */
+  protected String getReplicaThreadName(String datacenterToReplicateFrom, int threadIndexWithinPool) {
+    return ReplicaThread.getCanonicalReplicaThreadName(dataNodeId.getDatacenterName(), datacenterToReplicateFrom, threadIndexWithinPool);
+  }
+
   /**
    * Reads the replica tokens from storage, populates the Remote replica info,
    * and re-persists the tokens if they have been reset.