From 25fd7bd84f1931d2a44e90e629f794c4cd11aa46 Mon Sep 17 00:00:00 2001 From: David Capwell Date: Fri, 24 Jul 2020 18:16:37 -0700 Subject: [PATCH] Add support in jvm dtest to test thrift patch by David Capwell; reviewed by Alex Petrov,Jon Meredith for CASSANDRA-15967 --- build.xml | 1 + .../cassandra/service/CassandraDaemon.java | 24 +++++++-- .../distributed/test/ThriftClientTest.java | 51 +++++++++++++++++++ .../distributed/test/ThriftClientUtils.java | 38 ++++++++++++++ 4 files changed, 109 insertions(+), 5 deletions(-) create mode 100644 test/distributed/org/apache/cassandra/distributed/test/ThriftClientTest.java create mode 100644 test/distributed/org/apache/cassandra/distributed/test/ThriftClientUtils.java diff --git a/build.xml b/build.xml index 4c05a941cd..84172aa10a 100644 --- a/build.xml +++ b/build.xml @@ -1866,6 +1866,7 @@ + diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java index a317ab316e..7d85d9db19 100644 --- a/src/java/org/apache/cassandra/service/CassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java @@ -22,6 +22,7 @@ import java.io.IOException; import java.lang.management.ManagementFactory; import java.lang.management.MemoryPoolMXBean; import java.net.InetAddress; +import java.net.InetSocketAddress; import java.net.UnknownHostException; import java.rmi.AccessException; import java.rmi.AlreadyBoundException; @@ -372,11 +373,6 @@ public class CassandraDaemon if (sizeRecorderInterval > 0) ScheduledExecutors.optionalTasks.scheduleWithFixedDelay(SizeEstimatesRecorder.instance, 30, sizeRecorderInterval, TimeUnit.SECONDS); - // Thrift - InetAddress rpcAddr = DatabaseDescriptor.getRpcAddress(); - int rpcPort = DatabaseDescriptor.getRpcPort(); - int listenBacklog = DatabaseDescriptor.getRpcListenBacklog(); - thriftServer = new ThriftServer(rpcAddr, rpcPort, listenBacklog); initializeNativeTransport(); completeSetup(); @@ -384,6 +380,12 @@ public class CassandraDaemon public void initializeNativeTransport() { + // Thrift + InetAddress rpcAddr = DatabaseDescriptor.getRpcAddress(); + int rpcPort = DatabaseDescriptor.getRpcPort(); + int listenBacklog = DatabaseDescriptor.getRpcListenBacklog(); + thriftServer = new ThriftServer(rpcAddr, rpcPort, listenBacklog); + // Native transport InetAddress nativeAddr = DatabaseDescriptor.getRpcAddress(); int nativePort = DatabaseDescriptor.getNativeTransportPort(); @@ -398,6 +400,12 @@ public class CassandraDaemon throw new IllegalStateException("native transport should be set up before it can be started"); nativeServer.start(); + logger.info("Native server running on {}", new InetSocketAddress(DatabaseDescriptor.getRpcAddress(), DatabaseDescriptor.getNativeTransportPort())); + + if (thriftServer == null) + throw new IllegalStateException("thrift transport should be set up before it can be started"); + thriftServer.start(); + logger.info("Thrift server running on {}", new InetSocketAddress(DatabaseDescriptor.getRpcAddress(), DatabaseDescriptor.getRpcPort())); } private void validateTransportsCanStart() @@ -567,6 +575,12 @@ public class CassandraDaemon nativeServer.stopAndAwaitTermination(); nativeServer = null; } + + if (thriftServer != null) + { + thriftServer.stopAndAwaitTermination(); + thriftServer = null; + } } 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; + } +}