KAFKA-20866: Remove incremental-only append result state

clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedRecordAccumulator.java

@@ -158,7 +158,7 @@ public RecordAppendResult append(String topic,
                 }
 
                 if (appendResult.needsBufferExtension()) {
-                    extensionChunks = allocateExtensionChunks(appendResult.extensionBytesNeeded, dq, topic, effectivePartition);
+                    extensionChunks = allocateExtensionChunks(appendResult.extensionBytesNeeded(), dq, topic, effectivePartition);
                     if (extensionChunks == null) {
                         // Pool exhausted, so no writable batch is left to extend: retry, normally
                         // landing on the blocking new-batch path (needsNewBatch).

clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java

@@ -1290,46 +1290,27 @@ public PartitionerConfig() {
     }
 
     /*
-     * Result of an attempt to append a record to the accumulator. Carries exactly one of three
-     * mutually-exclusive outcomes: the record was appended ({@link RecordAppendResult#appended()}, {@code future} is set),
-     * the open batch needs more chunk capacity first ({@link RecordAppendResult#needsBufferExtension()}),
-     * or a new batch must be created for the record ({@link RecordAppendResult#needsNewBatch()}).
+     * Result of an attempt to append a record to the accumulator. A regular result represents
+     * either a successful append or a request to create a new batch. The incremental strategy
+     * uses a private subtype when an existing chunked batch needs more capacity.
      */
-    public static final class RecordAppendResult {
-        /**
-         * The three mutually-exclusive outcomes of an append attempt. Internal representation only;
-         * callers use {@link #appended()}, {@link #needsBufferExtension()}, and {@link #needsNewBatch()}.
-         */
-        private enum Outcome { APPENDED, NEEDS_BUFFER_EXTENSION, NEEDS_NEW_BATCH }
-
-        private final Outcome outcome;
+    public static class RecordAppendResult {
         public final FutureRecordMetadata future;
         public final boolean batchIsFull;
         public final boolean newBatchCreated;
-        /**
-         * Bytes of chunk capacity the open batch needs before the record fits (incremental
-         * strategy). Meaningful only for {@link Outcome#NEEDS_BUFFER_EXTENSION}: the append was NOT
-         * attempted ({@code future} is null); the caller allocates this many bytes, attaches them
-         * via {@link ChunkedProducerBatch#addBuffers}, and retries.
-         */
-        public final int extensionBytesNeeded;
         public final int appendedBytes;
 
-        /** The shared signal-only result for {@link Outcome#NEEDS_NEW_BATCH}; carries no per-append state. */
+        /** The shared signal-only result for a full or absent open batch. */
         public static final RecordAppendResult NEEDS_NEW_BATCH =
-                new RecordAppendResult(Outcome.NEEDS_NEW_BATCH, null, false, false, 0, 0);
+                new RecordAppendResult(null, false, false, 0);
 
-        private RecordAppendResult(Outcome outcome,
-                                   FutureRecordMetadata future,
+        private RecordAppendResult(FutureRecordMetadata future,
                                    boolean batchIsFull,
                                    boolean newBatchCreated,
-                                   int extensionBytesNeeded,
                                    int appendedBytes) {
-            this.outcome = outcome;
             this.future = future;
             this.batchIsFull = batchIsFull;
             this.newBatchCreated = newBatchCreated;
-            this.extensionBytesNeeded = extensionBytesNeeded;
             this.appendedBytes = appendedBytes;
         }
 
@@ -1339,7 +1320,7 @@ public static RecordAppendResult appended(FutureRecordMetadata future,
                                                   boolean newBatchCreated,
                                                   int appendedBytes) {
             Objects.requireNonNull(future, "future must be non-null for an appended result");
-            return new RecordAppendResult(Outcome.APPENDED, future, batchIsFull, newBatchCreated, 0, appendedBytes);
+            return new RecordAppendResult(future, batchIsFull, newBatchCreated, appendedBytes);
         }
 
         /**
@@ -1348,25 +1329,63 @@ public static RecordAppendResult appended(FutureRecordMetadata future,
          * bytes of chunk capacity and retry. The append was not attempted.
          */
         public static RecordAppendResult needsExtension(int extensionBytesNeeded) {
-            return new RecordAppendResult(Outcome.NEEDS_BUFFER_EXTENSION, null, false, false, extensionBytesNeeded, 0);
+            return new BufferExtensionResult(extensionBytesNeeded);
         }
 
         /**
          * @return {@code true} if the record was appended to the open batch ({@link #future} is then
          * non-null), {@code false} for the {@link #needsBufferExtension()} and {@link #needsNewBatch()} results.
          */
         public boolean appended() {
-            return outcome == Outcome.APPENDED;
+            return future != null;
         }
 
         /** @return {@code true} if the open batch needs more chunk capacity before the record fits. */
         public boolean needsBufferExtension() {
-            return outcome == Outcome.NEEDS_BUFFER_EXTENSION;
+            return false;
         }
 
         /** @return {@code true} if there is no open batch that can take the record, so a new one must be created. */
         public boolean needsNewBatch() {
-            return outcome == Outcome.NEEDS_NEW_BATCH;
+            return future == null;
+        }
+
+        /**
+         * Return the chunk capacity required by a buffer-extension result.
+         *
+         * @return the number of bytes to allocate
+         * @throws IllegalStateException if this is not a buffer-extension result
+         */
+        int extensionBytesNeeded() {
+            throw new IllegalStateException("Only a buffer-extension result has extension bytes");
+        }
+
+        /**
+         * The only result that carries incremental-strategy-specific state. Keeping this state
+         * out of the regular result avoids allocating space for it on every full-strategy append.
+         */
+        private static final class BufferExtensionResult extends RecordAppendResult {
+            private final int extensionBytesNeeded;
+
+            private BufferExtensionResult(int extensionBytesNeeded) {
+                super(null, false, false, 0);
+                this.extensionBytesNeeded = extensionBytesNeeded;
+            }
+
+            @Override
+            public boolean needsBufferExtension() {
+                return true;
+            }
+
+            @Override
+            public boolean needsNewBatch() {
+                return false;
+            }
+
+            @Override
+            int extensionBytesNeeded() {
+                return extensionBytesNeeded;
+            }
         }
     }
 

clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java

@@ -134,6 +134,16 @@ public void teardown() {
         this.metrics.close();
     }
 
+    @Test
+    public void testNeedsBufferExtensionResultExposesRequiredCapacity() {
+        RecordAccumulator.RecordAppendResult result = RecordAccumulator.RecordAppendResult.needsExtension(128);
+
+        assertFalse(result.appended());
+        assertTrue(result.needsBufferExtension());
+        assertFalse(result.needsNewBatch());
+        assertEquals(128, result.extensionBytesNeeded());
+    }
+
     @Test
     public void testDrainBatches() throws Exception {
         // test case: node1(tp1,tp2) , node2(tp3,tp4)