Fix #183 by using node name as the cassandra broadcast address

strapdata · Apr 18, 2018 · 15f8c58 · 15f8c58
1 parent 293c836
commit 15f8c58
Showing 1 changed file with 15 additions and 1 deletion.
diff --git a/core/src/main/java/org/elassandra/discovery/CassandraDiscovery.java b/core/src/main/java/org/elassandra/discovery/CassandraDiscovery.java
@@ -16,6 +16,7 @@
 package org.elassandra.discovery;
 
 import com.google.common.collect.Maps;
+import com.google.common.net.InetAddresses;
 
 import org.apache.cassandra.concurrent.Stage;
 import org.apache.cassandra.concurrent.StageManager;
@@ -476,7 +477,7 @@ public MetaDataVersionAckListener(long version, Discovery.AckListener ackListene
             clusterState.nodes().forEach((n) -> { 
                 if (!localNode.getId().equals(n.getId()) && 
                     n.status() == DiscoveryNodeStatus.ALIVE &&
-                    isNormal(Gossiper.instance.getEndpointStateForEndpoint(n.getInetAddress()))) 
+                    isNormal(n))
                     attendees.put(n.getId(), n); 
                 });
             logger.debug("new MetaDataVersionAckListener version={} attendees={}", expectedVersion, this.attendees.keySet());
@@ -515,6 +516,19 @@ private boolean isMember(InetAddress endpoint) {
         return !this.localAddress.equals(endpoint) && DatabaseDescriptor.getEndpointSnitch().getDatacenter(endpoint).equals(localDc);
     }
 
+    /**
+     * #183 lookup EndpointState with the node name = cassandra broadcast address.
+     * ES RPC adress can be different from the cassandra broadcast address.
+     */
+    private boolean isNormal(DiscoveryNode node) {
+        EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(InetAddresses.forString(node.getName()));
+        if (state == null) {
+            logger.warn("Node name=[{}] address=[{}] state not found", node.getName(), node.getInetAddress());
+            return false;
+        }
+        return state.isAlive() && state.getStatus().equals(VersionedValue.STATUS_NORMAL);
+    }
+
     private boolean isNormal(EndpointState state) {
         return state.isAlive() && state.getStatus().equals(VersionedValue.STATUS_NORMAL);
     }