Remove buggy thrift max message length option

patch by jbellis; reviewed by tjake for CASSANDRA-5529

CHANGES.txt

@@ -1,4 +1,5 @@
 1.1.12
+ * Remove buggy thrift max message length option (CASSANDRA-5529)
  * Add retry mechanism to OTC for non-droppable_verbs (CASSANDRA-5393)
  * Use allocator information to improve memtable memory usage estimate 
    (CASSANDRA-5497)
@@ -6,6 +7,7 @@
  * Fix Bound intersection computation (CASSANDRA-5551)
  * Fix NPE in Pig's widerow mode (CASSANDRA-5488)
 
+
 1.1.11
  * Fix trying to load deleted row into row cache on startup (CASSANDRA-4463)
  * Update offline scrub for 1.0 -> 1.1 directory structure (CASSANDRA-5195)

conf/cassandra.yaml

@@ -330,15 +330,11 @@ rpc_server_type: sync
 # rpc_send_buff_size_in_bytes:
 # rpc_recv_buff_size_in_bytes:
 
-# Frame size for thrift (maximum field length).
+# Frame size for thrift (maximum message length).
 # 0 disables TFramedTransport in favor of TSocket. This option
 # is deprecated; we strongly recommend using Framed mode.
 thrift_framed_transport_size_in_mb: 15
 
-# The max length of a thrift message, including all fields and
-# internal thrift overhead.
-thrift_max_message_length_in_mb: 16
-
 # Set to true to have Cassandra create a hard link to each sstable
 # flushed or streamed locally in a backups/ subdirectory of the
 # Keyspace data.  Removing these links is the operator's

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

@@ -77,7 +77,9 @@ public class Config
     public Integer rpc_send_buff_size_in_bytes;
     public Integer rpc_recv_buff_size_in_bytes;
 
+    @Deprecated
     public Integer thrift_max_message_length_in_mb = 16;
+
     public Integer thrift_framed_transport_size_in_mb = 15;
     public Boolean snapshot_before_compaction = false;
     public Boolean auto_snapshot = true;

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

@@ -316,11 +316,6 @@ public class DatabaseDescriptor
             if (conf.thrift_framed_transport_size_in_mb <= 0)
                 throw new ConfigurationException("thrift_framed_transport_size_in_mb must be positive");
 
-            if (conf.thrift_framed_transport_size_in_mb > 0 && conf.thrift_max_message_length_in_mb < conf.thrift_framed_transport_size_in_mb)
-            {
-                throw new ConfigurationException("thrift_max_message_length_in_mb must be greater than thrift_framed_transport_size_in_mb when using TFramedTransport");
-            }
-
             /* end point snitch */
             if (conf.endpoint_snitch == null)
             {
@@ -582,11 +577,6 @@ public class DatabaseDescriptor
         return authority;
     }
 
-    public static int getThriftMaxMessageLength()
-    {
-        return conf.thrift_max_message_length_in_mb * 1024 * 1024;
-    }
-
     public static int getThriftFramedTransportSize()
     {
         return conf.thrift_framed_transport_size_in_mb * 1024 * 1024;

src/java/org/apache/cassandra/hadoop/ColumnFamilyOutputFormat.java

@@ -118,7 +118,7 @@ public class ColumnFamilyOutputFormat extends OutputFormat<ByteBuffer,List<Mutat
     }
 
     /** Fills the deprecated OutputFormat interface for streaming. */
