[kafka] Establish compatibility framework

fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java

@@ -2645,6 +2645,22 @@ public class ConfigOptions {
                     .withDescription(
                             "The database for fluss kafka. The default database is `kafka`.");
 
+    public static final ConfigOption<String> KAFKA_DEFAULT_KEY_FORMAT =
+            key("kafka.default.key.format")
+                    .stringType()
+                    .defaultValue("raw")
+                    .withDescription(
+                            "The default format for Kafka record keys when a CreateTopics request does not specify fluss.key.format. "
+                                    + "Supported formats are raw and string.");
+
+    public static final ConfigOption<String> KAFKA_DEFAULT_VALUE_FORMAT =
+            key("kafka.default.value.format")
+                    .stringType()
+                    .defaultValue("raw")
+                    .withDescription(
+                            "The default format for Kafka record values when a CreateTopics request does not specify fluss.value.format. "
+                                    + "Supported formats are raw and string.");
+
     public static final ConfigOption<Duration> KAFKA_CONNECTION_MAX_IDLE_TIME =
             key("kafka.connection.max-idle-time")
                     .durationType()

fluss-kafka/pom.xml

@@ -64,11 +64,25 @@
         </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>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.fluss</groupId>
+            <artifactId>fluss-client</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.fluss</groupId>
             <artifactId>fluss-common</artifactId>
@@ -92,4 +106,4 @@
             <scope>test</scope>
         </dependency>
     </dependencies>
-</project>
+</project>

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 because Kafka record sets can reference its memory
+            // asynchronously. Release the decoder's ownership on every path; the request releases
+            // its retained reference after response handling or cancellation.
+            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,7 +19,9 @@
 
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
+import org.apache.fluss.kafka.format.KafkaDataFormat;
 import org.apache.fluss.rpc.RpcGatewayService;
+import org.apache.fluss.rpc.gateway.AdminGatewayProvider;
 import org.apache.fluss.rpc.gateway.TabletServerGateway;
 import org.apache.fluss.rpc.netty.server.RequestChannel;
 import org.apache.fluss.rpc.netty.server.RequestHandler;
@@ -53,6 +55,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 +69,15 @@ public RequestHandler<?> createRequestHandler(RpcGatewayService service) {
                             + service.getClass().getSimpleName());
         }
         TabletServerGateway gateway = (TabletServerGateway) service;
-        return new KafkaRequestHandler(gateway);
+        if (service instanceof AdminGatewayProvider) {
+            return new KafkaRequestHandler(
+                    service,
+                    gateway,
+                    ((AdminGatewayProvider) service).getAdminGateway(),
+                    conf.get(ConfigOptions.KAFKA_DATABASE),
+                    KafkaDataFormat.parse(conf.get(ConfigOptions.KAFKA_DEFAULT_KEY_FORMAT)),
+                    KafkaDataFormat.parse(conf.get(ConfigOptions.KAFKA_DEFAULT_VALUE_FORMAT)));
+        }
+        return new KafkaRequestHandler(service, gateway, conf.get(ConfigOptions.KAFKA_DATABASE));
     }
 }

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

@@ -46,6 +46,7 @@ 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;
@@ -60,10 +61,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();
@@ -100,6 +114,10 @@ public <T> T request() {
         return (T) request;
     }
 
+    public String listenerName() {
+        return listenerName;
+    }
+
     public ChannelHandlerContext ctx() {
         return ctx;
     }

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