diff --git a/CHANGES.txt b/CHANGES.txt index 547a1bb11f..791d7e110a 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -30,6 +30,8 @@ Merged from 4.0: Merged from 3.11: Merged from 3.0: * Implement the logic in bin/stop-server (CASSANDRA-18838) + * Fix nodetool enable/disablebinary to correctly set rpc readiness in gossip (CASSANDRA-18935) + * Implement the logic in bin/stop-server (CASSANDRA-18838) * Upgrade snappy-java to 1.1.10.4 (CASSANDRA-18878) * Add cqlshrc.sample and credentials.sample into Debian package (CASSANDRA-18818) diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java index 332f2a714f..1e1ac30d11 100644 --- a/src/java/org/apache/cassandra/service/CassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java @@ -665,10 +665,7 @@ public class CassandraDaemon { String nativeFlag = START_NATIVE_TRANSPORT.getString(); if (START_NATIVE_TRANSPORT.getBoolean() || (nativeFlag == null && DatabaseDescriptor.startNativeTransport())) - { startNativeTransport(); - StorageService.instance.setRpcReady(true); - } else logger.info("Not starting native transport as requested. Use JMX (StorageService->startNativeTransport()) or nodetool (enablebinary) to start it"); } @@ -815,13 +812,24 @@ public class CassandraDaemon if (nativeTransportService == null) throw new IllegalStateException("setup() must be called first for CassandraDaemon"); + // this iterates over a collection of servers and returns true if one of them is started + boolean alreadyRunning = nativeTransportService.isRunning(); + + // this might in practice start all servers which are not started yet nativeTransportService.start(); + + // interact with gossip only in case if no server was started before to signal they are started now + if (!alreadyRunning) + StorageService.instance.setRpcReady(true); } public void stopNativeTransport() { if (nativeTransportService != null) + { + StorageService.instance.setRpcReady(false); nativeTransportService.stop(); + } } public boolean isNativeTransportRunning() diff --git a/test/distributed/org/apache/cassandra/distributed/test/NativeProtocolTest.java b/test/distributed/org/apache/cassandra/distributed/test/NativeProtocolTest.java index 11ba9946cb..e958c60481 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/NativeProtocolTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/NativeProtocolTest.java @@ -33,6 +33,7 @@ import com.datastax.driver.core.SimpleStatement; import com.datastax.driver.core.Statement; import org.apache.cassandra.distributed.api.ICluster; import org.apache.cassandra.distributed.impl.RowUtil; +import org.apache.cassandra.utils.FBUtilities; import static org.apache.cassandra.config.CassandraRelevantProperties.JOIN_RING; import static org.apache.cassandra.distributed.action.GossipHelper.withProperty; @@ -120,4 +121,29 @@ public class NativeProtocolTest extends TestBaseImpl () -> StorageService.instance.isNativeTransportRunning())); } } + + @Test + public void testBinaryReflectsRpcReadiness() throws Throwable + { + try (Cluster cluster = builder().withNodes(1) + .withConfig(config -> config.with(NETWORK, GOSSIP, NATIVE_PROTOCOL) + .set("start_native_transport", "false")) + .start()) + { + IInvokableInstance i = cluster.get(1); + + // rpc is false when native transport is not enabled + assertFalse(i.callOnInstance((IIsolatedExecutor.SerializableCallable) () -> StorageService.instance.isNativeTransportRunning())); + assertFalse(i.callOnInstance((IIsolatedExecutor.SerializableCallable) () -> StorageService.instance.isRpcReady(FBUtilities.getBroadcastAddressAndPort()))); + + // but if we enable it, e.g. by nodetool enablebinary, rpc will be enabled + i.runOnInstance((IIsolatedExecutor.SerializableRunnable) () -> StorageService.instance.startNativeTransport()); + assertTrue(i.callOnInstance((IIsolatedExecutor.SerializableCallable) () -> StorageService.instance.isRpcReady(FBUtilities.getBroadcastAddressAndPort()))); + + // by calling e.g. nodetool disablebinary, rpc will be set to false again + i.runOnInstance((IIsolatedExecutor.SerializableRunnable) () -> StorageService.instance.stopNativeTransport()); + assertFalse(i.callOnInstance((IIsolatedExecutor.SerializableCallable) () -> StorageService.instance.isRpcReady(FBUtilities.getBroadcastAddressAndPort()))); + } + } } +