From 63a8bfe030ece2497c839bb34148b73e416466f4 Mon Sep 17 00:00:00 2001 From: Sylvain Lebresne Date: Tue, 30 Oct 2012 16:11:27 +0100 Subject: [PATCH] Fix binary protocol NEW_NODE event patch by slebresne; reviewed by yukim for CASSANDRA-4679 --- CHANGES.txt | 1 + doc/native_protocol.spec | 7 ++ .../gms/IEndpointStateChangeSubscriber.java | 1 - .../cassandra/service/CassandraDaemon.java | 22 +++--- .../service/IEndpointLifecycleSubscriber.java | 67 +++++++++++++++++++ .../cassandra/service/StorageService.java | 63 ++++++++++++++++- .../org/apache/cassandra/transport/Event.java | 7 +- .../apache/cassandra/transport/Server.java | 36 +++++----- .../AntiEntropyServiceTestAbstract.java | 1 - 9 files changed, 171 insertions(+), 34 deletions(-) create mode 100644 src/java/org/apache/cassandra/service/IEndpointLifecycleSubscriber.java diff --git a/CHANGES.txt b/CHANGES.txt index 2b87c85e2b..e25bfbfcb8 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -39,6 +39,7 @@ * Fix list prepend logic (CQL3) (CASSANDRA-4835) * Add booleans as literals in CQL3 (CASSANDRA-4776) * Allow renaming PK columns in CQL3 (CASSANDRA-4822) + * Fix binary protocol NEW_NODE event (CASSANDRA-4679) Merged from 1.1: * fix get_paged_slice to wrap to next row correctly (CASSANDRA-4816) * fix indexing empty column values (CASSANDRA-4832) diff --git a/doc/native_protocol.spec b/doc/native_protocol.spec index 5c84f7173d..f534cdeb80 100644 --- a/doc/native_protocol.spec +++ b/doc/native_protocol.spec @@ -481,6 +481,13 @@ Table of Contents All EVENT message have a streamId of -1 (Section 2.3). + Please note that "NEW_NODE" and "UP" events are sent based on internal Gossip + communication and as such may be sent a short delay before the binary + protocol server on the newly up node is fully started. Clients are thus + advise to wait a short time before trying to connect to the node (1 seconds + should be enough), otherwise they may experience a connection refusal at + first. + 5. Compression diff --git a/src/java/org/apache/cassandra/gms/IEndpointStateChangeSubscriber.java b/src/java/org/apache/cassandra/gms/IEndpointStateChangeSubscriber.java index fff613b843..2abee01fa8 100644 --- a/src/java/org/apache/cassandra/gms/IEndpointStateChangeSubscriber.java +++ b/src/java/org/apache/cassandra/gms/IEndpointStateChangeSubscriber.java @@ -27,7 +27,6 @@ import java.net.InetAddress; * instance to decide what he does with this change. Not all modules maybe interested * in all state changes. */ - public interface IEndpointStateChangeSubscriber { /** diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java index 03da7e362f..7b8e7d8b9f 100644 --- a/src/java/org/apache/cassandra/service/CassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java @@ -300,16 +300,7 @@ public class CassandraDaemon // start server internals StorageService.instance.registerDaemon(this); - try - { - StorageService.instance.initServer(); - } - catch (ConfigurationException e) - { - logger.error("Fatal configuration error", e); - System.err.println(e.getMessage() + "\nFatal configuration error; unable to start server. See log for stacktrace."); - System.exit(1); - } + StorageService.instance.initServerLocally(); Mx4jTool.maybeLoad(); @@ -357,6 +348,17 @@ public class CassandraDaemon nativeServer.start(); else logger.info("Not starting native transport as requested. Use JMX (StorageService->startNativeTransport()) to start it"); + + try + { + StorageService.instance.maybeJoinRing(StorageService.RING_DELAY); + } + catch (ConfigurationException e) + { + logger.error("Fatal configuration error", e); + System.err.println(e.getMessage() + "\nFatal configuration error; unable to start server. See log for stacktrace."); + System.exit(1); + } } /** diff --git a/src/java/org/apache/cassandra/service/IEndpointLifecycleSubscriber.java b/src/java/org/apache/cassandra/service/IEndpointLifecycleSubscriber.java new file mode 100644 index 0000000000..24cb3d780f --- /dev/null +++ b/src/java/org/apache/cassandra/service/IEndpointLifecycleSubscriber.java @@ -0,0 +1,67 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.cassandra.service; + +import java.net.InetAddress; + +/** + * Interface on which interested parties can be notified of high level endpoint + * state changes. + * + * Note that while IEndpointStateChangeSubscriber notify about gossip related + * changes (IEndpointStateChangeSubscriber.onJoin() is called when a node join + * gossip), this interface allows to be notified about higher level events. + */ +public interface IEndpointLifecycleSubscriber +{ + /** + * Called when a new node joins the cluster, i.e. either has just been + * bootstrapped or "instajoins". + * + * @param endpoint the newly added endpoint. + */ + public void onJoinCluster(InetAddress endpoint); + + /** + * Called when a new node leave the cluster (decommission or removeToken). + * + * @param endpoint the endpoint that is leaving. + */ + public void onLeaveCluster(InetAddress endpoint); + + /** + * Called when a node is marked UP. + * + * @param endpoint the endpoint marked UP. + */ + public void onUp(InetAddress endpoint); + + /** + * Called when a node is marked DOWN. + * + * @param endpoint the endpoint marked DOWN. + */ + public void onDown(InetAddress endpoint); + + /** + * Called when a node has moved (to a new token). + * + * @param endpoint the endpoint that has moved. + */ + public void onMove(InetAddress endpoint); +} diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 58ce112628..ad4a534acb 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -105,7 +105,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe * This pool is used by tasks that can have longer execution times, and usually are non periodic. */ public static final DebuggableScheduledThreadPoolExecutor tasks = new DebuggableScheduledThreadPoolExecutor("NonPeriodicTasks"); -/** + /** * tasks that do not need to be waited for on shutdown/drain */ public static final DebuggableScheduledThreadPoolExecutor optionalTasks = new DebuggableScheduledThreadPoolExecutor("OptionalTasks"); @@ -181,6 +181,8 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe private static ScheduledRangeTransferExecutorService rangeXferExecutor = new ScheduledRangeTransferExecutorService(); + private final List lifecycleSubscribers = new CopyOnWriteArrayList(); + public void finishBootstrapping() { isBootstrapMode = false; @@ -253,6 +255,16 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe this.daemon = daemon; } + public void register(IEndpointLifecycleSubscriber subscriber) + { + lifecycleSubscribers.add(subscriber); + } + + public void unregister(IEndpointLifecycleSubscriber subscriber) + { + lifecycleSubscribers.remove(subscriber); + } + // should only be called via JMX public void stopGossiping() { @@ -387,6 +399,12 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe } public synchronized void initServer(int delay) throws ConfigurationException + { + initServerLocally(); + maybeJoinRing(delay); + } + + public void initServerLocally() { logger.info("Cassandra version: " + FBUtilities.getReleaseVersionString()); logger.info("Thrift API version: " + Constants.VERSION); @@ -484,6 +502,18 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe } }, "StorageServiceShutdownHook"); Runtime.getRuntime().addShutdownHook(drainOnShutdown); + } + + public synchronized void maybeJoinRing(int delay) throws ConfigurationException + { + // This method should only be called as part of the server initialization, so if initialized == true, we've already gone + // through that. If the ring must be joined after the server initialization, use joinTokenRing() directly. + if (initialized) + { + if (isClientMode) + throw new UnsupportedOperationException("StorageService does not support switching modes."); + return; + } if (Boolean.parseBoolean(System.getProperty("cassandra.join_ring", "true"))) { @@ -1249,8 +1279,21 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe logger.debug("Node " + endpoint + " state normal, token " + tokens); if (tokenMetadata.isMember(endpoint)) + { logger.info("Node " + endpoint + " state jump to normal"); + if (!isClientMode) + { + for (IEndpointLifecycleSubscriber subscriber : lifecycleSubscribers) + subscriber.onUp(endpoint); + } + } + else if (!isClientMode) + { + for (IEndpointLifecycleSubscriber subscriber : lifecycleSubscribers) + subscriber.onJoinCluster(endpoint); + } + // Order Matters, TM.updateHostID() should be called before TM.updateNormalToken(), (see CASSANDRA-4300). if (Gossiper.instance.usesHostId(endpoint)) tokenMetadata.updateHostId(Gossiper.instance.getHostId(endpoint), endpoint); @@ -1344,8 +1387,16 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe SystemTable.updateLocalTokens(Collections.emptyList(), localTokensToRemove); if (tokenMetadata.isMoving(endpoint)) // if endpoint was moving to a new token + { tokenMetadata.removeFromMoving(endpoint); + if (!isClientMode) + { + for (IEndpointLifecycleSubscriber subscriber : lifecycleSubscribers) + subscriber.onMove(endpoint); + } + } + calculatePendingRanges(); } @@ -1504,6 +1555,11 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe Gossiper.instance.removeEndpoint(endpoint); tokenMetadata.removeEndpoint(endpoint); tokenMetadata.removeBootstrapTokens(tokens); + if (!isClientMode) + { + for (IEndpointLifecycleSubscriber subscriber : lifecycleSubscribers) + subscriber.onLeaveCluster(endpoint); + } calculatePendingRanges(); if (!isClientMode) { @@ -1856,6 +1912,11 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe public void onDead(InetAddress endpoint, EndpointState state) { MessagingService.instance().convict(endpoint); + if (!isClientMode) + { + for (IEndpointLifecycleSubscriber subscriber : lifecycleSubscribers) + subscriber.onDown(endpoint); + } } public void onRestart(InetAddress endpoint, EndpointState state) diff --git a/src/java/org/apache/cassandra/transport/Event.java b/src/java/org/apache/cassandra/transport/Event.java index 855049dbeb..33b08f4bd5 100644 --- a/src/java/org/apache/cassandra/transport/Event.java +++ b/src/java/org/apache/cassandra/transport/Event.java @@ -58,7 +58,7 @@ public abstract class Event public static class TopologyChange extends Event { - public enum Change { NEW_NODE, REMOVED_NODE } + public enum Change { NEW_NODE, REMOVED_NODE, MOVED_NODE } public final Change change; public final InetSocketAddress node; @@ -80,6 +80,11 @@ public abstract class Event return new TopologyChange(Change.REMOVED_NODE, new InetSocketAddress(host, port)); } + public static TopologyChange movedNode(InetAddress host, int port) + { + return new TopologyChange(Change.MOVED_NODE, new InetSocketAddress(host, port)); + } + // Assumes the type has already by been deserialized private static TopologyChange deserializeEvent(ChannelBuffer cb) { diff --git a/src/java/org/apache/cassandra/transport/Server.java b/src/java/org/apache/cassandra/transport/Server.java index e820554839..e629ca4851 100644 --- a/src/java/org/apache/cassandra/transport/Server.java +++ b/src/java/org/apache/cassandra/transport/Server.java @@ -35,10 +35,10 @@ import org.jboss.netty.logging.Slf4JLoggerFactory; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.apache.cassandra.gms.*; import org.apache.cassandra.service.CassandraDaemon; import org.apache.cassandra.service.IMigrationListener; import org.apache.cassandra.service.MigrationManager; +import org.apache.cassandra.service.IEndpointLifecycleSubscriber; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.transport.messages.EventMessage; @@ -63,7 +63,7 @@ public class Server implements CassandraDaemon.Server { this.socket = socket; EventNotifier notifier = new EventNotifier(this); - Gossiper.instance.register(notifier); + StorageService.instance.register(notifier); MigrationManager.instance.register(notifier); } @@ -203,7 +203,7 @@ public class Server implements CassandraDaemon.Server } } - private static class EventNotifier implements IEndpointStateChangeSubscriber, IMigrationListener + private static class EventNotifier implements IEndpointLifecycleSubscriber, IMigrationListener { private final Server server; @@ -227,35 +227,31 @@ public class Server implements CassandraDaemon.Server } } - public void onJoin(InetAddress endpoint, EndpointState epState) + public void onJoinCluster(InetAddress endpoint) { server.connectionTracker.send(Event.TopologyChange.newNode(getRpcAddress(endpoint), server.socket.getPort())); } - public void onChange(InetAddress endpoint, ApplicationState state, VersionedValue value) - { - } - - public void onAlive(InetAddress endpoint, EndpointState state) - { - server.connectionTracker.send(Event.StatusChange.nodeUp(getRpcAddress(endpoint), server.socket.getPort())); - } - - public void onDead(InetAddress endpoint, EndpointState state) - { - server.connectionTracker.send(Event.StatusChange.nodeDown(getRpcAddress(endpoint), server.socket.getPort())); - } - - public void onRemove(InetAddress endpoint) + public void onLeaveCluster(InetAddress endpoint) { server.connectionTracker.send(Event.TopologyChange.removedNode(getRpcAddress(endpoint), server.socket.getPort())); } - public void onRestart(InetAddress endpoint, EndpointState state) + public void onMove(InetAddress endpoint) + { + server.connectionTracker.send(Event.TopologyChange.movedNode(getRpcAddress(endpoint), server.socket.getPort())); + } + + public void onUp(InetAddress endpoint) { 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())); + } + public void onCreateKeyspace(String ksName) { server.connectionTracker.send(new Event.SchemaChange(Event.SchemaChange.Change.CREATED, ksName)); diff --git a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java b/test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java index 87cc910414..ef7a2abed5 100644 --- a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java +++ b/test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java @@ -79,7 +79,6 @@ public abstract class AntiEntropyServiceTestAbstract extends SchemaLoader init(); LOCAL = FBUtilities.getBroadcastAddress(); - StorageService.instance.initServer(0); // generate a fake endpoint for which we can spoof receiving/sending trees REMOTE = InetAddress.getByName("127.0.0.2"); store = null;