Merge branch 'cassandra-2.1' into cassandra-2.2

CHANGES.txt

@@ -13,6 +13,7 @@
  * Disable reloading of GossipingPropertyFileSnitch (CASSANDRA-9474)
  * Verify tables in pseudo-system keyspaces at startup (CASSANDRA-10761)
 Merged from 2.1:
+ * Sane default (200Mbps) for inter-DC streaming througput (CASSANDRA-9708)
  * Match cassandra-loader options in COPY FROM (CASSANDRA-9303)
  * Fix binding to any address in CqlBulkRecordWriter (CASSANDRA-9309)
  * cqlsh fails to decode utf-8 characters for text typed columns (CASSANDRA-10875)

NEWS.txt

@@ -36,6 +36,13 @@ Operations
     - GC logging is now enabled by default (but you can disable it if you want by
       commenting the relevant lines of the cassandra-env file).
 
+Upgrading
+---------
+    - The default for the inter-DC stream throughput setting
+      (inter_dc_stream_throughput_outbound_megabits_per_sec in cassandra.yaml) is
+      the same than the one for intra-DC one (200Mbps) instead of being unlimited.
+      Having it unlimited was never intended and was a bug.
+
 New features
 ------------
     - Time windows in DTCS are now limited to 1 day by default to be able to

conf/cassandra.yaml

@@ -640,7 +640,8 @@ sstable_preemptive_open_interval_in_mb: 50
 # this setting allows users to throttle inter dc stream throughput in addition
 # to throttling all network stream traffic as configured with
 # stream_throughput_outbound_megabits_per_sec
-# inter_dc_stream_throughput_outbound_megabits_per_sec:
+# When unset, the default is 200 Mbps or 25 MB/s
+# inter_dc_stream_throughput_outbound_megabits_per_sec: 200
 
 # How long the coordinator should wait for read operations to complete
 read_request_timeout_in_ms: 5000

src/java/org/apache/cassandra/config/Config.java

@@ -158,7 +158,7 @@ public class Config
     public Integer max_streaming_retries = 3;
 
     public volatile Integer stream_throughput_outbound_megabits_per_sec = 200;
-    public volatile Integer inter_dc_stream_throughput_outbound_megabits_per_sec = 0;
+    public volatile Integer inter_dc_stream_throughput_outbound_megabits_per_sec = 200;
 
     public String[] data_file_directories;
 

src/java/org/apache/cassandra/service/StorageService.java

@@ -1134,6 +1134,18 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
         return DatabaseDescriptor.getStreamThroughputOutboundMegabitsPerSec();
     }
 
+    public void setInterDCStreamThroughputMbPerSec(int value)
+    {
+        DatabaseDescriptor.setInterDCStreamThroughputOutboundMegabitsPerSec(value);
+        logger.info("setinterdcstreamthroughput: throttle set to {}", value);
+    }
+
+    public int getInterDCStreamThroughputMbPerSec()
+    {
+        return DatabaseDescriptor.getInterDCStreamThroughputOutboundMegabitsPerSec();
+    }
+
+
     public int getCompactionThroughputMbPerSec()
     {
         return DatabaseDescriptor.getCompactionThroughputMbPerSec();

src/java/org/apache/cassandra/service/StorageServiceMBean.java

@@ -485,6 +485,9 @@ public interface StorageServiceMBean extends NotificationEmitter
     public void setStreamThroughputMbPerSec(int value);
     public int getStreamThroughputMbPerSec();
 
+    public void setInterDCStreamThroughputMbPerSec(int value);
+    public int getInterDCStreamThroughputMbPerSec();
+
     public int getCompactionThroughputMbPerSec();
     public void setCompactionThroughputMbPerSec(int value);
 

src/java/org/apache/cassandra/tools/BulkLoader.java

@@ -51,6 +51,7 @@ public class BulkLoader
     private static final String USER_OPTION = "username";
     private static final String PASSWD_OPTION = "password";
     private static final String THROTTLE_MBITS = "throttle";
+    private static final String INTER_DC_THROTTLE_MBITS = "inter-dc-throttle";
 
     /* client encryption options */
     private static final String SSL_TRUSTSTORE = "truststore";
@@ -83,6 +84,7 @@ public class BulkLoader
                 handler,
                 options.connectionsPerHost);
         DatabaseDescriptor.setStreamThroughputOutboundMegabitsPerSec(options.throttle);
+        DatabaseDescriptor.setInterDCStreamThroughputOutboundMegabitsPerSec(options.interDcThrottle);
         StreamResultFuture future = null;
 
         ProgressIndicator indicator = new ProgressIndicator();
@@ -308,6 +310,7 @@ public class BulkLoader
         public String user;
         public String passwd;
         public int throttle = 0;
+        public int interDcThrottle = 0;
         public int storagePort;
         public int sslStoragePort;
         public EncryptionOptions encOptions = new EncryptionOptions.ClientEncryptionOptions();
@@ -435,6 +438,7 @@ public class BulkLoader
                 opts.storagePort = config.storage_port;
                 opts.sslStoragePort = config.ssl_storage_port;
                 opts.throttle = config.stream_throughput_outbound_megabits_per_sec;
+                opts.interDcThrottle = config.inter_dc_stream_throughput_outbound_megabits_per_sec;
                 opts.encOptions = config.client_encryption_options;
                 opts.serverEncOptions = config.server_encryption_options;
 
@@ -443,6 +447,11 @@ public class BulkLoader
                     opts.throttle = Integer.parseInt(cmd.getOptionValue(THROTTLE_MBITS));
                 }
 
