dnadolny commented on code in PR #21065:
URL: https://github.com/apache/kafka/pull/21065#discussion_r2666490172
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -3591,6 +3596,44 @@ public void
testWhenProduceResponseReturnsWithALeaderShipChangeErrorAndNewLeader
}
}
+ @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");
+ }
Review Comment:
Ah for some reason I thought I had to make a message >= pooled size to get
it to use the pool, but I retested and see that you're right and I don't need
that. I've just changed it to `"value"`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
