Repository: apache/kafka
File: coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java (Java)

Code before:

                // Prepare the records.
                List<SimpleRecord> recordsToAppend = new ArrayList<>(records.size());
                for (U record : records) {
                    recordsToAppend.add(new SimpleRecord(
                        currentTimeMs,
                        serializer.serializeKey(record),
                        serializer.serializeValue(record)
                    ));
                }

                if (isAtomic) {
                    // Compute the estimated size of the records.
                    int estimatedSize = AbstractRecords.estimateSizeInBytes(
                        currentBatch.builder.magic(),
                        compression.type(),
                        recordsToAppend
                    );
                    int estimatedSizeUpperBound = AbstractRecords.estimateSizeInBytes(
                        currentBatch.builder.magic(),
                        Compression.none().build().type(),
                        recordsToAppend
                    );

                    // Check if the current batch has enough space. We check this before
                    // replaying the records in order to avoid having to revert back
                    // changes if the records do not fit within a batch.
                    if (estimatedSize > currentBatch.builder.maxAllowedBytes()) {
                        throw new RecordTooLargeException("Message batch size is " + estimatedSize +
                            " bytes in append to partition " + tp + " which exceeds the maximum " +
                            "configured size of " + currentBatch.maxBatchSize + ".");
                    }

                    if (!currentBatch.builder.hasRoomFor(estimatedSizeUpperBound)) {
                        // Otherwise, we write the current batch, allocate a new one and re-verify
                        // whether the records fit in it.
                        // If flushing fails, we don't catch the exception in order to let
                        // the caller fail the current operation.
                        flushCurrentBatch();
                        maybeAllocateNewBatch(
                            producerId,
                            producerEpoch,
                            verificationGuard,
                            currentTimeMs
                        );
                    }
                }


                for (int i = 0; i < records.size(); i++) {
                    U recordToReplay = records.get(i);

Review comment (bug):
Let's update the comments.

eg.
```
                    // Fail immediately when the records are too large to fit in a single batch,
                    // even when assuming an optimistic compression ratio.
```

Code after:
                // Prepare the records.
                List<SimpleRecord> recordsToAppend = new ArrayList<>(records.size());
                for (U record : records) {
                    recordsToAppend.add(new SimpleRecord(
                        currentTimeMs,
                        serializer.serializeKey(record),
                        serializer.serializeValue(record)
                    ));
                }

                if (isAtomic) {
                    // Compute the size of the records.
                    int estimatedSizeUpperBound = AbstractRecords.estimateSizeInBytes(
                        currentBatch.builder.magic(),
                        CompressionType.NONE,
                        recordsToAppend
                    );

                    if (!currentBatch.builder.hasRoomFor(estimatedSizeUpperBound)) {
                        // Start a new batch when the total uncompressed data size would exceed
                        // the max batch size. We still allow atomic writes with an uncompressed size
                        // larger than the max batch size as long as they compress down to under the max
                        // batch size. These large writes go into a batch by themselves.
                        flushCurrentBatch();
                        maybeAllocateNewBatch(
                            producerId,
                            producerEpoch,
                            verificationGuard,
                            currentTimeMs
                        );
                    }
                }


                for (int i = 0; i < records.size(); i++) {
                    U recordToReplay = records.get(i);
                    SimpleRecord recordToAppend = recordsToAppend.get(i);

                    if (!isAtomic) {
                        // Check if the current batch has enough space. We check this before
                        // replaying the record in order to avoid having to revert back
                        // changes if the record do not fit within a batch.
                        boolean hasRoomFor = currentBatch.builder.hasRoomFor(
                            recordToAppend.timestamp(),
                            recordToAppend.key(),
                            recordToAppend.value(),
                            recordToAppend.headers()
                        );

                        if (!hasRoomFor) {
                            // If flushing fails, we don't catch the exception in order to let