Use nodeName instead of replicaId

clientlib/src/main/proto/yelp/nrtsearch/luceneserver.proto

@@ -917,11 +917,13 @@ message StateResponse {
 message AddReplicaRequest {
     int32 magicNumber = 1; //magic number send on all requests since these are meant for internal communication only
     string indexName = 2; //index name
-    int32 replicaId = 3; //replica Id
+    int32 replicaId = 3 [deprecated = true]; //replica Id
     string hostName = 4; // replica host name
     int32 port = 5; // replica port number
     // index id
     string indexId = 6;
+    // node name
+    string nodeName = 7;
 }
 
 message AddReplicaResponse {
@@ -1046,6 +1048,7 @@ message GetNodesResponse {
 message NodeInfo {
     string hostname = 1; //name or ip address of the remote host that this node is connected to for binary replication
     int32 port = 2; //port number of the remote host that this node is connected to for binary replication
+    string nodeName = 3; //name of the remote node
 }
 
 message DeleteByQueryRequest {

grpc-gateway/luceneserver.swagger.json

@@ -4142,6 +4142,9 @@
         "port": {
           "type": "integer",
           "format": "int32"
+        },
+        "nodeName": {
+          "type": "string"
         }
       }
     },

src/main/java/com/yelp/nrtsearch/server/config/NrtsearchConfig.java

@@ -125,7 +125,7 @@ public NrtsearchConfig(InputStream yamlStream) {
         configReader.getLong("maxConnectionAgeForReplication", AS_LARGE_AS_INFINITE);
     maxConnectionAgeGraceForReplication =
         configReader.getLong("maxConnectionAgeGraceForReplication", AS_LARGE_AS_INFINITE);
-    nodeName = configReader.getString("nodeName", DEFAULT_NODE_NAME);
+    nodeName = substituteEnvVariables(configReader.getString("nodeName", DEFAULT_NODE_NAME));
     hostName = substituteEnvVariables(configReader.getString("hostName", DEFAULT_HOSTNAME));
     stateDir = configReader.getString("stateDir", DEFAULT_STATE_DIR.toString());
     indexDir = configReader.getString("indexDir", DEFAULT_INDEX_DIR.toString());

src/main/java/com/yelp/nrtsearch/server/grpc/ReplicationServerClient.java

@@ -190,13 +190,13 @@ private void shutdown() throws InterruptedException {
   }
 
   public void addReplicas(
-      String indexName, String indexId, int replicaId, String hostName, int port) {
+      String indexName, String indexId, String nodeName, String hostName, int port) {
     AddReplicaRequest addReplicaRequest =
         AddReplicaRequest.newBuilder()
             .setMagicNumber(BINARY_MAGIC)
             .setIndexName(indexName)
             .setIndexId(indexId)
-            .setReplicaId(replicaId)
+            .setNodeName(nodeName)
             .setHostName(hostName)
             .setPort(port)
             .build();

src/main/java/com/yelp/nrtsearch/server/handler/AddReplicaHandler.java

@@ -61,7 +61,7 @@ private AddReplicaResponse handle(IndexState indexState, AddReplicaRequest addRe
     }
     try {
       shardState.nrtPrimaryNode.addReplica(
-          addReplicaRequest.getReplicaId(),
+          addReplicaRequest.getNodeName(),
           // channel for primary to talk to replica
           new ReplicationServerClient(
               addReplicaRequest.getHostName(), addReplicaRequest.getPort(), useKeepAlive));

src/main/java/com/yelp/nrtsearch/server/handler/GetNodesInfoHandler.java

@@ -56,6 +56,7 @@ private GetNodesResponse handle(IndexState indexState) {
             NodeInfo.newBuilder()
                 .setHostname(hostPort.getHostName())
                 .setPort(hostPort.getPort())
+                .setNodeName(replica.getNodeName())
                 .build());
       }
     }

src/main/java/com/yelp/nrtsearch/server/handler/WriteNRTPointHandler.java

@@ -76,7 +76,7 @@ private SearcherVersion handle(IndexState indexState, String indexId) {
         Iterator<NRTPrimaryNode.ReplicaDetails> it = replicasInfos.iterator();
         while (it.hasNext()) {
           NRTPrimaryNode.ReplicaDetails replicaDetails = it.next();
-          int replicaID = replicaDetails.getReplicaId();
+          String nodeName = replicaDetails.getNodeName();
           ReplicationServerClient currentReplicaServerClient =
               replicaDetails.getReplicationServerClient();
           try {
@@ -89,8 +89,8 @@ private SearcherVersion handle(IndexState indexState, String indexId) {
             Status status = e.getStatus();
             if (status.getCode().equals(Status.UNAVAILABLE.getCode())) {
               logger.info(
-                  "NRTPRimaryNode: sendNRTPoint, lost connection to replicaId: "
-                      + replicaDetails.getReplicaId()
+                  "NRTPRimaryNode: sendNRTPoint, lost connection to nodeName: "
+                      + nodeName
                       + " host: "
                       + replicaDetails.getHostPort().getHostName()
                       + " port: "
@@ -99,13 +99,13 @@ private SearcherVersion handle(IndexState indexState, String indexId) {
             }
           } catch (Exception e) {
             shardState.nrtPrimaryNode.message(
-                "top: failed to connect R"
-                    + replicaID
+                "top: failed to connect "
+                    + nodeName
                     + " for newNRTPoint; skipping: "
                     + e.getMessage());
             logger.info(
-                "top: failed to connect R"
-                    + replicaID
+                "top: failed to connect "
+                    + nodeName
                     + " for newNRTPoint; skipping: "
                     + e.getMessage());
           }

src/main/java/com/yelp/nrtsearch/server/index/ShardState.java

@@ -82,7 +82,6 @@
 
 public class ShardState implements Closeable {
   private static final Logger logger = LoggerFactory.getLogger(ShardState.class);
-  public static final int REPLICA_ID = 0;
   public static final String INDEX_DATA_DIR_NAME = "index";
   public static final String TAXONOMY_DATA_DIR_NAME = "taxonomy";
   final ExecutorService searchExecutor;
@@ -980,7 +979,7 @@ public synchronized void startReplica(
               indexStateManager.getIndexId(),
               primaryAddress,
               hostPort,
-              REPLICA_ID,
+              configuration.getNodeName(),
               indexDir,
               new ShardSearcherFactory(true, false),
               verbose ? System.out : new PrintStream(OutputStream.nullOutputStream()),
@@ -1115,7 +1114,7 @@ public void run() {
                 .addReplicas(
                     shardState.indexStateManager.getCurrent().getName(),
                     shardState.indexStateManager.getIndexId(),
-                    REPLICA_ID,
+                    nrtReplicaNode.getNodeName(),
                     nrtReplicaNode.getHostPort().getHostName(),
                     nrtReplicaNode.getHostPort().getPort());
           }

src/main/java/com/yelp/nrtsearch/server/nrt/NRTPrimaryNode.java

@@ -92,12 +92,12 @@ public NrtDataManager getNrtDataManager() {
   }
 
   public static class ReplicaDetails {
-    private final int replicaId;
+    private final String nodeName;
     private final HostPort hostPort;
     private final ReplicationServerClient replicationServerClient;
 
-    public int getReplicaId() {
-      return replicaId;
+    public String getNodeName() {
+      return nodeName;
     }
 
     public ReplicationServerClient getReplicationServerClient() {
@@ -108,9 +108,9 @@ public HostPort getHostPort() {
       return hostPort;
     }
 
-    ReplicaDetails(int replicaId, ReplicationServerClient replicationServerClient) {
-      this.replicaId = replicaId;
+    ReplicaDetails(String nodeName, ReplicationServerClient replicationServerClient) {
       this.replicationServerClient = replicationServerClient;
+      this.nodeName = nodeName;
       this.hostPort =
           new HostPort(replicationServerClient.getHost(), replicationServerClient.getPort());
     }
@@ -124,7 +124,7 @@ public boolean equals(Object o) {
       if (this == o) return true;
       if (o == null || getClass() != o.getClass()) return false;
       ReplicaDetails that = (ReplicaDetails) o;
-      return replicaId == that.replicaId && Objects.equals(hostPort, that.hostPort);
+      return Objects.equals(nodeName, that.nodeName) && Objects.equals(hostPort, that.hostPort);
     }
 
     /*
@@ -133,7 +133,7 @@ public boolean equals(Object o) {
      * */
     @Override
     public int hashCode() {
-      return Objects.hash(replicaId, hostPort);
+      return Objects.hash(nodeName, hostPort);
     }
   }
 
@@ -207,7 +207,7 @@ void sendNewNRTPointToReplicas() {
     Iterator<ReplicaDetails> it = replicasInfos.iterator();
     while (it.hasNext()) {
       ReplicaDetails replicaDetails = it.next();
-      int replicaID = replicaDetails.replicaId;
+      String nodeName = replicaDetails.getNodeName();
       ReplicationServerClient currentReplicaServerClient = replicaDetails.replicationServerClient;
       try {
         currentReplicaServerClient.newNRTPoint(
@@ -216,17 +216,17 @@ void sendNewNRTPointToReplicas() {
         Status status = e.getStatus();
         if (status.getCode().equals(Status.UNAVAILABLE.getCode())) {
           logger.warn(
-              "NRTPrimaryNode: sendNRTPoint, lost connection to replicaId: {} host: {} port: {}",
-              replicaDetails.replicaId,
+              "NRTPrimaryNode: sendNRTPoint, lost connection to nodeName: {} host: {} port: {}",
+              nodeName,
               replicaDetails.replicationServerClient.getHost(),
               replicaDetails.replicationServerClient.getPort());
           currentReplicaServerClient.close();
           it.remove();
         } else if (status.getCode().equals(Status.FAILED_PRECONDITION.getCode())
             || status.getCode().equals(Status.NOT_FOUND.getCode())) {
           logger.warn(
-              "NRTPrimaryNode: sendNRTPoint, replicaId: {} host: {} port: {} cannot process nrt point, closing connection: {}",
-              replicaDetails.replicaId,
+              "NRTPrimaryNode: sendNRTPoint, nodeName: {} host: {} port: {} cannot process nrt point, closing connection: {}",
+              nodeName,
               replicaDetails.replicationServerClient.getHost(),
               replicaDetails.replicationServerClient.getPort(),
               status);
@@ -236,8 +236,8 @@ void sendNewNRTPointToReplicas() {
       } catch (Exception e) {
         String msg =
             String.format(
-                "top: failed to connect R%d for newNRTPoint; skipping: %s",
-                replicaID, e.getMessage());
+                "top: failed to connect %s for newNRTPoint; skipping: %s",
+                nodeName, e.getMessage());
         message(msg);
         logger.warn(msg);
       }
@@ -476,11 +476,34 @@ public void setRAMBufferSizeMB(double mb) {
     writer.getConfig().setRAMBufferSizeMB(mb);
   }
 
-  public void addReplica(int replicaID, ReplicationServerClient replicationServerClient)
+  public void addReplica(String nodeName, ReplicationServerClient replicationServerClient)
       throws IOException {
-    logMessage("add replica: " + warmingSegments.size() + " current warming merges ");
-    ReplicaDetails replicaDetails = new ReplicaDetails(replicaID, replicationServerClient);
+    ReplicaDetails replicaDetails = new ReplicaDetails(nodeName, replicationServerClient);
+    logMessage(
+        String.format(
+            "Add replica %s (%s:%d) : %d current warming merges ",
+            replicaDetails.nodeName,
+            replicaDetails.replicationServerClient.getHost(),
+            replicaDetails.replicationServerClient.getPort(),
+            warmingSegments.size()));
     if (!replicasInfos.contains(replicaDetails)) {
+      Iterator<ReplicaDetails> it = replicasInfos.iterator();
+      while (it.hasNext()) {
+        ReplicaDetails existingReplicaDetails = it.next();
+        // This replica may have reused the address of a previous one that has not been cleaned up
+        // yet. We should remove any existing replica with the same address to avoid duplicate
+        // requests.
+        if (existingReplicaDetails.hostPort.equals(replicaDetails.hostPort)) {
+          logMessage(
+              String.format(
+                  "Removing existing replica with same address %s (%s:%d)",
+                  existingReplicaDetails.nodeName,
+                  existingReplicaDetails.hostPort.getHostName(),
+                  existingReplicaDetails.hostPort.getPort()));
+          existingReplicaDetails.replicationServerClient.close();
+          it.remove();
+        }
+      }
       replicasInfos.add(replicaDetails);
     }
     // Step through all currently warming segments and try to add this replica if it isn't there
@@ -492,7 +515,8 @@ public void addReplica(int replicaID, ReplicationServerClient replicationServerC
         if (preCopy.connections.contains(replicationServerClient)) {
           logMessage(
               String.format(
-                  "Replica %s:%d is already warming this segment",
+                  "Replica %s (%s:%d) is already warming this segment",
+                  replicaDetails.nodeName,
                   replicaDetails.replicationServerClient.getHost(),
                   replicaDetails.replicationServerClient.getPort()));
           // It's possible (maybe) that the replica started up, then a merge kicked off, and it
@@ -513,7 +537,8 @@ public void addReplica(int replicaID, ReplicationServerClient replicationServerC
             filesMetadata)) {
           logMessage(
               String.format(
-                  "Start precopying merged segments for new replica %s:%d",
+                  "Start precopying merged segments for new replica %s (%s:%d)",
+                  replicaDetails.nodeName,
                   replicaDetails.replicationServerClient.getHost(),
                   replicaDetails.replicationServerClient.getPort()));
         } else {
@@ -522,7 +547,8 @@ public void addReplica(int replicaID, ReplicationServerClient replicationServerC
           // nrt point sent to this replica
           logMessage(
               String.format(
-                  "Merge precopy already completed, unable to add new replica %s:%d",
+                  "Merge precopy already completed, unable to add new replica %s (%s:%d)",
+                  replicaDetails.nodeName,
                   replicaDetails.replicationServerClient.getHost(),
                   replicaDetails.replicationServerClient.getPort()));
         }
@@ -543,7 +569,8 @@ public void close() throws IOException {
       ReplicationServerClient replicationServerClient = replicaDetails.getReplicationServerClient();
       HostPort replicaHostPort = replicaDetails.getHostPort();
       logger.info(
-          "CLOSE NRT PRIMARY, closing replica channel host:{}, port:{}",
+          "CLOSE NRT PRIMARY, closing replica channel nodeName: {} host:{}, port:{}",
+          replicaDetails.getNodeName(),
           replicaHostPort.getHostName(),
           replicaHostPort.getPort());
       replicationServerClient.close();

src/main/java/com/yelp/nrtsearch/server/nrt/NRTReplicaNode.java

@@ -48,6 +48,7 @@ public class NRTReplicaNode extends ReplicaNode {
   private final ReplicaDeleterManager replicaDeleterManager;
   private final String indexName;
   private final String indexId;
+  private final String nodeName;
   private final boolean ackedCopy;
   private final boolean filterIncompatibleSegmentReaders;
   final NrtCopyThread nrtCopyThread;
@@ -62,7 +63,7 @@ public NRTReplicaNode(
       String indexId,
       ReplicationServerClient primaryAddress,
       HostPort hostPort,
-      int replicaId,
+      String nodeName,
       Directory indexDir,
       SearcherFactory searcherFactory,
       PrintStream printStream,
@@ -71,10 +72,12 @@ public NRTReplicaNode(
       boolean filterIncompatibleSegmentReaders,
       int lowPriorityCopyPercentage)
       throws IOException {
-    super(replicaId, indexDir, searcherFactory, printStream);
+    // the id is always 0, the nodeName is the identifier
+    super(0, indexDir, searcherFactory, printStream);
     this.primaryAddress = primaryAddress;
     this.indexName = indexName;
     this.indexId = indexId;
+    this.nodeName = nodeName;
     this.ackedCopy = ackedCopy;
     this.hostPort = hostPort;
     replicaDeleterManager = decInitialCommit ? new ReplicaDeleterManager(this) : null;
@@ -232,7 +235,7 @@ protected void launch(CopyJob job) {
   protected void sendNewReplica() throws IOException {
     logger.info(String.format("send new_replica to primary: %s", primaryAddress));
     primaryAddress.addReplicas(
-        indexName, this.indexId, this.id, hostPort.getHostName(), hostPort.getPort());
+        indexName, this.indexId, this.nodeName, hostPort.getHostName(), hostPort.getPort());
   }
 
   public CopyJob launchPreCopyFiles(
@@ -281,6 +284,10 @@ public ReplicationServerClient getPrimaryAddress() {
     return primaryAddress;
   }
 
+  public String getNodeName() {
+    return nodeName;
+  }
+
   public HostPort getHostPort() {
     return hostPort;
   }
@@ -290,7 +297,8 @@ public HostPort getHostPort() {
   public boolean isKnownToPrimary() {
     GetNodesResponse getNodesResponse = primaryAddress.getConnectedNodes(indexName);
     for (NodeInfo nodeInfo : getNodesResponse.getNodesList()) {
-      if (hostPort.equals(new HostPort(nodeInfo.getHostname(), nodeInfo.getPort()))) {
+      if (nodeName.equals(nodeInfo.getNodeName())
+          && hostPort.equals(new HostPort(nodeInfo.getHostname(), nodeInfo.getPort()))) {
         return true;
       }
     }

src/test/java/com/yelp/nrtsearch/server/grpc/TestServer.java

@@ -560,7 +560,7 @@ public void registerWithPrimary(String indexName, long timeoutMs) throws IOExcep
             .setMagicNumber(BINARY_MAGIC)
             .setIndexName(indexName)
             .setIndexId(indexStateManager.getIndexId())
-            .setReplicaId(ShardState.REPLICA_ID)
+            .setNodeName(getGlobalState().getNodeName())
             .setHostName("localhost")
             .setPort(getGlobalState().getReplicationPort())
             .build();