Re: [PR] KAFKA-20807: Refactor ByteBufferOutputStream for extensibility (for KIP-1332) [kafka]

["junrao (via GitHub)" <gi...@apache.org>]

junrao commented on code in PR #22843:
URL: https://github.com/apache/kafka/pull/22843#discussion_r3723033749


##########
clients/src/main/java/org/apache/kafka/common/utils/internals/ByteBufferOutputStream.java:
##########
@@ -20,90 +20,34 @@
 import java.nio.ByteBuffer;
 
 /**
- * A ByteBuffer-backed OutputStream that expands the internal ByteBuffer as required. Given this, the caller should
- * always access the underlying ByteBuffer via the {@link #buffer()} method until all writes are completed.
- *
- * This class is typically used for 2 purposes:
- *
- * 1. Write to a ByteBuffer when there is a chance that we may need to expand it in order to fit all the desired data
- * 2. Write to a ByteBuffer via methods that expect an OutputStream interface
- *
- * Hard to track bugs can happen when this class is used for the second reason and unexpected buffer expansion happens.
- * So, it's best to assume that buffer expansion can always happen. An improvement would be to create a separate class
- * that throws an error if buffer expansion is required to avoid the issue altogether.
+ * An {@link OutputStream} backed by one or more {@link ByteBuffer}s, exposing the written bytes via
+ * {@link #buffer()}. The default single-buffer implementation is {@link SingleByteBufferOutputStream}.
  */
-public class ByteBufferOutputStream extends OutputStream {
-
-    private static final float REALLOCATION_FACTOR = 1.1f;
+public abstract class ByteBufferOutputStream extends OutputStream {
 
-    private final int initialCapacity;
-    private final int initialPosition;
-    private ByteBuffer buffer;
+    @Override
+    public abstract void write(int b);
 
-    /**
-     * Creates an instance of this class that will write to the received `buffer` up to its `limit`. If necessary to
-     * satisfy `write` or `position` calls, larger buffers will be allocated so the {@link #buffer()} method may return
-     * a different buffer than the received `buffer` parameter.
-     *
-     * Prefer one of the constructors that allocate the internal buffer for clearer semantics.
-     */
-    public ByteBufferOutputStream(ByteBuffer buffer) {
-        this.buffer = buffer;
-        this.initialPosition = buffer.position();
-        this.initialCapacity = buffer.capacity();
-    }
+    @Override
+    public abstract void write(byte[] bytes, int off, int len);
 
-    public ByteBufferOutputStream(int initialCapacity) {
-        this(initialCapacity, false);
-    }
+    public abstract void write(ByteBuffer sourceBuffer);
 
-    public ByteBufferOutputStream(int initialCapacity, boolean directBuffer) {
-        this(directBuffer ? ByteBuffer.allocateDirect(initialCapacity) : ByteBuffer.allocate(initialCapacity));
-    }
+    public abstract ByteBuffer buffer();
 
-    public void write(int b) {
-        ensureRemaining(1);
-        buffer.put((byte) b);
-    }
+    public abstract int position();
 
-    public void write(byte[] bytes, int off, int len) {
-        ensureRemaining(len);
-        buffer.put(bytes, off, len);
-    }
+    public abstract void position(int position);
 
-    public void write(ByteBuffer sourceBuffer) {
-        ensureRemaining(sourceBuffer.remaining());
-        buffer.put(sourceBuffer);
-    }
+    public abstract int remaining();
 
-    public ByteBuffer buffer() {
-        return buffer;
-    }
-
-    public int position() {
-        return buffer.position();
-    }
-
-    public int remaining() {
-        return buffer.remaining();
-    }
-
-    public int limit() {
-        return buffer.limit();
-    }
-
-    public void position(int position) {
-        ensureRemaining(position - buffer.position());
-        buffer.position(position);
-    }
+    public abstract int limit();

Review Comment:
   Could we move the simple ones like `limit` and `ensureRemaining` in this PR?



-- 
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: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org