cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject [1/2] cassandra git commit: Followup commit for 7816
Date Thu, 19 Mar 2015 17:50:59 GMT
Repository: cassandra
Updated Branches:
  refs/heads/trunk b25adc765 -> 4597bb5b1


Followup commit for 7816

patch by Stephania ; reviewed by tjake for CASSANDRA-7816


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/763130bd
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/763130bd
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/763130bd

Branch: refs/heads/trunk
Commit: 763130bdbde2f4cec2e8973bcd5203caf51cc89f
Parents: 9b9dda6
Author: T Jake Luciani <jake@apache.org>
Authored: Thu Mar 19 13:39:23 2015 -0400
Committer: T Jake Luciani <jake@apache.org>
Committed: Thu Mar 19 13:41:20 2015 -0400

----------------------------------------------------------------------
 src/java/org/apache/cassandra/gms/EndpointState.java | 12 ------------
 src/java/org/apache/cassandra/gms/Gossiper.java      | 12 ++----------
 src/java/org/apache/cassandra/transport/Server.java  | 11 +++++++++--
 3 files changed, 11 insertions(+), 24 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/763130bd/src/java/org/apache/cassandra/gms/EndpointState.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/gms/EndpointState.java b/src/java/org/apache/cassandra/gms/EndpointState.java
index 74433ba..1029374 100644
--- a/src/java/org/apache/cassandra/gms/EndpointState.java
+++ b/src/java/org/apache/cassandra/gms/EndpointState.java
@@ -47,14 +47,12 @@ public class EndpointState
     /* fields below do not get serialized */
     private volatile long updateTimestamp;
     private volatile boolean isAlive;
-    private volatile boolean hasPendingEcho;
 
     EndpointState(HeartBeatState initialHbState)
     {
         hbState = initialHbState;
         updateTimestamp = System.nanoTime();
         isAlive = true;
-        hasPendingEcho = false;
     }
 
     HeartBeatState getHeartBeatState()
@@ -116,16 +114,6 @@ public class EndpointState
         isAlive = false;
     }
 
-    public boolean hasPendingEcho()
-    {
-        return hasPendingEcho;
-    }
-
-    public void markPendingEcho(boolean val)
-    {
-        hasPendingEcho = val;
-    }
-
     public String toString()
     {
         return "EndpointState: HeartBeatState = " + hbState + ", AppStateMap = " + applicationState;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/763130bd/src/java/org/apache/cassandra/gms/Gossiper.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java
index ac98c53..9c0ef8a 100644
--- a/src/java/org/apache/cassandra/gms/Gossiper.java
+++ b/src/java/org/apache/cassandra/gms/Gossiper.java
@@ -883,12 +883,6 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
             return;
         }
 
-        if (localState.hasPendingEcho())
-        {
-            logger.debug("{} has already a pending echo, skipping it", localState);
-            return;
-        }
-
         localState.markDead();
 
         MessageOut<EchoMessage> echoMessage = new MessageOut<EchoMessage>(MessagingService.Verb.ECHO,
new EchoMessage(), EchoMessage.serializer);
@@ -902,19 +896,17 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
 
             public void response(MessageIn msg)
             {
-                localState.markPendingEcho(false);
                 realMarkAlive(addr, localState);
             }
         };
 
-        localState.markPendingEcho(true);
         MessagingService.instance().sendRR(echoMessage, addr, echoHandler);
     }
 
     private void realMarkAlive(final InetAddress addr, final EndpointState localState)
     {
         if (logger.isTraceEnabled())
-                logger.trace("marking as alive {}", addr);
+            logger.trace("marking as alive {}", addr);
         localState.markAlive();
         localState.updateTimestamp(); // prevents doStatusCheck from racing us and evicting
if it was down > aVeryLongTime
         liveEndpoints.add(addr);
@@ -925,7 +917,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
         for (IEndpointStateChangeSubscriber subscriber : subscribers)
             subscriber.onAlive(addr, localState);
         if (logger.isTraceEnabled())
-                logger.trace("Notified " + subscribers);
+            logger.trace("Notified " + subscribers);
     }
 
     private void markDead(InetAddress addr, EndpointState localState)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/763130bd/src/java/org/apache/cassandra/transport/Server.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/transport/Server.java b/src/java/org/apache/cassandra/transport/Server.java
index f396fd9..8f0f89f 100644
--- a/src/java/org/apache/cassandra/transport/Server.java
+++ b/src/java/org/apache/cassandra/transport/Server.java
@@ -22,7 +22,9 @@ import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.UnknownHostException;
 import java.util.EnumMap;
+import java.util.Map;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLEngine;
@@ -341,6 +343,7 @@ public class Server implements CassandraDaemon.Server
     private static class EventNotifier extends MigrationListener implements IEndpointLifecycleSubscriber
     {
         private final Server server;
+        private final Map<InetAddress, Event.StatusChange.Status> lastStatusChange
= new ConcurrentHashMap<>();
         private static final InetAddress bindAll;
         static {
             try
@@ -395,12 +398,16 @@ public class Server implements CassandraDaemon.Server
 
         public void onUp(InetAddress endpoint)
         {
-            server.connectionTracker.send(Event.StatusChange.nodeUp(getRpcAddress(endpoint),
server.socket.getPort()));
+            Event.StatusChange.Status prev = lastStatusChange.put(endpoint, Event.StatusChange.Status.UP);
+            if (prev == null || prev != Event.StatusChange.Status.UP)
+                server.connectionTracker.send(Event.StatusChange.nodeUp(getRpcAddress(endpoint),
server.socket.getPort()));
         }
 
         public void onDown(InetAddress endpoint)
         {
-            server.connectionTracker.send(Event.StatusChange.nodeDown(getRpcAddress(endpoint),
server.socket.getPort()));
+            Event.StatusChange.Status prev = lastStatusChange.put(endpoint, Event.StatusChange.Status.DOWN);
+            if (prev == null || prev != Event.StatusChange.Status.DOWN)
+                server.connectionTracker.send(Event.StatusChange.nodeDown(getRpcAddress(endpoint),
server.socket.getPort()));
         }
 
         public void onCreateKeyspace(String ksName)


Mime
View raw message