KAFKA-13277; Fix size calculation for tagged string fields in message generator (#11308)

Reviewers: Colin P. McCabe <cmccabe@apache.org>

clients/src/test/java/org/apache/kafka/common/message/MessageTest.java

@@ -1161,6 +1161,22 @@ public final class MessageTest {
         verifyWriteSucceeds((short) 6, createTopics);
     }
 
+    @Test
+    public void testLongTaggedString() throws Exception {
+        char[] chars = new char[1024];
+        Arrays.fill(chars, 'a');
+        String longString = new String(chars);
+        SimpleExampleMessageData message = new SimpleExampleMessageData()
+                .setMyString(longString);
+        ObjectSerializationCache cache = new ObjectSerializationCache();
+        short version = 1;
+        int size = message.size(cache, version);
+        ByteBuffer buf = ByteBuffer.allocate(size);
+        ByteBufferAccessor byteBufferAccessor = new ByteBufferAccessor(buf);
+        message.write(byteBufferAccessor, cache, version);
+        assertEquals(size, buf.position());
+    }
+
     private void verifyWriteRaisesNpe(short version, Message message) {
         ObjectSerializationCache cache = new ObjectSerializationCache();
         assertThrows(NullPointerException.class, () -> {

generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java

@@ -1156,7 +1156,8 @@ public final class MessageDataGenerator implements MessageClassGenerator {
                                 buffer.printf("int _stringPrefixSize = " +
                                     "ByteUtils.sizeOfUnsignedVarint(_stringBytes.length + 1);%n");
                                 buffer.printf("_size.addBytes(_stringBytes.length + _stringPrefixSize + " +
-                                    "ByteUtils.sizeOfUnsignedVarint(_stringPrefixSize));%n");
+                                    "ByteUtils.sizeOfUnsignedVarint(_stringPrefixSize + _stringBytes.length));%n");
+
                             } else {
                                 buffer.printf("_size.addBytes(_stringBytes.length + " +
                                     "ByteUtils.sizeOfUnsignedVarint(_stringBytes.length + 1));%n");