[kafka] Cover Produce lifecycle and replication regressions

fluss-kafka/pom.xml

@@ -64,6 +64,19 @@
         </dependency>
 
         <!-- test dependency -->
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>fluss-client</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </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

@@ -19,6 +19,7 @@
 
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
+import org.apache.fluss.kafka.backend.produce.KafkaProduceConversionExecutor;
 import org.apache.fluss.rpc.RpcGatewayService;
 import org.apache.fluss.rpc.gateway.TabletServerGateway;
 import org.apache.fluss.rpc.netty.server.RequestChannel;
@@ -27,11 +28,16 @@
 import org.apache.fluss.shaded.netty4.io.netty.channel.ChannelHandler;
 
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+import static org.apache.fluss.utils.Preconditions.checkState;
 
 /** The Kafka protocol plugin. */
 public class KafkaProtocolPlugin implements NetworkProtocolPlugin {
 
     private Configuration conf;
+    private KafkaProduceConversionExecutor conversionExecutor;
+    private CompletableFuture<Void> closeFuture;
 
     @Override
     public String name() {
@@ -40,7 +46,7 @@ public String name() {
 
     @Override
     public void setup(Configuration conf) {
-        this.conf = conf;
+        this.conf = new Configuration(conf);
     }
 
     @Override
@@ -53,19 +59,48 @@ 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));
     }
 
     @Override
-    public RequestHandler<?> createRequestHandler(RpcGatewayService service) {
+    public synchronized RequestHandler<?> createRequestHandler(RpcGatewayService service) {
         if (!(service instanceof TabletServerGateway)) {
             throw new IllegalArgumentException(
                     "Kafka protocol endpoints can only be enabled on TabletServers, but the service is "
                             + service.getClass().getSimpleName());
         }
+        checkState(closeFuture == null, "Kafka protocol plugin has already been closed.");
+        if (conversionExecutor == null) {
+            // Mirror the RPC worker concurrency, with a separate queue capped at 1024 requests.
+            int queueCapacity =
+                    Math.max(
+                            1,
+                            Math.min(
+                                    1024,
+                                    conf.get(ConfigOptions.NETTY_SERVER_MAX_QUEUED_REQUESTS)));
+            int threads =
+                    Math.max(
+                            1,
+                            Math.min(
+                                    queueCapacity,
+                                    conf.get(ConfigOptions.NETTY_SERVER_NUM_WORKER_THREADS)));
+            conversionExecutor = new KafkaProduceConversionExecutor(threads, queueCapacity);
+        }
         TabletServerGateway gateway = (TabletServerGateway) service;
-        return new KafkaRequestHandler(gateway);
+        return new KafkaRequestHandler(service, gateway, conversionExecutor);
+    }
+
+    @Override
+    public synchronized CompletableFuture<Void> closeAsync() {
+        if (closeFuture == null) {
+            closeFuture =
+                    conversionExecutor == null
+                            ? CompletableFuture.completedFuture(null)
+                            : conversionExecutor.closeAsync();
+        }
+        return closeFuture;
     }
 }

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;
+        }
     }
 }