GEODE-8542: java.lang.IllegalStateException: tcp message exceeded max… (

#5562)

* GEODE-8542: java.lang.IllegalStateException: tcp message exceeded max size of 16777215

Limit the size of message chunks to the maximum message size allowed
by org.apache.geode.internal.tcp.Connection.

* implementing the change that Bill requested

geode-core/src/main/java/org/apache/geode/internal/tcp/MsgStreamer.java

@@ -27,6 +27,7 @@
 import it.unimi.dsi.fastutil.objects.ObjectIterator;
 
 import org.apache.geode.DataSerializer;
+import org.apache.geode.annotations.VisibleForTesting;
 import org.apache.geode.distributed.internal.DMStats;
 import org.apache.geode.distributed.internal.DistributionMessage;
 import org.apache.geode.internal.Assert;
@@ -129,7 +130,8 @@ public List<?> getSentConnections() {
     this.stats = stats;
     this.msg = msg;
     this.cons = cons;
-    this.buffer = bufferPool.acquireDirectSenderBuffer(sendBufferSize);
+    int bufferSize = Math.min(sendBufferSize, Connection.MAX_MSG_SIZE);
+    this.buffer = bufferPool.acquireDirectSenderBuffer(bufferSize);
     this.buffer.clear();
     this.buffer.position(Connection.MSG_HEADER_BYTES);
     this.msgId = MsgIdGenerator.NO_MSG_ID;
@@ -349,6 +351,11 @@ public void realFlush(boolean lastFlushForMessage) {
     this.buffer.position(Connection.MSG_HEADER_BYTES);
   }
 
+  @VisibleForTesting
+  protected ByteBuffer getBuffer() {
+    return buffer;
+  }
+
   @Override
   public void close() throws IOException {
     try {

geode-core/src/test/java/org/apache/geode/internal/tcp/MsgStreamerTest.java

@@ -77,6 +77,31 @@ public void streamerListReleaseWithException() throws IOException {
     verify(pool, times(2)).releaseSenderBuffer(isA(ByteBuffer.class));
   }
 
+  @Test
+  public void streamerRespectsMaxMessageSize() {
+    InternalDistributedMember member1;
+    member1 = new InternalDistributedMember("localhost", 1234);
+
+    DistributionMessage message = new SerialAckedMessage();
+    message.setRecipients(Arrays.asList(member1));
+
+    when(connection1.getRemoteAddress()).thenReturn(member1);
+    when(connection1.getRemoteVersion()).thenReturn(KnownVersion.CURRENT);
+    // create a streamer for a Connection that has a buffer size that's larger than the
+    // biggest message we can actually send. This is picked up by the MsgStreamer to allocate
+    // a buffer
+    when(connection1.getSendBufferSize()).thenReturn(Connection.MAX_MSG_SIZE + 1);
+    List<Connection> connections = Arrays.asList(connection1);
+
+    final BaseMsgStreamer msgStreamer =
+        MsgStreamer.create(connections, message, false, stats, pool);
+    // the streamer ought to have limited the message buffer to MAX_MSG_SIZE
+    assertThat(((MsgStreamer) msgStreamer).getBuffer().capacity())
+        .isEqualTo(Connection.MAX_MSG_SIZE);
+  }
+
+
+
   protected BaseMsgStreamer createMsgStreamer(boolean mixedDestinationVersions) {
 
     InternalDistributedMember member1, member2;
@@ -92,9 +117,9 @@ protected BaseMsgStreamer createMsgStreamer(boolean mixedDestinationVersions) {
     when(connection2.getRemoteAddress()).thenReturn(member2);
     when(connection2.getSendBufferSize()).thenReturn(Connection.SMALL_BUFFER_SIZE);
     if (mixedDestinationVersions) {
-      when(connection1.getRemoteVersion()).thenReturn(KnownVersion.GEODE_1_12_0);
+      when(connection2.getRemoteVersion()).thenReturn(KnownVersion.GEODE_1_12_0);
     } else {
-      when(connection1.getRemoteVersion()).thenReturn(KnownVersion.CURRENT);
+      when(connection2.getRemoteVersion()).thenReturn(KnownVersion.CURRENT);
     }
     List<Connection> connections = Arrays.asList(connection1, connection2);