Merge branch 'cassandra-2.2' into cassandra-3.0

build.xml

@@ -1812,6 +1812,7 @@
       <jar jarfile="${build.dir}/dtest-${base.version}.jar">
           <zipgroupfileset dir="${build.lib}" includes="*.jar" excludes="META-INF/*.SF"/>
           <fileset dir="${build.classes.main}"/>
+          <fileset dir="${build.classes.thrift}"/>
           <fileset dir="${test.classes}"/>
           <fileset dir="${test.conf}" />
       </jar>

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()

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());
+        }
+    }
+}

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;
+    }
+}