+                if (cmd.hasOption(INTER_DC_THROTTLE_MBITS))
+                {
+                    opts.interDcThrottle = Integer.parseInt(cmd.getOptionValue(INTER_DC_THROTTLE_MBITS));
+                }
+
                 if (cmd.hasOption(SSL_TRUSTSTORE))
                 {
                     opts.encOptions.truststore = cmd.getOptionValue(SSL_TRUSTSTORE);
@@ -511,6 +520,7 @@ public class BulkLoader
             options.addOption("d",  INITIAL_HOST_ADDRESS_OPTION, "initial hosts", "Required. try to connect to these hosts (comma separated) initially for ring information");
             options.addOption("p",  NATIVE_PORT_OPTION, "rpc port", "port used for native connection (default 9042)");
             options.addOption("t",  THROTTLE_MBITS, "throttle", "throttle speed in Mbits (default unlimited)");
+            options.addOption("idct",  INTER_DC_THROTTLE_MBITS, "inter-dc-throttle", "inter-datacenter throttle speed in Mbits (default unlimited)");
             options.addOption("u",  USER_OPTION, "username", "username for cassandra authentication");
             options.addOption("pw", PASSWD_OPTION, "password", "password for cassandra authentication");
             options.addOption("cph", CONNECTIONS_PER_HOST, "connectionsPerHost", "number of concurrent connections-per-host.");
@@ -537,7 +547,7 @@ public class BulkLoader
                             "you will need to have the files Standard1-g-1-Data.db and Standard1-g-1-Index.db into a directory /path/to/Keyspace1/Standard1/.";
             String footer = System.lineSeparator() +
                             "You can provide cassandra.yaml file with -f command line option to set up streaming throughput, client and server encryption options. " +
-                            "Only stream_throughput_outbound_megabits_per_sec, server_encryption_options and client_encryption_options are read from yaml. " +
+                            "Only stream_throughput_outbound_megabits_per_sec, inter_dc_stream_throughput_outbound_megabits_per_sec, server_encryption_options and client_encryption_options are read from yaml. " +
                             "You can override options read from cassandra.yaml with corresponding command line options.";
             new HelpFormatter().printHelp(usage, header, options, footer);
         }

src/java/org/apache/cassandra/tools/NodeProbe.java

@@ -941,7 +941,15 @@ public class NodeProbe implements AutoCloseable
         return ssProxy.getStreamThroughputMbPerSec();
     }
 
-    public double getTraceProbability() {return ssProxy.getTraceProbability();}
+    public int getInterDCStreamThroughput()
+    {
+        return ssProxy.getInterDCStreamThroughputMbPerSec();
+    }
+
+    public double getTraceProbability()
+    {
+        return ssProxy.getTraceProbability();
+    }
 
     public int getExceptionCount()
     {
@@ -983,6 +991,11 @@ public class NodeProbe implements AutoCloseable
         ssProxy.setStreamThroughputMbPerSec(value);
     }
 
+    public void setInterDCStreamThroughput(int value)
+    {
+        ssProxy.setInterDCStreamThroughputMbPerSec(value);
+    }
+
     public void setTraceProbability(double value)
     {
         ssProxy.setTraceProbability(value);

src/java/org/apache/cassandra/tools/NodeTool.java

@@ -81,6 +81,7 @@ public class NodeTool
                 GetCompactionThroughput.class,
                 GetStreamThroughput.class,
                 GetTraceProbability.class,
+                GetInterDCStreamThroughput.class,
                 GetEndpoints.class,
                 GetSSTables.class,
                 GossipInfo.class,
@@ -102,6 +103,7 @@ public class NodeTool
                 SetCompactionThreshold.class,
                 SetCompactionThroughput.class,
                 SetStreamThroughput.class,
+                SetInterDCStreamThroughput.class,
                 SetTraceProbability.class,
                 Snapshot.class,
                 ListSnapshots.class,

src/java/org/apache/cassandra/tools/nodetool/GetInterDCStreamThroughput.java

@@ -0,0 +1,33 @@
+/*
+ * 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.tools.nodetool;
+
+import io.airlift.command.Command;
+
+import org.apache.cassandra.tools.NodeProbe;
+import org.apache.cassandra.tools.NodeTool.NodeToolCmd;
+
+@Command(name = "getinterdcstreamthroughput", description = "Print the Mb/s throughput cap for inter-datacenter streaming in the system")
+public class GetInterDCStreamThroughput extends NodeToolCmd
+{
+    @Override
+    public void execute(NodeProbe probe)
+    {
+        System.out.println("Current inter-datacenter stream throughput: " + probe.getInterDCStreamThroughput() + " Mb/s");
+    }
+}

src/java/org/apache/cassandra/tools/nodetool/SetInterDCStreamThroughput.java

@@ -0,0 +1,37 @@
+/*
+ * 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.tools.nodetool;
+
+import io.airlift.command.Arguments;
+import io.airlift.command.Command;
+
+import org.apache.cassandra.tools.NodeProbe;
+import org.apache.cassandra.tools.NodeTool.NodeToolCmd;
+
+@Command(name = "setinterdcstreamthroughput", description = "Set the Mb/s throughput cap for inter-datacenter streaming in the system, or 0 to disable throttling")
+public class SetInterDCStreamThroughput extends NodeToolCmd
+{
+    @Arguments(title = "inter_dc_stream_throughput", usage = "<value_in_mb>", description = "Value in Mb, 0 to disable throttling", required = true)
+    private Integer interDCStreamThroughput = null;
+
+    @Override
+    public void execute(NodeProbe probe)
+    {
+        probe.setInterDCStreamThroughput(interDCStreamThroughput);
+    }
+}