You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by ja...@apache.org on 2015/03/19 18:41:38 UTC
cassandra git commit: Followup commit for 7816
Repository: cassandra
Updated Branches:
refs/heads/cassandra-2.1 9b9dda6bb -> 763130bdb
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/cassandra-2.1
Commit: 763130bdbde2f4cec2e8973bcd5203caf51cc89f
Parents: 9b9dda6
Author: T Jake Luciani <ja...@apache.org>
Authored: Thu Mar 19 13:39:23 2015 -0400
Committer: T Jake Luciani <ja...@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)