[kafka] Handle basic Produce requests independently of storage

fluss-kafka/pom.xml

@@ -64,6 +64,13 @@
         </dependency>
 
         <!-- test dependency -->
+        <dependency>
+            <groupId>org.apache.curator</groupId>
+            <artifactId>curator-test</artifactId>
+            <version>${curator.version}</version>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.fluss</groupId>
             <artifactId>fluss-test-utils</artifactId>

fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaChannelInitializer.java

@@ -31,17 +31,20 @@
 public class KafkaChannelInitializer extends NettyChannelInitializer {
 
     private final RequestChannel[] requestChannels;
+    private final String listenerName;
     private final int maxRequestSize;
     private final LengthFieldPrepender prepender = new LengthFieldPrepender(4);
     private final boolean preferHeap;
 
     public KafkaChannelInitializer(
             RequestChannel[] requestChannels,
+            String listenerName,
             long maxIdleTimeSeconds,
             int maxRequestSize,
             boolean preferHeap) {
         super(maxIdleTimeSeconds);
         this.requestChannels = requestChannels;
+        this.listenerName = listenerName;
         this.maxRequestSize = maxRequestSize;
         this.preferHeap = preferHeap;
     }
@@ -53,6 +56,6 @@ protected void initChannel(SocketChannel ch) throws Exception {
         ch.pipeline().addLast(prepender);
         addFrameDecoder(ch, maxRequestSize, 4, preferHeap);
         ch.pipeline().addLast("flowController", new FlowControlHandler());
-        ch.pipeline().addLast(new KafkaCommandDecoder(requestChannels));
+        ch.pipeline().addLast(new KafkaCommandDecoder(requestChannels, listenerName));
     }
 }

fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaCommandDecoder.java

@@ -27,6 +27,7 @@
 import org.apache.fluss.utils.MathUtils;
 
 import org.apache.kafka.common.errors.LeaderNotAvailableException;
+import org.apache.kafka.common.message.ApiVersionsRequestData;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.requests.AbstractRequest;
 import org.apache.kafka.common.requests.AbstractResponse;