-    @Deprecated @Override
+    @Deprecated
     public ColumnFamilyRecordWriter getRecordWriter(org.apache.hadoop.fs.FileSystem filesystem, org.apache.hadoop.mapred.JobConf job, String name, org.apache.hadoop.util.Progressable progress) throws IOException
     {
         return new ColumnFamilyRecordWriter(job, new Progressable(progress));
@@ -155,7 +155,7 @@ public class ColumnFamilyOutputFormat extends OutputFormat<ByteBuffer,List<Mutat
     {
         logger.debug("Creating authenticated client for CF output format");
         TTransport transport = ConfigHelper.getOutputTransportFactory(conf).openTransport(socket, conf);
-        TBinaryProtocol binaryProtocol = new TBinaryProtocol(transport, ConfigHelper.getThriftMaxMessageLength(conf));
+        TBinaryProtocol binaryProtocol = new TBinaryProtocol(transport, true, true);
         Cassandra.Client client = new Cassandra.Client(binaryProtocol);
         client.set_keyspace(ConfigHelper.getOutputKeyspace(conf));
         if (ConfigHelper.getOutputKeyspaceUserName(conf) != null)

src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java

@@ -162,7 +162,7 @@ public class ColumnFamilyRecordReader extends RecordReader<ByteBuffer, SortedMap
             String location = getLocation();
             socket = new TSocket(location, ConfigHelper.getInputRpcPort(conf));
             TTransport transport = ConfigHelper.getInputTransportFactory(conf).openTransport(socket, conf);
-            TBinaryProtocol binaryProtocol = new TBinaryProtocol(transport, ConfigHelper.getThriftMaxMessageLength(conf));
+            TBinaryProtocol binaryProtocol = new TBinaryProtocol(transport, true, true);
             client = new Cassandra.Client(binaryProtocol);
 
             // log in

src/java/org/apache/cassandra/hadoop/ConfigHelper.java

@@ -77,7 +77,6 @@ public class ConfigHelper
     private static final String INPUT_TRANSPORT_FACTORY_CLASS = "cassandra.input.transport.factory.class";
     private static final String OUTPUT_TRANSPORT_FACTORY_CLASS = "cassandra.output.transport.factory.class";
     private static final String THRIFT_FRAMED_TRANSPORT_SIZE_IN_MB = "cassandra.thrift.framed.size_mb";
-    private static final String THRIFT_MAX_MESSAGE_LENGTH_IN_MB = "cassandra.thrift.message.max_size_mb";
 
     private static final Logger logger = LoggerFactory.getLogger(ConfigHelper.class);
 
@@ -482,18 +481,10 @@ public class ConfigHelper
         return conf.getInt(THRIFT_FRAMED_TRANSPORT_SIZE_IN_MB, 15) * 1024 * 1024; // 15MB is default in Cassandra
     }
 
+    @Deprecated
     public static void setThriftMaxMessageLengthInMb(Configuration conf, int maxMessageSizeInMB)
     {
-        conf.setInt(THRIFT_MAX_MESSAGE_LENGTH_IN_MB, maxMessageSizeInMB);
+        // SEE CASSANDRA-5529
-    }
-
-    /**
-     * @param conf The configuration to use.
-     * @return Value (converts MBs to Bytes) set by {@link setThriftMaxMessageLengthInMb(Configuration, int)} or default of 16MB
-     */
-    public static int getThriftMaxMessageLength(Configuration conf)
-    {
-        return conf.getInt(THRIFT_MAX_MESSAGE_LENGTH_IN_MB, 16) * 1024 * 1024; // 16MB is default in Cassandra
     }
 
     public static CompressionParameters getOutputCompressionParamaters(Configuration conf)
@@ -557,7 +548,7 @@ public class ConfigHelper
         {
             TSocket socket = new TSocket(host, port);
             TTransport transport = getInputTransportFactory(conf).openTransport(socket, conf);
-            return new Cassandra.Client(new TBinaryProtocol(transport, getThriftMaxMessageLength(conf)));
+            return new Cassandra.Client(new TBinaryProtocol(transport, true, true));
         }
         catch (LoginException e)
         {

src/java/org/apache/cassandra/thrift/CassandraDaemon.java

@@ -125,7 +125,7 @@ public class CassandraDaemon extends org.apache.cassandra.service.AbstractCassan
             logger.info(String.format("Binding thrift service to %s:%s", listenAddr, listenPort));
 
             // Protocol factory
-            TProtocolFactory tProtocolFactory = new TBinaryProtocol.Factory(true, true, DatabaseDescriptor.getThriftMaxMessageLength());
+            TProtocolFactory tProtocolFactory = new TBinaryProtocol.Factory(true, true);
 
             // Transport factory
             int tFramedTransportSize = DatabaseDescriptor.getThriftFramedTransportSize();

src/java/org/apache/cassandra/thrift/TBinaryProtocol.java

@@ -31,20 +31,6 @@ import java.nio.ByteBuffer;
 
 public class TBinaryProtocol extends org.apache.thrift.protocol.TBinaryProtocol
 {
-
-    public TBinaryProtocol(TTransport trans)
-    {
-        this(trans, false, true);
-    }
-
-    public TBinaryProtocol(TTransport trans, int readLength)
-    {
-        this(trans);
-
-        if (readLength > 0)
-            setReadLength(readLength);
-    }
-
     public TBinaryProtocol(TTransport trans, boolean strictRead, boolean strictWrite)
     {
         super(trans);
@@ -64,11 +50,6 @@ public class TBinaryProtocol extends org.apache.thrift.protocol.TBinaryProtocol
             super(strictRead, strictWrite, 0);
         }
 
-        public Factory(boolean strictRead, boolean strictWrite, int readLength)
-        {
-            super(strictRead, strictWrite, readLength);
-        }
-
         public TProtocol getProtocol(TTransport trans)
         {
             TBinaryProtocol protocol = new TBinaryProtocol(trans, strictRead_, strictWrite_);