Make sure maxMessageSize is correctly set when Thrift is configured to use a non blocking server

server/base/src/main/java/org/apache/accumulo/server/rpc/TServerUtils.java

@@ -69,6 +69,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.primitives.Ints;
+
 /**
  * Factory methods for creating Thrift server objects
  */
@@ -221,7 +223,8 @@ private static ServerAddress createThreadedSelectorServer(HostAndPort address,
       long timeBetweenThreadChecks, long maxMessageSize) throws TTransportException {
 
     final TNonblockingServerSocket transport =
-        new TNonblockingServerSocket(new InetSocketAddress(address.getHost(), address.getPort()));
+        new TNonblockingServerSocket(new InetSocketAddress(address.getHost(), address.getPort()), 0,
+            Ints.saturatedCast(maxMessageSize));
 
     TThreadedSelectorServer.Args options = new TThreadedSelectorServer.Args(transport);
 
@@ -256,7 +259,8 @@ private static ServerAddress createNonBlockingServer(HostAndPort address, TProce
       long maxMessageSize) throws TTransportException {
 
     final TNonblockingServerSocket transport =
-        new TNonblockingServerSocket(new InetSocketAddress(address.getHost(), address.getPort()));
+        new TNonblockingServerSocket(new InetSocketAddress(address.getHost(), address.getPort()), 0,
+            Ints.saturatedCast(maxMessageSize));
     final CustomNonBlockingServer.Args options = new CustomNonBlockingServer.Args(transport);
 
     options.protocolFactory(protocolFactory);

test/src/main/java/org/apache/accumulo/test/functional/ThriftMaxFrameSizeIT.java

@@ -0,0 +1,109 @@
+/*
+ * 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
+ *
+ *   https://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.accumulo.test.functional;
+
+import static org.apache.accumulo.harness.AccumuloITBase.SUNNY_DAY;
+import static org.apache.accumulo.test.functional.ConfigurableMacBase.configureForSsl;
+
+import java.time.Duration;
+
+import org.apache.accumulo.core.client.Accumulo;
+import org.apache.accumulo.core.client.AccumuloClient;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.harness.AccumuloClusterHarness;
+import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
+import org.apache.accumulo.server.rpc.ThriftServerType;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.thrift.TConfiguration;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+@Tag(SUNNY_DAY)
+public class ThriftMaxFrameSizeIT extends AccumuloClusterHarness {
+
+  private ThriftServerType serverType;
+
+  @Override
+  protected Duration defaultTimeout() {
+    return Duration.ofMinutes(1);
+  }
+
+  @Override
+  public void configureMiniCluster(MiniAccumuloConfigImpl cfg, Configuration hadoopCoreSite) {
+    cfg.setProperty(Property.GENERAL_RPC_SERVER_TYPE, serverType.name());
+    if (serverType == ThriftServerType.SSL) {
+      configureForSsl(cfg,
+          getSslDir(createTestDir(this.getClass().getName() + "_" + this.testName())));
+    }
+  }
+
+  @Nested
+  class TestDefault extends TestMaxFrameSize {
+    {
+      serverType = ThriftServerType.getDefault();
+    }
+  }
+
+  @Nested
+  class TestThreadedSelector extends TestMaxFrameSize {
+    {
+      serverType = ThriftServerType.THREADED_SELECTOR;
+    }
+  }
+
+  @Nested
+  class TestCustomHsHa extends TestMaxFrameSize {
+    {
+      serverType = ThriftServerType.CUSTOM_HS_HA;
+    }
+  }
+
+  @Nested
+  class TestThreadPool extends TestMaxFrameSize {
+    {
+      serverType = ThriftServerType.THREADPOOL;
+    }
+  }
+
+  @Nested
+  class TestSsl extends TestMaxFrameSize {
+    {
+      serverType = ThriftServerType.SSL;
+    }
+  }
+
+  protected abstract class TestMaxFrameSize {
+
+    @Test
+    public void testMaxFrameSizeLargerThanDefault() throws Exception {
+
+      // Ingest with a value width greater than the thrift default size to verify our setting works
+      // for max frame wize
+      try (AccumuloClient accumuloClient = Accumulo.newClient().from(getClientProps()).build()) {
+        String table = getUniqueNames(1)[0];
+        ReadWriteIT.ingest(accumuloClient, 1, 1, TConfiguration.DEFAULT_MAX_FRAME_SIZE + 1, 0,
+            table);
+        ReadWriteIT.verify(accumuloClient, 1, 1, TConfiguration.DEFAULT_MAX_FRAME_SIZE + 1, 0,
+            table);
+      }
+    }
+  }
+
+}