@@ -55,6 +56,7 @@ public class KafkaCommandDecoder extends SimpleChannelInboundHandler<ByteBuf> {
 
     private final RequestChannel[] requestChannels;
     private final int numChannels;
+    private final String listenerName;
 
     // Need to use a Queue to store the inflight responses, because Kafka clients require the
     // responses to be sent in order.
@@ -65,18 +67,18 @@ public class KafkaCommandDecoder extends SimpleChannelInboundHandler<ByteBuf> {
     protected volatile ChannelHandlerContext ctx;
     protected SocketAddress remoteAddress;
 
-    public KafkaCommandDecoder(RequestChannel[] requestChannels) {
+    public KafkaCommandDecoder(RequestChannel[] requestChannels, String listenerName) {
         super(false);
         this.requestChannels = requestChannels;
         this.numChannels = requestChannels.length;
+        this.listenerName = listenerName;
     }
 
     @Override
     public void channelRead0(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception {
         CompletableFuture<AbstractResponse> future = new CompletableFuture<>();
-        boolean needRelease = false;
         try {
-            KafkaRequest request = parseRequest(ctx, future, buffer);
+            KafkaRequest request = parseRequest(ctx, future, buffer, listenerName);
             inflightResponses.addLast(request);
             future.whenCompleteAsync((r, t) -> sendResponse(ctx), ctx.executor());
             int channelIndex =
@@ -86,16 +88,15 @@ public void channelRead0(ChannelHandlerContext ctx, ByteBuf buffer) throws Excep
             if (!isActive.get()) {
                 LOG.warn("Received a request on an inactive channel: {}", remoteAddress);
                 request.fail(new LeaderNotAvailableException("Channel is inactive"));
-                needRelease = true;
             }
         } catch (Throwable t) {
-            needRelease = true;
             LOG.error("Error handling request", t);
             future.completeExceptionally(t);
         } finally {
-            if (needRelease) {
-                ReferenceCountUtil.release(buffer);
-            }
+            // KafkaRequest retains the buffer to transfer ownership to request processing. Release
+            // the decoder's ownership on every path. KafkaRequest.releaseBuffer() is idempotent
+            // because worker cleanup and response completion can both release that ownership.
+            ReferenceCountUtil.release(buffer);
         }
     }
 
@@ -184,19 +185,39 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E
     }
 
     private static KafkaRequest parseRequest(
-            ChannelHandlerContext ctx, CompletableFuture<AbstractResponse> future, ByteBuf buffer) {
+            ChannelHandlerContext ctx,
+            CompletableFuture<AbstractResponse> future,
+            ByteBuf buffer,
+            String listenerName) {
         ByteBuffer nioBuffer = buffer.nioBuffer();
         RequestHeader header = RequestHeader.parse(nioBuffer);
         if (isUnsupportedApiVersionRequest(header)) {
             ApiVersionsRequest request =
-                    new ApiVersionsRequest.Builder(header.apiVersion()).build();
+                    new ApiVersionsRequest(
+                            new ApiVersionsRequestData(),
+                            API_VERSIONS.oldestVersion(),
+                            header.apiVersion());
             return new KafkaRequest(
-                    API_VERSIONS, header.apiVersion(), header, request, buffer, ctx, future);
+                    API_VERSIONS,
+                    header.apiVersion(),
+                    header,
+                    request,
+                    listenerName,
+                    buffer,
+                    ctx,
+                    future);
         }
         RequestAndSize request =
                 AbstractRequest.parseRequest(header.apiKey(), header.apiVersion(), nioBuffer);
         return new KafkaRequest(
-                header.apiKey(), header.apiVersion(), header, request.request, buffer, ctx, future);
+                header.apiKey(),
+                header.apiVersion(),
+                header,
+                request.request,
+                listenerName,
+                buffer,
+                ctx,
+                future);
     }
 
     private static boolean isUnsupportedApiVersionRequest(RequestHeader header) {

fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaProtocolPlugin.java

@@ -53,6 +53,7 @@ public ChannelHandler createChannelHandler(
             RequestChannel[] requestChannels, String listenerName) {
         return new KafkaChannelInitializer(
                 requestChannels,
+                listenerName,
                 conf.get(ConfigOptions.KAFKA_CONNECTION_MAX_IDLE_TIME).getSeconds(),
                 (int) conf.get(ConfigOptions.NETTY_SERVER_MAX_REQUEST_SIZE).getBytes(),
                 conf.getBoolean(ConfigOptions.NETTY_CLIENT_ALLOCATOR_HEAP_BUFFER_FIRST));
@@ -66,6 +67,6 @@ public RequestHandler<?> createRequestHandler(RpcGatewayService service) {
                             + service.getClass().getSimpleName());
         }
         TabletServerGateway gateway = (TabletServerGateway) service;
-        return new KafkaRequestHandler(gateway);
+        return new KafkaRequestHandler(service, gateway);
     }
 }

fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaRequest.java

@@ -35,6 +35,7 @@
 
 import java.nio.ByteBuffer;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
 /** Represents a request received from Kafka protocol channel. */
