Repository: apache/kafka
File: clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java (Java)

Code before:
            // TEST that a subsequent retry skips the backoff as a new leader information is available.
            sender.runOnce(); //send produce request
            assertEquals(1, sender.inFlightBatches(tp0).size());
            assertEquals(1, sender.inFlightBatches(tp1).size());
            assertTrue(client.hasInFlightRequests());
        } finally {
            m.close();
        }
    }

    @Test
    public void testNoBufferReuseWhenBatchHasTimeoutErrors() throws Exception {
        long totalSize = 1024 * 1024;
        try (Metrics m = new Metrics()) {
            BufferPool pool = new BufferPool(totalSize, batchSize, m, time, "producer-internal-metrics");

            // Allocate and store a poolable buffer, then return it to the pool so the Sender can pick it up
            ByteBuffer buffer = pool.allocate(batchSize, 0);
            pool.deallocate(buffer);

            setupWithTransactionState(null, false, pool);
            StringBuilder sb = new StringBuilder();
            // Make a value the right size so that RecordAccumulator calls BufferPool.allocate with the batchSize so it's pooled
            for (int i = 0; i < 16290; i++) {
                sb.append("a");
            }
            String largeValue = sb.toString();
            appendToAccumulator(tp0, 0L, "key", largeValue);
            sender.runOnce();  // connect
            sender.runOnce();  // send produce request

            assertEquals(1, client.inFlightRequestCount());
            assertEquals(1, sender.inFlightBatches(tp0).size());

            ProducerBatch batch = sender.inFlightBatches(tp0).get(0);
            // Validate the backing array of the buffer is the same as the pooled one from the start
            assertSame(buffer.array(), batch.records().buffer().array(), "Sender should have allocated the same buffer we created");

            time.sleep(DELIVERY_TIMEOUT_MS + 100);
            sender.runOnce();

            ByteBuffer newBuffer = pool.allocate(batchSize, 0);

            // TEST buffer should not be reused
            assertNotSame(buffer.array(), newBuffer.array(), "Buffer should not be reused");
        }
    }


    private void verifyErrorMessage(ProduceResponse response, String expectedMessage) throws Exception {
        Future<RecordMetadata> future = appendToAccumulator(tp0, 0L, "key", "value");

Review comment (performance):
The above could just be 

`String largeValue = "a".repeat(16290);`

Also, why is it necessary to create a large value? RecordAccumulator picks batchSize for allocation if the data size is smaller.

```
                    int size = Math.max(this.batchSize, AbstractRecords.estimateSizeInBytesUpperBound(
                            RecordBatch.CURRENT_MAGIC_VALUE, compression.type(), key, value, headers));
                    log.trace("Allocating a new {} byte message buffer for topic {} partition {} with remaining timeout {}ms", size, topic, effectivePartition, maxTimeToBlock);
                    // This call may block if we exhausted buffer space.
                    buffer = free.allocate(size, maxTimeToBlock);
```

Code after:
            assertEquals(tp0LeaderEpoch + 1, tp0NewLeaderInfo.epoch.get());
            Metadata.LeaderAndEpoch tp1NewLeaderInfo = metadata.currentLeader(tp1);
            assertEquals(newNodeForTp1, tp1NewLeaderInfo.leader.get());
            assertEquals(tp1LeaderEpoch + 1, tp1NewLeaderInfo.epoch.get());

            // Validate metadata-refresh is requested as NOT_LEADER_OR_FOLLOWER received earlier
            assertTrue(metadata.updateRequested());

            // TEST that a subsequent retry skips the backoff as a new leader information is available.
            sender.runOnce(); //send produce request
            assertEquals(1, sender.inFlightBatches(tp0).size());
            assertEquals(1, sender.inFlightBatches(tp1).size());
            assertTrue(client.hasInFlightRequests());
        } finally {
            m.close();
        }
    }

    @Test
    public void testNoBufferReuseWhenBatchExpires() throws Exception {
        long totalSize = 1024 * 1024;
        try (Metrics m = new Metrics()) {
            BufferPool pool = new BufferPool(totalSize, batchSize, m, time, "producer-internal-metrics");

            // Allocate and store a poolable buffer, then return it to the pool so the Sender can pick it up
            ByteBuffer buffer = pool.allocate(batchSize, 0);
            pool.deallocate(buffer);

            setupWithTransactionState(null, false, pool);
            appendToAccumulator(tp0, 0L, "key", "value");
            sender.runOnce();  // connect
            sender.runOnce();  // send produce request

            assertEquals(1, client.inFlightRequestCount());
            assertEquals(1, sender.inFlightBatches(tp0).size());

            ProducerBatch batch = sender.inFlightBatches(tp0).get(0);
            // Validate the backing array of the buffer is the same as the pooled one from the start
            assertSame(buffer.array(), batch.records().buffer().array(), "Sender should have allocated the same buffer we created");

            time.sleep(DELIVERY_TIMEOUT_MS + 100);
            sender.runOnce();

            ByteBuffer newBuffer = pool.allocate(batchSize, 0);

            // TEST buffer should not be reused
            assertNotSame(buffer.array(), newBuffer.array(), "Buffer should not be reused");
        }
    }