diff --git a/build.xml b/build.xml index d50007e46a..d0a2f868fd 100644 --- a/build.xml +++ b/build.xml @@ -1812,6 +1812,7 @@ + diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java index 85a002f142..271f763896 100644 --- a/src/java/org/apache/cassandra/service/CassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java @@ -408,11 +408,6 @@ public class CassandraDaemon // due to scheduling errors or race conditions ScheduledExecutors.optionalTasks.scheduleWithFixedDelay(ColumnFamilyStore.getBackgroundCompactionTaskSubmitter(), 5, 1, TimeUnit.MINUTES); - // Thrift - InetAddress rpcAddr = DatabaseDescriptor.getRpcAddress(); - int rpcPort = DatabaseDescriptor.getRpcPort(); - int listenBacklog = DatabaseDescriptor.getRpcListenBacklog(); - thriftServer = new ThriftServer(rpcAddr, rpcPort, listenBacklog); initializeNativeTransport(); completeSetup(); @@ -420,6 +415,13 @@ public class CassandraDaemon public void initializeNativeTransport() { + // Thrift + InetAddress rpcAddr = DatabaseDescriptor.getRpcAddress(); + int rpcPort = DatabaseDescriptor.getRpcPort(); + int listenBacklog = DatabaseDescriptor.getRpcListenBacklog(); + if (thriftServer == null) + thriftServer = new ThriftServer(rpcAddr, rpcPort, listenBacklog); + // Native transport if (nativeTransportService == null) nativeTransportService = new NativeTransportService(); @@ -568,6 +570,12 @@ public class CassandraDaemon nativeTransportService.destroy(); nativeTransportService = null; } + + if (thriftServer != null) + { + thriftServer.stop(); + thriftServer = null; + } } @@ -680,14 +688,27 @@ public class CassandraDaemon if (nativeTransportService == null) throw new IllegalStateException("setup() must be called first for CassandraDaemon"); - else - nativeTransportService.start(); + + nativeTransportService.start(); + + if (thriftServer == null) + throw new IllegalStateException("thrift transport should be set up before it can be started"); + thriftServer.start(); } public void stopNativeTransport() { if (nativeTransportService != null) + { nativeTransportService.stop(); + nativeTransportService = null; + } + + if (thriftServer != null) + { + thriftServer.stop(); + thriftServer = null; + } } public boolean isNativeTransportRunning() diff --git a/test/distributed/org/apache/cassandra/distributed/test/ThriftClientTest.java b/test/distributed/org/apache/cassandra/distributed/test/ThriftClientTest.java new file mode 100644 index 0000000000..c5d5b9b335 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/ThriftClientTest.java @@ -0,0 +1,51 @@ +package org.apache.cassandra.distributed.test; + +import java.io.IOException; +import java.util.Arrays; +import java.util.Collections; + +import org.junit.Test; + +import org.apache.cassandra.db.marshal.CompositeType; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.distributed.api.QueryResults; +import org.apache.cassandra.distributed.api.SimpleQueryResult; +import org.apache.cassandra.distributed.shared.AssertUtils; +import org.apache.cassandra.thrift.Column; +import org.apache.cassandra.thrift.ColumnOrSuperColumn; +import org.apache.cassandra.thrift.Mutation; +import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.thrift.TException; + +public class ThriftClientTest extends TestBaseImpl +{ + @Test + public void writeThenReadCQL() throws IOException, TException + { + try (Cluster cluster = init(Cluster.build(1).withConfig(c -> c.with(Feature.NATIVE_PROTOCOL)).start())) + { + cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, value int, PRIMARY KEY (pk))"); + + ThriftClientUtils.thriftClient(cluster.get(1), thrift -> { + thrift.set_keyspace(KEYSPACE); + Mutation mutation = new Mutation(); + ColumnOrSuperColumn csoc = new ColumnOrSuperColumn(); + Column column = new Column(); + column.setName(CompositeType.build(ByteBufferUtil.bytes("value"))); + column.setValue(ByteBufferUtil.bytes(0)); + column.setTimestamp(System.currentTimeMillis()); + csoc.setColumn(column); + mutation.setColumn_or_supercolumn(csoc); + + thrift.batch_mutate(Collections.singletonMap(ByteBufferUtil.bytes(0), + Collections.singletonMap("tbl", Arrays.asList(mutation))), + org.apache.cassandra.thrift.ConsistencyLevel.ALL); + }); + + SimpleQueryResult qr = cluster.coordinator(1).executeWithResult("SELECT * FROM " + KEYSPACE + ".tbl", ConsistencyLevel.ALL); + AssertUtils.assertRows(qr, QueryResults.builder().row(0, 0).build()); + } + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/ThriftClientUtils.java b/test/distributed/org/apache/cassandra/distributed/test/ThriftClientUtils.java new file mode 100644 index 0000000000..51b153cb26 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/ThriftClientUtils.java @@ -0,0 +1,38 @@ +package org.apache.cassandra.distributed.test; + +import java.net.InetSocketAddress; + +import org.apache.cassandra.distributed.api.IInstance; +import org.apache.cassandra.thrift.Cassandra; +import org.apache.cassandra.thrift.TFramedTransportFactory; +import org.apache.thrift.TException; +import org.apache.thrift.protocol.TBinaryProtocol; +import org.apache.thrift.transport.TTransport; + +public final class ThriftClientUtils +{ + private ThriftClientUtils() + { + + } + + public static void thriftClient(IInstance instance, ThriftConsumer fn) throws TException + { + //TODO dtest APIs only expose native address, doesn't expose all addresses we listen to, so assume the default thrift port + thriftClient(new InetSocketAddress(instance.broadcastAddress().getAddress(), 9160), fn); + } + + public static void thriftClient(InetSocketAddress address, ThriftConsumer fn) throws TException + { + try (TTransport transport = new TFramedTransportFactory().openTransport(address.getAddress().getHostAddress(), address.getPort())) + { + Cassandra.Client client = new Cassandra.Client(new TBinaryProtocol(transport)); + fn.accept(client); + } + } + + public interface ThriftConsumer + { + void accept(Cassandra.Client client) throws TException; + } +}