@@ -46,10 +47,12 @@ public class KafkaRequest implements RpcRequest {
     private final long requestId = ID_GENERATOR.getAndIncrement();
     private final RequestHeader header;
     private final AbstractRequest request;
+    private final String listenerName;
     private final ByteBuf buffer;
     private final ChannelHandlerContext ctx;
     private final long startTimeMs;
     private final CompletableFuture<AbstractResponse> future;
+    private final AtomicBoolean bufferReleased = new AtomicBoolean();
     private volatile boolean cancelled = false;
 
     protected KafkaRequest(
@@ -60,10 +63,23 @@ protected KafkaRequest(
             ByteBuf buffer,
             ChannelHandlerContext ctx,
             CompletableFuture<AbstractResponse> future) {
+        this(apiKey, apiVersion, header, request, "UNKNOWN", buffer, ctx, future);
+    }
+
+    protected KafkaRequest(
+            ApiKeys apiKey,
+            short apiVersion,
+            RequestHeader header,
+            AbstractRequest request,
+            String listenerName,
+            ByteBuf buffer,
+            ChannelHandlerContext ctx,
+            CompletableFuture<AbstractResponse> future) {
         this.apiKey = apiKey;
         this.apiVersion = apiVersion;
         this.header = header;
         this.request = request;
+        this.listenerName = listenerName;
         this.buffer = buffer.retain();
         this.ctx = ctx;
         this.startTimeMs = System.currentTimeMillis();
@@ -77,7 +93,9 @@ public RequestType getRequestType() {
 
     @Override
     public void releaseBuffer() {
-        ReferenceCountUtil.safeRelease(buffer);
+        if (bufferReleased.compareAndSet(false, true)) {
+            ReferenceCountUtil.safeRelease(buffer);
+        }
     }
 
     public ApiKeys apiKey() {
@@ -100,6 +118,10 @@ public <T> T request() {
         return (T) request;
     }
 
+    public String listenerName() {
+        return listenerName;
+    }
+
     public ChannelHandlerContext ctx() {
         return ctx;
     }
@@ -149,12 +171,17 @@ private ByteBuf serialize(AbstractResponse response) {
         int headerSize = headerData.size(cache, headerVersion);
         ApiMessage apiMessage = response.data();
         int messageSize = apiMessage.size(cache, apiVersion);
-        final ByteBuf buffer = ctx.alloc().buffer(headerSize + messageSize);
-        buffer.writerIndex(headerSize + messageSize);
-        final ByteBuffer nioBuffer = buffer.nioBuffer();
-        final ByteBufferAccessor writable = new ByteBufferAccessor(nioBuffer);
-        headerData.write(writable, cache, headerVersion);
-        apiMessage.write(writable, cache, apiVersion);
-        return buffer;
+        final ByteBuf responseBuffer = ctx.alloc().buffer(headerSize + messageSize);
+        try {
+            responseBuffer.writerIndex(headerSize + messageSize);
+            final ByteBuffer nioBuffer = responseBuffer.nioBuffer();
+            final ByteBufferAccessor writable = new ByteBufferAccessor(nioBuffer);
+            headerData.write(writable, cache, headerVersion);
+            apiMessage.write(writable, cache, apiVersion);
+            return responseBuffer;
+        } catch (Throwable t) {
+            ReferenceCountUtil.safeRelease(responseBuffer);
+            throw t;
+        }
     }
 }

fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaRequestContext.java

@@ -0,0 +1,96 @@
+/*
+ * 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.fluss.kafka;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.shaded.netty4.io.netty.channel.Channel;
+
+import org.apache.kafka.common.protocol.ApiKeys;
+
+import java.net.SocketAddress;
+
+/** Immutable wire-level context made available to Kafka API handlers. */
+@Internal
+public final class KafkaRequestContext {
+
+    private final int correlationId;
+    private final String clientId;
+    private final ApiKeys apiKey;
+    private final short apiVersion;
+    private final String listenerName;
+    private final SocketAddress localAddress;
+    private final SocketAddress remoteAddress;
+    private final long receivedTimeMs;
+
+    private KafkaRequestContext(KafkaRequest request) {
+        this.correlationId = request.header().correlationId();
+        this.clientId = request.header().clientId();
+        this.apiKey = request.apiKey();
+        this.apiVersion = request.apiVersion();
+        this.listenerName = request.listenerName();
+        Channel channel = request.ctx().channel();
+        this.localAddress = channel == null ? null : channel.localAddress();
+        this.remoteAddress = channel == null ? null : channel.remoteAddress();
+        this.receivedTimeMs = request.startTimeMs();
+    }
+
+    /** Creates a context from a network request. */
+    public static KafkaRequestContext fromRequest(KafkaRequest request) {
+        return new KafkaRequestContext(request);
+    }
+
+    /** Returns the request correlation ID. */
+    public int correlationId() {
+        return correlationId;
+    }
+
+    /** Returns the client ID, or {@code null} when the request did not provide one. */
+    public String clientId() {
+        return clientId;
+    }
+
+    /** Returns the Kafka API key. */
+    public ApiKeys apiKey() {
+        return apiKey;
+    }
+
+    /** Returns the Kafka request version. */
+    public short apiVersion() {
+        return apiVersion;
+    }
+
+    /** Returns the listener that accepted the request. */
+    public String listenerName() {
+        return listenerName;
+    }
+
+    /** Returns the local socket address. */
+    public SocketAddress localAddress() {
+        return localAddress;
+    }
+
+    /** Returns the remote socket address. */
+    public SocketAddress remoteAddress() {
+        return remoteAddress;
+    }
+
+    /** Returns the wall-clock time at which the request was received. */
+    public long receivedTimeMs() {
+        return receivedTimeMs;
+    }
+}