private void generateVariableLengthFieldSize(FieldSpec field,
                                                 Versions possibleVersions,
                                                 boolean tagged) {
        IsNullConditional.forField(field).
            alwaysEmitBlockScope(true).
            possibleVersions(possibleVersions).
            nullableVersions(field.nullableVersions()).
            ifNull(() -> {
                if (!tagged || !field.defaultString().equals("null")) {
                    VersionConditional.forVersions(fieldFlexibleVersions(field), possibleVersions).
                        ifMember(__ -> {
                            if (tagged) {
                                buffer.printf("_numTaggedFields++;%n");
                                buffer.printf("_size.addBytes(%d);%n",
                                    MessageGenerator.sizeOfUnsignedVarint(field.tag().get()));
                                buffer.printf("_size.addBytes(%d);%n", MessageGenerator.sizeOfUnsignedVarint(
                                    MessageGenerator.sizeOfUnsignedVarint(0)));
                            }
                            buffer.printf("_size.addBytes(%d);%n", MessageGenerator.sizeOfUnsignedVarint(0));
                        }).
                        ifNotMember(__ -> {
                            if (tagged) {
                                throw new RuntimeException("Tagged field " + field.name() +
                                    " should not be present in non-flexible versions.");
                            }
                            if (field.type().isString()) {
                                buffer.printf("_size.addBytes(2);%n");
                            } else if (field.type().isStruct()) {
                                buffer.printf("_size.addBytes(1);%n");
                            } else {
                                buffer.printf("_size.addBytes(4);%n");
                            }
                        }).
                        generate(buffer);
                }
            }).
            ifShouldNotBeNull(() -> {
                if (tagged) {
                    if (!field.defaultString().equals("null")) {
                        field.generateNonDefaultValueCheck(headerGenerator,
                            structRegistry, buffer, "this.", Versions.NONE);
                        buffer.incrementIndent();
                    }
                    buffer.printf("_numTaggedFields++;%n");
                    buffer.printf("_size.addBytes(%d);%n",
                        MessageGenerator.sizeOfUnsignedVarint(field.tag().get()));
                }
                if (field.type().isString()) {
                    generateStringToBytes(field.camelCaseName());
                    VersionConditional.forVersions(fieldFlexibleVersions(field), possibleVersions).
                        ifMember(__ -> {
                            headerGenerator.addImport(MessageGenerator.BYTE_UTILS_CLASS);
                            if (tagged) {
                                buffer.printf("int _stringPrefixSize = " +
                                    "ByteUtils.sizeOfUnsignedVarint(_stringBytes.length + 1);%n");
                                buffer.printf("_size.addBytes(_stringBytes.length + _stringPrefixSize + " +
                                    "ByteUtils.sizeOfUnsignedVarint(_stringPrefixSize + _stringBytes.length));%n");

                            } else {
                                buffer.printf("_size.addBytes(_stringBytes.length + " +
                                    "ByteUtils.sizeOfUnsignedVarint(_stringBytes.length + 1));%n");
                            }
                        }).
                        ifNotMember(__ -> {
                            if (tagged) {
                                throw new RuntimeException("Tagged field " + field.name() +
                                    " should not be present in non-flexible versions.");
                            }
                            buffer.printf("_size.addBytes(_stringBytes.length + 2);%n");
                        }).
                        generate(buffer);
                } else if (field.type().isArray()) {
                    if (tagged) {
                        buffer.printf("int _sizeBeforeArray = _size.totalSize();%n");
                    }
                    VersionConditional.forVersions(fieldFlexibleVersions(field), possibleVersions).
                        ifMember(__ -> {
                            headerGenerator.addImport(MessageGenerator.BYTE_UTILS_CLASS);
                            buffer.printf("_size.addBytes(ByteUtils.sizeOfUnsignedVarint(%s.size() + 1));%n",
                                field.camelCaseName());
                        }).
                        ifNotMember(__ -> {
                            buffer.printf("_size.addBytes(4);%n");
                        }).
                        generate(buffer);
                    FieldType.ArrayType arrayType = (FieldType.ArrayType) field.type();
                    FieldType elementType = arrayType.elementType();
                    if (elementType.fixedLength().isPresent()) {
                        buffer.printf("_size.addBytes(%s.size() * %d);%n",
                            field.camelCaseName(),
                            elementType.fixedLength().get());
                    } else if (elementType instanceof FieldType.ArrayType) {
                        throw new RuntimeException("Arrays of arrays are not supported " +
                            "(use a struct).");
                    } else {
                        buffer.printf("for (%s %sElement : %s) {%n",
                            elementType.getBoxedJavaType(headerGenerator),
                            field.camelCaseName(), field.camelCaseName());
                        buffer.incrementIndent();
                        generateVariableLengthArrayElementSize(fieldFlexibleVersions(field),
                            String.format("%sElement", field.camelCaseName()),
                            elementType,
                            possibleVersions);
                        buffer.decrementIndent();
                        buffer.printf("}%n");
                    }
                    if (tagged) {
                        headerGenerator.addImport(MessageGenerator.BYTE_UTILS_CLASS);
                        buffer.printf("int _arraySize = _size.totalSize() - _sizeBeforeArray;%n");
                        buffer.printf("_cache.setArraySizeInBytes(%s, _arraySize);%n",
                            field.camelCaseName());
                        buffer.printf("_size.addBytes(ByteUtils.sizeOfUnsignedVarint(_arraySize));%n");
                    }
                } else if (field.type().isBytes()) {
                    if (tagged) {
                        buffer.printf("int _sizeBeforeBytes = _size.totalSize();%n");
                    }
                    if (field.zeroCopy()) {
                        buffer.printf("_size.addZeroCopyBytes(%s.remaining());%n", field.camelCaseName());
                    } else {
                        buffer.printf("_size.addBytes(%s.length);%n", field.camelCaseName());
                    }
                    VersionConditional.forVersions(fieldFlexibleVersions(field), possibleVersions).
                        ifMember(__ -> {
                            headerGenerator.addImport(MessageGenerator.BYTE_UTILS_CLASS);
                            if (field.zeroCopy()) {
                                buffer.printf("_size.addBytes(" +
                                        "ByteUtils.sizeOfUnsignedVarint(%s.remaining() + 1));%n", field.camelCaseName());
                            } else {
                                buffer.printf("_size.addBytes(ByteUtils.sizeOfUnsignedVarint(%s.length + 1));%n",
                                    field.camelCaseName());
                            }
                        }).
                        ifNotMember(__ -> {
                            buffer.printf("_size.addBytes(4);%n");
                        }).
                        generate(buffer);
                    if (tagged) {
                        headerGenerator.addImport(MessageGenerator.BYTE_UTILS_CLASS);
                        buffer.printf("int _bytesSize = _size.totalSize() - _sizeBeforeBytes;%n");
                        buffer.printf("_size.addBytes(ByteUtils.sizeOfUnsignedVarint(_bytesSize));%n");
                    }
                } else if (field.type().isRecords()) {
                    buffer.printf("_size.addZeroCopyBytes(%s.sizeInBytes());%n", field.camelCaseName());
                    VersionConditional.forVersions(fieldFlexibleVersions(field), possibleVersions).
                        ifMember(__ -> {
                            headerGenerator.addImport(MessageGenerator.BYTE_UTILS_CLASS);
                            buffer.printf("_size.addBytes(" +
                                "ByteUtils.sizeOfUnsignedVarint(%s.sizeInBytes() + 1));%n", field.camelCaseName());
                        }).
                        ifNotMember(__ -> {
                            buffer.printf("_size.addBytes(4);%n");
                        }).
                        generate(buffer);
                } else if (field.type().isStruct()) {
                    // Adding a byte if the field is nullable. A byte works for both regular and tagged struct fields.
                    VersionConditional.forVersions(field.nullableVersions(), possibleVersions).
                        ifMember(__ -> {
                            buffer.printf("_size.addBytes(1);%n");
                        }).
                        generate(buffer);

                    if (tagged) {
                        buffer.printf("int _sizeBeforeStruct = _size.totalSize();%n", field.camelCaseName());
                        buffer.printf("this.%s.addSize(_size, _cache, _version);%n", field.camelCaseName());
                        buffer.printf("int _structSize = _size.totalSize() - _sizeBeforeStruct;%n", field.camelCaseName());
                        buffer.printf("_size.addBytes(ByteUtils.sizeOfUnsignedVarint(_structSize));%n");
                    } else {
                        buffer.printf("this.%s.addSize(_size, _cache, _version);%n", field.camelCaseName());
                    }
                } else {
                    throw new RuntimeException("unhandled type " + field.type());
                }
                if (tagged && !field.defaultString().equals("null")) {
                    buffer.decrementIndent();
                    buffer.printf("}%n");
                }
            }).
            generate(buffer);
    }
Generate the size calculator for a variable-length array element.
Array elements cannot be null.
generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java