lianetm commented on code in PR #22654:
URL: https://github.com/apache/kafka/pull/22654#discussion_r3590444883
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:
##########
@@ -1251,23 +1266,69 @@ public PartitionerConfig() {
}
/*
- * Metadata about a record just appended to the record accumulator
+ * Result of an attempt to append a record to the accumulator. Exactly one
of three states:
+ * the record was appended ({@link RecordAppendResult#appended()}, {@code
future} is set), the
+ * open batch needs more chunk capacity first ({@code
needsBufferExtension}), or a new batch
+ * must be created for the record ({@code needsNewBatch}).
Review Comment:
very nice idea indeed, but we can't use sealed (or record) on the clients
module yet (still on java 11 :S).
Still the intention makes sense, I did update to improve on the same
direction, using what we can: enum to make the cases mutually exclusive,
explicit creation of the different results with named funcs to limit it
(private constructor).
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedBufferPool.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.producer.internals;
+
+import org.apache.kafka.clients.producer.BufferExhaustedException;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.utils.Time;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+
+/**
+ * A {@link BufferPool} dedicated to chunk-sized buffer reuse (chunk size =
{@link #poolableSize()}).
+ * <p>
+ * Adds {@link #allocateChunks(int, long)} to acquire multiple chunks
atomically.
+ */
+public class ChunkedBufferPool extends BufferPool {
+
+ public ChunkedBufferPool(long memory, int chunkSize, Metrics metrics, Time
time, String metricGrpName) {
+ super(memory, chunkSize, metrics, time, metricGrpName);
+ }
+
+ /**
+ * Allocate {@code ceil(totalSize / chunkSize)} chunk-sized buffers
atomically, mirroring
+ * {@link BufferPool#allocate}: satisfied immediately if memory is
available, else blocks up to
+ * {@code maxTimeToBlockMs} for the whole request (FIFO on {@link
#waiters}).
+ * The reservation is tracked as bytes against {@link
#nonPooledAvailableMemory} plus chunks polled
+ * from {@link #free}. Any failure refunds the whole reservation and
signals the next waiter
+ * before the exception propagates, so a failed request leaves nothing
reserved.
+ *
+ * @param totalSize minimum total bytes of capacity required across
the returned chunks
+ * @param maxTimeToBlockMs maximum time in milliseconds to block waiting
for memory
+ * @return list of {@code ceil(totalSize / chunkSize)} {@code
ByteBuffer}s, each of capacity
+ * {@code chunkSize}
+ * @throws InterruptedException if interrupted while waiting
+ * @throws IllegalArgumentException if {@code totalSize <= 0}, or if the
request rounded up to
+ * whole chunks exceeds {@code totalMemory()}
+ * @throws BufferExhaustedException if the request can't be satisfied
within {@code maxTimeToBlockMs}
+ * @throws KafkaException if the pool is closed during the wait
+ */
+ public List<ByteBuffer> allocateChunks(int totalSize, long
maxTimeToBlockMs) throws InterruptedException {
+ if (totalSize <= 0)
+ throw new IllegalArgumentException("totalSize must be positive: "
+ totalSize);
+ throwIfChunksNeededExceedsPool(totalSize);
+
+ int chunkSize = poolableSize();
+ int numChunks = (int) (((long) totalSize + chunkSize - 1L) /
chunkSize);
Review Comment:
yes, done
--
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]
