From 59e479f27fc935146161af164fe804ff6684ecee Mon Sep 17 00:00:00 2001 From: Yang Guo Date: Tue, 8 Sep 2026 11:29:44 +0800 Subject: [PATCH 1/6] [kafka] Add request dispatch and transport framework Introduce API registration, request context, version validation, and asynchronous error mapping. Fix request buffer ownership and response serialization cleanup while preserving the existing ApiVersions entry point. Validated with mvn -o -pl fluss-rpc,fluss-kafka verify. Co-Authored-By: Codex AI-Model: gpt-6 AI-Contributed/Feature: 555/555 AI-Contributed/UT: 525/525 --- .../fluss/kafka/KafkaChannelInitializer.java | 5 +- .../fluss/kafka/KafkaCommandDecoder.java | 45 +++-- .../fluss/kafka/KafkaProtocolPlugin.java | 1 + .../org/apache/fluss/kafka/KafkaRequest.java | 43 ++++- .../fluss/kafka/KafkaRequestContext.java | 96 ++++++++++ .../kafka/dispatcher/KafkaApiHandler.java | 37 ++++ .../kafka/dispatcher/KafkaApiRegistry.java | 80 +++++++++ .../fluss/kafka/dispatcher/KafkaApiSpec.java | 82 +++++++++ .../dispatcher/KafkaRequestDispatcher.java | 108 ++++++++++++ .../fluss/kafka/error/KafkaErrorMapper.java | 45 +++++ .../fluss/kafka/KafkaCommandDecoderTest.java | 118 +++++++++++++ .../fluss/kafka/KafkaRequestHandlerTest.java | 56 ++++-- .../apache/fluss/kafka/KafkaRequestTest.java | 58 ++++++ .../dispatcher/KafkaApiRegistryTest.java | 127 ++++++++++++++ .../KafkaRequestDispatcherTest.java | 166 ++++++++++++++++++ .../fluss/rpc/netty/server/NettyServer.java | 13 +- 16 files changed, 1039 insertions(+), 41 deletions(-) create mode 100644 fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaRequestContext.java create mode 100644 fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaApiHandler.java create mode 100644 fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaApiRegistry.java create mode 100644 fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaApiSpec.java create mode 100644 fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcher.java create mode 100644 fluss-kafka/src/main/java/org/apache/fluss/kafka/error/KafkaErrorMapper.java create mode 100644 fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaCommandDecoderTest.java create mode 100644 fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaRequestTest.java create mode 100644 fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaApiRegistryTest.java create mode 100644 fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcherTest.java diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaChannelInitializer.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaChannelInitializer.java index 5e7551a9af7..29bdc745ca9 100644 --- a/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaChannelInitializer.java +++ b/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)); } } diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaCommandDecoder.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaCommandDecoder.java index 43a0533b2d3..637a1aaa1fe 100644 --- a/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaCommandDecoder.java +++ b/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 { 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 { 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 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 future, ByteBuf buffer) { + ChannelHandlerContext ctx, + CompletableFuture 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) { diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaProtocolPlugin.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaProtocolPlugin.java index d92ba5e68fc..c966f745b8c 100644 --- a/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaProtocolPlugin.java +++ b/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)); diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaRequest.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaRequest.java index 25e409a7455..0d2799a7a18 100644 --- a/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaRequest.java +++ b/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 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 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 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 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; + } } } diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaRequestContext.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaRequestContext.java new file mode 100644 index 00000000000..e75a20babcc --- /dev/null +++ b/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; + } +} diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaApiHandler.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaApiHandler.java new file mode 100644 index 00000000000..36995b8e27f --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaApiHandler.java @@ -0,0 +1,37 @@ +/* + * 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.dispatcher; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.kafka.KafkaRequestContext; + +import org.apache.kafka.common.requests.AbstractRequest; +import org.apache.kafka.common.requests.AbstractResponse; + +import java.util.concurrent.CompletableFuture; + +/** Handles one Kafka API without blocking the request processor thread. */ +@Internal +public interface KafkaApiHandler { + + /** Returns the capability implemented by this handler. */ + KafkaApiSpec apiSpec(); + + /** Handles a parsed request asynchronously. */ + CompletableFuture handle(KafkaRequestContext context, R request); +} diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaApiRegistry.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaApiRegistry.java new file mode 100644 index 00000000000..b4a1dfd8ea3 --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaApiRegistry.java @@ -0,0 +1,80 @@ +/* + * 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.dispatcher; + +import org.apache.fluss.annotation.Internal; + +import org.apache.kafka.common.protocol.ApiKeys; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.fluss.utils.Preconditions.checkArgument; +import static org.apache.fluss.utils.Preconditions.checkNotNull; +import static org.apache.fluss.utils.Preconditions.checkState; + +/** Registry and single source of truth for Kafka APIs exposed by one server. */ +@Internal +public final class KafkaApiRegistry { + + private final Map> handlers = new HashMap<>(); + private boolean frozen; + + /** Creates an empty API registry. */ + public KafkaApiRegistry() {} + + /** Registers a handler. Registrations are rejected after {@link #freeze()} is called. */ + public void register(KafkaApiHandler handler) { + checkNotNull(handler); + checkState(!frozen, "Kafka API registry is already frozen."); + ApiKeys apiKey = handler.apiSpec().apiKey(); + checkArgument(!handlers.containsKey(apiKey), "Kafka API %s is already registered.", apiKey); + handlers.put(apiKey, handler); + } + + /** Prevents further registrations. */ + public void freeze() { + frozen = true; + } + + /** Returns a routable handler, or {@code null} when the API is not exposed by this server. */ + public KafkaApiHandler lookup(ApiKeys apiKey) { + KafkaApiHandler handler = handlers.get(apiKey); + if (handler == null || !handler.apiSpec().advertised()) { + return null; + } + return handler; + } + + /** Returns the sorted API specifications advertised by this server. */ + public List advertisedApiSpecs() { + List specs = new ArrayList<>(); + for (KafkaApiHandler handler : handlers.values()) { + KafkaApiSpec spec = handler.apiSpec(); + if (spec.advertised()) { + specs.add(spec); + } + } + Collections.sort(specs, Comparator.comparingInt(spec -> spec.apiKey().id)); + return Collections.unmodifiableList(specs); + } +} diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaApiSpec.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaApiSpec.java new file mode 100644 index 00000000000..50d6a7ebab2 --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaApiSpec.java @@ -0,0 +1,82 @@ +/* + * 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.dispatcher; + +import org.apache.fluss.annotation.Internal; + +import org.apache.kafka.common.protocol.ApiKeys; + +import static org.apache.fluss.utils.Preconditions.checkArgument; +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** Describes the versions actually supported by a Kafka API handler. */ +@Internal +public final class KafkaApiSpec { + + private final ApiKeys apiKey; + private final short minVersion; + private final short maxVersion; + private final boolean advertised; + + /** Creates an API specification. */ + public KafkaApiSpec(ApiKeys apiKey, short minVersion, short maxVersion, boolean advertised) { + this.apiKey = checkNotNull(apiKey); + checkArgument(minVersion >= 0, "Minimum version must not be negative."); + checkArgument( + minVersion <= maxVersion, + "Minimum version %s must not exceed maximum version %s.", + minVersion, + maxVersion); + checkArgument( + minVersion >= apiKey.oldestVersion() && maxVersion <= apiKey.latestVersion(), + "Version range [%s, %s] is outside the Kafka library range [%s, %s] for %s.", + minVersion, + maxVersion, + apiKey.oldestVersion(), + apiKey.latestVersion(), + apiKey); + this.minVersion = minVersion; + this.maxVersion = maxVersion; + this.advertised = advertised; + } + + /** Returns the Kafka API key. */ + public ApiKeys apiKey() { + return apiKey; + } + + /** Returns the oldest supported request version. */ + public short minVersion() { + return minVersion; + } + + /** Returns the newest supported request version. */ + public short maxVersion() { + return maxVersion; + } + + /** Returns whether this API is allowed to be routed and advertised. */ + public boolean advertised() { + return advertised; + } + + /** Returns whether the supplied request version is supported. */ + public boolean supportsVersion(short version) { + return version >= minVersion && version <= maxVersion; + } +} diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcher.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcher.java new file mode 100644 index 00000000000..efdfe33dd66 --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcher.java @@ -0,0 +1,108 @@ +/* + * 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.dispatcher; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.kafka.KafkaRequest; +import org.apache.fluss.kafka.KafkaRequestContext; +import org.apache.fluss.kafka.error.KafkaErrorMapper; + +import org.apache.kafka.common.errors.UnsupportedVersionException; +import org.apache.kafka.common.requests.AbstractRequest; +import org.apache.kafka.common.requests.AbstractResponse; + +import java.util.concurrent.CompletableFuture; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** Validates and dispatches parsed Kafka requests to independently registered API handlers. */ +@Internal +public final class KafkaRequestDispatcher { + + private final KafkaApiRegistry registry; + private final KafkaErrorMapper errorMapper; + + /** Creates a dispatcher backed by the supplied registry and error mapper. */ + public KafkaRequestDispatcher(KafkaApiRegistry registry, KafkaErrorMapper errorMapper) { + this.registry = checkNotNull(registry); + this.errorMapper = checkNotNull(errorMapper); + } + + /** Dispatches a request and always completes with a Kafka protocol response. */ + public CompletableFuture dispatch(KafkaRequest request) { + AbstractRequest abstractRequest = request.request(); + KafkaApiHandler handler = registry.lookup(request.apiKey()); + if (handler == null) { + return completedErrorResponse( + abstractRequest, + new UnsupportedVersionException( + "Kafka API " + request.apiKey() + " is not supported by this server.")); + } + + KafkaApiSpec spec = handler.apiSpec(); + if (!spec.supportsVersion(request.apiVersion())) { + return completedErrorResponse( + abstractRequest, + new UnsupportedVersionException( + String.format( + "Version %s is not supported for %s. Supported versions are [%s, %s].", + request.apiVersion(), + request.apiKey(), + spec.minVersion(), + spec.maxVersion()))); + } + + CompletableFuture responseFuture; + try { + responseFuture = + invoke(handler, KafkaRequestContext.fromRequest(request), abstractRequest); + if (responseFuture == null) { + throw new NullPointerException("Kafka API handler returned a null future."); + } + } catch (Throwable t) { + return completedErrorResponse(abstractRequest, t); + } + + CompletableFuture result = new CompletableFuture<>(); + responseFuture.whenComplete( + (response, failure) -> { + if (failure == null && response != null) { + result.complete(response); + } else { + Throwable responseFailure = + failure == null + ? new NullPointerException( + "Kafka API handler returned a null response.") + : failure; + result.complete(errorMapper.toResponse(abstractRequest, responseFailure)); + } + }); + return result; + } + + @SuppressWarnings("unchecked") + private static CompletableFuture invoke( + KafkaApiHandler handler, KafkaRequestContext context, AbstractRequest request) { + return ((KafkaApiHandler) handler).handle(context, request); + } + + private CompletableFuture completedErrorResponse( + AbstractRequest request, Throwable failure) { + return CompletableFuture.completedFuture(errorMapper.toResponse(request, failure)); + } +} diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/error/KafkaErrorMapper.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/error/KafkaErrorMapper.java new file mode 100644 index 00000000000..4396566396d --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/error/KafkaErrorMapper.java @@ -0,0 +1,45 @@ +/* + * 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.error; + +import org.apache.fluss.annotation.Internal; + +import org.apache.kafka.common.requests.AbstractRequest; +import org.apache.kafka.common.requests.AbstractResponse; + +import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutionException; + +/** Maps failures from the compatibility layer to version-aware Kafka responses. */ +@Internal +public final class KafkaErrorMapper { + + /** Converts a failure to the error response defined by the parsed Kafka request. */ + public AbstractResponse toResponse(AbstractRequest request, Throwable failure) { + return request.getErrorResponse(unwrap(failure)); + } + + private static Throwable unwrap(Throwable failure) { + Throwable current = failure; + while ((current instanceof CompletionException || current instanceof ExecutionException) + && current.getCause() != null) { + current = current.getCause(); + } + return current; + } +} diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaCommandDecoderTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaCommandDecoderTest.java new file mode 100644 index 00000000000..a2b8a0dc60b --- /dev/null +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaCommandDecoderTest.java @@ -0,0 +1,118 @@ +/* + * 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.rpc.netty.server.RequestChannel; +import org.apache.fluss.shaded.netty4.io.netty.buffer.ByteBuf; +import org.apache.fluss.shaded.netty4.io.netty.buffer.Unpooled; +import org.apache.fluss.shaded.netty4.io.netty.channel.embedded.EmbeddedChannel; + +import org.apache.kafka.common.message.ApiVersionsRequestData; +import org.apache.kafka.common.message.ApiVersionsResponseData; +import org.apache.kafka.common.message.ProduceRequestData; +import org.apache.kafka.common.message.ProduceResponseData; +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.requests.AbstractRequest; +import org.apache.kafka.common.requests.ApiVersionsRequest; +import org.apache.kafka.common.requests.ApiVersionsResponse; +import org.apache.kafka.common.requests.ProduceRequest; +import org.apache.kafka.common.requests.ProduceResponse; +import org.apache.kafka.common.requests.RequestHeader; +import org.apache.kafka.common.requests.RequestUtils; +import org.apache.kafka.common.requests.ResponseHeader; +import org.junit.jupiter.api.Test; + +import java.nio.ByteBuffer; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests response ordering and ownership in {@link KafkaCommandDecoder}. */ +public class KafkaCommandDecoderTest { + + @Test + public void testAcksZeroSuppressesResponseAndUnblocksFollowingResponse() { + RequestChannel requestChannel = new RequestChannel(100); + EmbeddedChannel channel = + new EmbeddedChannel( + new KafkaCommandDecoder(new RequestChannel[] {requestChannel}, "KAFKA")); + short produceVersion = ApiKeys.PRODUCE.latestVersion(); + ProduceRequest produceRequest = + new ProduceRequest( + new ProduceRequestData().setAcks((short) 0).setTimeoutMs(1000), + produceVersion); + RequestHeader produceHeader = + new RequestHeader(ApiKeys.PRODUCE, produceVersion, "client", 1); + ByteBuf produceBuffer = serialize(produceHeader, produceRequest); + + short apiVersionsVersion = ApiKeys.API_VERSIONS.latestVersion(); + ApiVersionsRequest apiVersionsRequest = + new ApiVersionsRequest.Builder( + new ApiVersionsRequestData(), + apiVersionsVersion, + apiVersionsVersion) + .build(); + RequestHeader apiVersionsHeader = + new RequestHeader(ApiKeys.API_VERSIONS, apiVersionsVersion, "client", 2); + ByteBuf apiVersionsBuffer = serialize(apiVersionsHeader, apiVersionsRequest); + + try { + channel.writeInbound(produceBuffer); + channel.writeInbound(apiVersionsBuffer); + KafkaRequest first = (KafkaRequest) requestChannel.pollRequest(1000); + KafkaRequest second = (KafkaRequest) requestChannel.pollRequest(1000); + assertThat(first).isNotNull(); + assertThat(second).isNotNull(); + + second.complete(new ApiVersionsResponse(new ApiVersionsResponseData())); + channel.runPendingTasks(); + Object blockedResponse = channel.readOutbound(); + assertThat(blockedResponse).isNull(); + + first.complete(new ProduceResponse(new ProduceResponseData())); + channel.runPendingTasks(); + + ByteBuf response = channel.readOutbound(); + try { + assertThat(response).isNotNull(); + ResponseHeader responseHeader = + ResponseHeader.parse( + response.nioBuffer(), + apiVersionsHeader.toResponseHeader().headerVersion()); + assertThat(responseHeader.correlationId()).isEqualTo(2); + Object additionalResponse = channel.readOutbound(); + assertThat(additionalResponse).isNull(); + } finally { + if (response != null) { + response.release(); + } + } + + assertThat(produceBuffer.refCnt()).isZero(); + assertThat(apiVersionsBuffer.refCnt()).isZero(); + } finally { + channel.finishAndReleaseAll(); + } + } + + private static ByteBuf serialize(RequestHeader header, AbstractRequest request) { + ByteBuffer serialized = + RequestUtils.serialize( + header.data(), header.headerVersion(), request.data(), request.version()); + return Unpooled.wrappedBuffer(serialized); + } +} diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaRequestHandlerTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaRequestHandlerTest.java index 24e4ce8a6ce..8613d83df32 100644 --- a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaRequestHandlerTest.java +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaRequestHandlerTest.java @@ -24,6 +24,7 @@ import org.apache.kafka.common.protocol.ApiKeys; import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.requests.AbstractRequest; import org.apache.kafka.common.requests.AbstractResponse; import org.apache.kafka.common.requests.ApiVersionsRequest; import org.apache.kafka.common.requests.ApiVersionsResponse; @@ -46,21 +47,15 @@ public void testKafkaApiVersionsNotSupported() { new ApiVersionsRequest.Builder().build(latestVersion); ChannelHandlerContext ctx = new TestingChannelHandlerContext(); KafkaRequest request = - new KafkaRequest( + newRequest( ApiKeys.API_VERSIONS, (short) (latestVersion + 1), // unsupported version new RequestHeader(ApiKeys.API_VERSIONS, latestVersion, "client-id", 0), apiVersionsRequest, - ByteBufAllocator.DEFAULT.buffer(), - ctx, - new CompletableFuture<>()); + ctx); handler.handleApiVersionsRequest(request); - ByteBuf responseBuffer = request.responseBuffer(); - ApiVersionsResponse response = - (ApiVersionsResponse) - AbstractResponse.parseResponse( - responseBuffer.nioBuffer(), request.header()); + ApiVersionsResponse response = (ApiVersionsResponse) parseResponse(request); Map errorCounts = response.errorCounts(); assertThat(1).isEqualTo(errorCounts.size()); assertThat(1).isEqualTo(errorCounts.get(Errors.UNSUPPORTED_VERSION)); @@ -74,21 +69,15 @@ public void testKafkaApiVersionsRequest() { new ApiVersionsRequest.Builder().build(latestVersion); ChannelHandlerContext ctx = new TestingChannelHandlerContext(); KafkaRequest request = - new KafkaRequest( + newRequest( ApiKeys.API_VERSIONS, latestVersion, new RequestHeader(ApiKeys.API_VERSIONS, latestVersion, "client-id", 0), apiVersionsRequest, - ByteBufAllocator.DEFAULT.buffer(), - ctx, - new CompletableFuture<>()); + ctx); handler.handleApiVersionsRequest(request); - ByteBuf responseBuffer = request.responseBuffer(); - ApiVersionsResponse response = - (ApiVersionsResponse) - AbstractResponse.parseResponse( - responseBuffer.nioBuffer(), request.header()); + ApiVersionsResponse response = (ApiVersionsResponse) parseResponse(request); Map errorCounts = response.errorCounts(); assertThat(1).isEqualTo(errorCounts.size()); assertThat(1).isEqualTo(errorCounts.get(Errors.NONE)); @@ -112,6 +101,37 @@ public void testKafkaApiVersionsRequest() { }); } + private static KafkaRequest newRequest( + ApiKeys apiKey, + short apiVersion, + RequestHeader header, + AbstractRequest requestBody, + ChannelHandlerContext context) { + ByteBuf requestBuffer = ByteBufAllocator.DEFAULT.buffer(); + try { + return new KafkaRequest( + apiKey, + apiVersion, + header, + requestBody, + requestBuffer, + context, + new CompletableFuture<>()); + } finally { + // Mirror KafkaCommandDecoder's ownership transfer to KafkaRequest. + requestBuffer.release(); + } + } + + private static AbstractResponse parseResponse(KafkaRequest request) { + ByteBuf responseBuffer = request.responseBuffer(); + try { + return AbstractResponse.parseResponse(responseBuffer.nioBuffer(), request.header()); + } finally { + responseBuffer.release(); + } + } + private static KafkaRequestHandler createKafkaRequestHandler() { return new KafkaRequestHandler(new TestingTabletGatewayService()); } diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaRequestTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaRequestTest.java new file mode 100644 index 00000000000..2aa9caf9ba9 --- /dev/null +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaRequestTest.java @@ -0,0 +1,58 @@ +/* + * 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.shaded.netty4.io.netty.buffer.ByteBuf; +import org.apache.fluss.shaded.netty4.io.netty.channel.ChannelHandlerContext; + +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.requests.ApiVersionsRequest; +import org.apache.kafka.common.requests.RequestHeader; +import org.junit.jupiter.api.Test; + +import java.util.concurrent.CompletableFuture; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** Tests for {@link KafkaRequest}. */ +public class KafkaRequestTest { + + @Test + public void testReleaseBufferIsIdempotent() { + short version = ApiKeys.API_VERSIONS.oldestVersion(); + ByteBuf buffer = mock(ByteBuf.class); + when(buffer.retain()).thenReturn(buffer); + KafkaRequest request = + new KafkaRequest( + ApiKeys.API_VERSIONS, + version, + new RequestHeader(ApiKeys.API_VERSIONS, version, "client-id", 1), + new ApiVersionsRequest.Builder().build(version), + buffer, + mock(ChannelHandlerContext.class), + new CompletableFuture<>()); + + request.releaseBuffer(); + request.releaseBuffer(); + + verify(buffer).retain(); + verify(buffer).release(); + } +} diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaApiRegistryTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaApiRegistryTest.java new file mode 100644 index 00000000000..84a889c16bd --- /dev/null +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaApiRegistryTest.java @@ -0,0 +1,127 @@ +/* + * 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.dispatcher; + +import org.apache.fluss.kafka.KafkaRequestContext; + +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.requests.AbstractResponse; +import org.apache.kafka.common.requests.ApiVersionsRequest; +import org.junit.jupiter.api.Test; + +import java.util.concurrent.CompletableFuture; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link KafkaApiRegistry}. */ +public class KafkaApiRegistryTest { + + @Test + public void testRejectDuplicateRegistrationAndRegistrationAfterFreeze() { + KafkaApiRegistry registry = brokerRegistry(); + TestingApiVersionsHandler handler = new TestingApiVersionsHandler(true); + registry.register(handler); + + assertThatThrownBy(() -> registry.register(handler)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("already registered"); + + registry.freeze(); + assertThatThrownBy(() -> registry.register(handler)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("already frozen"); + } + + @Test + public void testOnlyAdvertiseEnabledHandlers() { + KafkaApiRegistry registry = brokerRegistry(); + registry.register(new TestingApiVersionsHandler(true)); + assertThat(registry.advertisedApiSpecs()).hasSize(1); + + KafkaApiRegistry hiddenRegistry = brokerRegistry(); + hiddenRegistry.register(new TestingApiVersionsHandler(false)); + assertThat(hiddenRegistry.advertisedApiSpecs()).isEmpty(); + assertThat(hiddenRegistry.lookup(ApiKeys.API_VERSIONS)).isNull(); + } + + @Test + public void testAdvertisedSpecIsSameSpecUsedForRouting() { + KafkaApiRegistry registry = brokerRegistry(); + TestingApiVersionsHandler handler = new TestingApiVersionsHandler(true); + registry.register(handler); + registry.freeze(); + + KafkaApiSpec advertisedSpec = registry.advertisedApiSpecs().get(0); + KafkaApiHandler routedHandler = registry.lookup(ApiKeys.API_VERSIONS); + + assertThat(routedHandler).isSameAs(handler); + assertThat(routedHandler.apiSpec()).isSameAs(advertisedSpec); + for (short version : ApiKeys.API_VERSIONS.allVersions()) { + assertThat(advertisedSpec.supportsVersion(version)).isTrue(); + } + assertThat( + advertisedSpec.supportsVersion( + (short) (ApiKeys.API_VERSIONS.latestVersion() + 1))) + .isFalse(); + } + + @Test + public void testRejectInvalidVersionRange() { + assertThatThrownBy(() -> new KafkaApiSpec(ApiKeys.API_VERSIONS, (short) 1, (short) 0, true)) + .isInstanceOf(IllegalArgumentException.class); + assertThatThrownBy( + () -> + new KafkaApiSpec( + ApiKeys.API_VERSIONS, + ApiKeys.API_VERSIONS.oldestVersion(), + (short) (ApiKeys.API_VERSIONS.latestVersion() + 1), + true)) + .isInstanceOf(IllegalArgumentException.class); + } + + private static KafkaApiRegistry brokerRegistry() { + return new KafkaApiRegistry(); + } + + private static final class TestingApiVersionsHandler + implements KafkaApiHandler { + + private final KafkaApiSpec spec; + + private TestingApiVersionsHandler(boolean advertised) { + this.spec = + new KafkaApiSpec( + ApiKeys.API_VERSIONS, + ApiKeys.API_VERSIONS.oldestVersion(), + ApiKeys.API_VERSIONS.latestVersion(), + advertised); + } + + @Override + public KafkaApiSpec apiSpec() { + return spec; + } + + @Override + public CompletableFuture handle( + KafkaRequestContext context, ApiVersionsRequest request) { + throw new UnsupportedOperationException(); + } + } +} diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcherTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcherTest.java new file mode 100644 index 00000000000..97f42350198 --- /dev/null +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcherTest.java @@ -0,0 +1,166 @@ +/* + * 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.dispatcher; + +import org.apache.fluss.kafka.KafkaRequest; +import org.apache.fluss.kafka.KafkaRequestContext; +import org.apache.fluss.kafka.error.KafkaErrorMapper; +import org.apache.fluss.shaded.netty4.io.netty.channel.ChannelHandlerContext; + +import org.apache.kafka.common.errors.InvalidRequestException; +import org.apache.kafka.common.message.ApiVersionsResponseData; +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.requests.AbstractResponse; +import org.apache.kafka.common.requests.ApiVersionsRequest; +import org.apache.kafka.common.requests.ApiVersionsResponse; +import org.apache.kafka.common.requests.RequestHeader; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BiFunction; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** Tests request routing and failure handling without a concrete API implementation. */ +class KafkaRequestDispatcherTest { + + @Test + void testUnregisteredApiReturnsUnsupportedVersion() { + KafkaApiRegistry registry = new KafkaApiRegistry(); + registry.freeze(); + + AbstractResponse response = + new KafkaRequestDispatcher(registry, new KafkaErrorMapper()) + .dispatch(request((short) 0)) + .join(); + + assertThat(response.errorCounts()).containsEntry(Errors.UNSUPPORTED_VERSION, 1); + } + + @Test + void testUnsupportedVersionDoesNotInvokeHandler() { + KafkaRequestDispatcher dispatcher = + dispatcher( + (context, request) -> { + throw new AssertionError( + "Unsupported versions must not be dispatched."); + }); + + AbstractResponse response = dispatcher.dispatch(request((short) 1)).join(); + + assertThat(response.errorCounts()).containsEntry(Errors.UNSUPPORTED_VERSION, 1); + } + + @Test + void testDispatchWaitsForHandlerAndPreservesContext() { + CompletableFuture handlerResult = new CompletableFuture<>(); + AtomicReference receivedContext = new AtomicReference<>(); + KafkaRequestDispatcher dispatcher = + dispatcher( + (context, request) -> { + receivedContext.set(context); + return handlerResult; + }); + KafkaRequest request = request((short) 0); + + CompletableFuture result = dispatcher.dispatch(request); + + assertThat(result).isNotDone(); + assertThat(receivedContext.get().clientId()).isEqualTo("client"); + assertThat(receivedContext.get().correlationId()).isEqualTo(42); + assertThat(receivedContext.get().listenerName()).isEqualTo("KAFKA"); + assertThat(receivedContext.get().apiKey()).isEqualTo(ApiKeys.API_VERSIONS); + assertThat(receivedContext.get().apiVersion()).isZero(); + AbstractResponse response = new ApiVersionsResponse(new ApiVersionsResponseData()); + handlerResult.complete(response); + assertThat(result.join()).isSameAs(response); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testSynchronousAndAsynchronousFailuresBecomeErrorResponses(boolean synchronous) { + InvalidRequestException failure = new InvalidRequestException("invalid request"); + KafkaRequestDispatcher dispatcher = + dispatcher( + (context, request) -> { + if (synchronous) { + throw failure; + } + CompletableFuture result = new CompletableFuture<>(); + result.completeExceptionally(new CompletionException(failure)); + return result; + }); + + AbstractResponse response = dispatcher.dispatch(request((short) 0)).join(); + + assertThat(response.errorCounts()).containsEntry(Errors.INVALID_REQUEST, 1); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testNullFutureAndNullResponseBecomeErrorResponses(boolean nullFuture) { + KafkaRequestDispatcher dispatcher = + dispatcher( + (context, request) -> + nullFuture ? null : CompletableFuture.completedFuture(null)); + + AbstractResponse response = dispatcher.dispatch(request((short) 0)).join(); + + assertThat(response.errorCounts()).containsEntry(Errors.UNKNOWN_SERVER_ERROR, 1); + } + + private static KafkaRequestDispatcher dispatcher( + BiFunction> + action) { + KafkaApiRegistry registry = new KafkaApiRegistry(); + registry.register( + new KafkaApiHandler() { + @Override + public KafkaApiSpec apiSpec() { + return new KafkaApiSpec(ApiKeys.API_VERSIONS, (short) 0, (short) 0, true); + } + + @Override + public CompletableFuture handle( + KafkaRequestContext context, ApiVersionsRequest request) { + return action.apply(context, request); + } + }); + registry.freeze(); + return new KafkaRequestDispatcher(registry, new KafkaErrorMapper()); + } + + private static KafkaRequest request(short version) { + KafkaRequest request = mock(KafkaRequest.class); + when(request.apiKey()).thenReturn(ApiKeys.API_VERSIONS); + when(request.apiVersion()).thenReturn(version); + when(request.request()).thenReturn(new ApiVersionsRequest.Builder().build(version)); + when(request.header()) + .thenReturn(new RequestHeader(ApiKeys.API_VERSIONS, version, "client", 42)); + when(request.listenerName()).thenReturn("KAFKA"); + when(request.ctx()).thenReturn(mock(ChannelHandlerContext.class)); + return request; + } +} diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServer.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServer.java index 03d798fb371..df2256985c2 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServer.java +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/NettyServer.java @@ -232,8 +232,17 @@ private static List loadProtocols( NetworkProtocolPlugin kafkaPlugin = loadProtocolPlugin(NetworkProtocolPlugin.KAFKA_PROTOCOL_NAME); kafkaPlugin.setup(conf); - listeners.removeAll(kafkaPlugin.listenerNames()); - protocolPlugins.add(kafkaPlugin); + List kafkaListenerNames = kafkaPlugin.listenerNames(); + boolean hasKafkaEndpoint = + endpoints.stream() + .anyMatch( + endpoint -> + kafkaListenerNames.contains( + endpoint.getListenerName())); + if (hasKafkaEndpoint) { + listeners.removeAll(kafkaListenerNames); + protocolPlugins.add(kafkaPlugin); + } } // Add the Fluss protocol plugin in the end to allow other protocol From 6fb0e747b7950f055951163b91b872fb28150aec Mon Sep 17 00:00:00 2001 From: Yang Guo Date: Wed, 16 Sep 2026 17:35:40 +0800 Subject: [PATCH 2/6] [kafka] Complete dispatcher futures when error mapping fails Catch failures while mapping asynchronous handler results so the dispatcher future completes exceptionally instead of remaining pending. Cover handler futures that fail before and after dispatch registers its callback. Validated with Java 11: Maven reactor build and 17 targeted Kafka tests, including Checkstyle, Spotless, and license checks. Co-Authored-By: Codex AI-Model: gpt-6 AI-Contributed/Feature: 27/27 AI-Contributed/UT: 28/28 --- .../dispatcher/KafkaRequestDispatcher.java | 27 +++++++++++------- .../KafkaRequestDispatcherTest.java | 28 +++++++++++++++++++ 2 files changed, 45 insertions(+), 10 deletions(-) diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcher.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcher.java index efdfe33dd66..235303f5f0e 100644 --- a/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcher.java +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcher.java @@ -43,7 +43,7 @@ public KafkaRequestDispatcher(KafkaApiRegistry registry, KafkaErrorMapper errorM this.errorMapper = checkNotNull(errorMapper); } - /** Dispatches a request and always completes with a Kafka protocol response. */ + /** Dispatches a request and maps handler failures to Kafka protocol responses. */ public CompletableFuture dispatch(KafkaRequest request) { AbstractRequest abstractRequest = request.request(); KafkaApiHandler handler = registry.lookup(request.apiKey()); @@ -81,15 +81,22 @@ public CompletableFuture dispatch(KafkaRequest request) { CompletableFuture result = new CompletableFuture<>(); responseFuture.whenComplete( (response, failure) -> { - if (failure == null && response != null) { - result.complete(response); - } else { - Throwable responseFailure = - failure == null - ? new NullPointerException( - "Kafka API handler returned a null response.") - : failure; - result.complete(errorMapper.toResponse(abstractRequest, responseFailure)); + try { + if (failure == null && response != null) { + result.complete(response); + } else { + Throwable responseFailure = + failure == null + ? new NullPointerException( + "Kafka API handler returned a null response.") + : failure; + result.complete( + errorMapper.toResponse(abstractRequest, responseFailure)); + } + } catch (Throwable completionFailure) { + // Completion callbacks must never leave the ordered Kafka response queue + // waiting on a future that can no longer become terminal. + result.completeExceptionally(completionFailure); } }); return result; diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcherTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcherTest.java index 97f42350198..a1e244bf310 100644 --- a/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcherTest.java +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcherTest.java @@ -40,6 +40,8 @@ import java.util.function.BiFunction; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -131,6 +133,32 @@ void testNullFutureAndNullResponseBecomeErrorResponses(boolean nullFuture) { assertThat(response.errorCounts()).containsEntry(Errors.UNKNOWN_SERVER_ERROR, 1); } + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testErrorMappingFailureCompletesDispatcherFutureExceptionally( + boolean handlerAlreadyCompleted) { + CompletableFuture handlerResult = new CompletableFuture<>(); + KafkaRequestDispatcher dispatcher = dispatcher((context, request) -> handlerResult); + KafkaRequest request = request((short) 0); + ApiVersionsRequest requestBody = mock(ApiVersionsRequest.class); + IllegalStateException mappingFailure = new IllegalStateException("error mapping failure"); + when(requestBody.getErrorResponse(any(Throwable.class))).thenThrow(mappingFailure); + when(request.request()).thenReturn(requestBody); + InvalidRequestException handlerFailure = new InvalidRequestException("invalid request"); + if (handlerAlreadyCompleted) { + handlerResult.completeExceptionally(handlerFailure); + } + + CompletableFuture result = dispatcher.dispatch(request); + if (!handlerAlreadyCompleted) { + assertThat(result).isNotDone(); + handlerResult.completeExceptionally(handlerFailure); + } + + assertThat(result).isCompletedExceptionally(); + assertThatThrownBy(result::join).hasCause(mappingFailure); + } + private static KafkaRequestDispatcher dispatcher( BiFunction> action) { From 274c00d7af9f0347adeb9372e5f67b41222eed2e Mon Sep 17 00:00:00 2001 From: Yang Guo Date: Sat, 10 Oct 2026 17:11:28 +0800 Subject: [PATCH 3/6] [kafka] Reuse Netty idle timeout for Kafka listeners Read netty.connection.max-idle-time when building the Kafka pipeline, keeping the existing IdleStateHandler and ALL_IDLE close behavior. Cover default, custom, and disabled timeouts, legacy option isolation, and idle connection closure through a real Kafka listener. Validated with Java 11 using this worktree's reactor dependencies: mvn -o -pl fluss-kafka -am verify -Dtest=Kafka*Test,Kafka*ITCase -Dsurefire.failIfNoSpecifiedTests=false All 25 Kafka cases passed in both test executions; Checkstyle, Spotless and RAT checks passed. The new configuration regression cases failed on the previous implementation before the wiring change. Co-Authored-By: Codex AI-Model: gpt-6 AI-Contributed/Feature: 2/2 AI-Contributed/UT: 175/175 --- .../fluss/kafka/KafkaProtocolPlugin.java | 2 +- .../kafka/KafkaProtocolPluginITCase.java | 67 +++++++++++ .../fluss/kafka/KafkaProtocolPluginTest.java | 108 ++++++++++++++++++ 3 files changed, 176 insertions(+), 1 deletion(-) create mode 100644 fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaProtocolPluginITCase.java create mode 100644 fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaProtocolPluginTest.java diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaProtocolPlugin.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaProtocolPlugin.java index c966f745b8c..2bd2f5d81bc 100644 --- a/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaProtocolPlugin.java +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaProtocolPlugin.java @@ -54,7 +54,7 @@ public ChannelHandler createChannelHandler( return new KafkaChannelInitializer( requestChannels, listenerName, - conf.get(ConfigOptions.KAFKA_CONNECTION_MAX_IDLE_TIME).getSeconds(), + conf.get(ConfigOptions.NETTY_CONNECTION_MAX_IDLE_TIME).getSeconds(), (int) conf.get(ConfigOptions.NETTY_SERVER_MAX_REQUEST_SIZE).getBytes(), conf.getBoolean(ConfigOptions.NETTY_CLIENT_ALLOCATOR_HEAP_BUFFER_FIRST)); } diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaProtocolPluginITCase.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaProtocolPluginITCase.java new file mode 100644 index 00000000000..8cf4835148f --- /dev/null +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaProtocolPluginITCase.java @@ -0,0 +1,67 @@ +/* + * 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.cluster.Endpoint; +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.metrics.groups.MetricGroup; +import org.apache.fluss.metrics.util.NOPMetricsGroup; +import org.apache.fluss.rpc.TestingTabletGatewayService; +import org.apache.fluss.rpc.netty.server.NettyServer; +import org.apache.fluss.rpc.netty.server.RequestsMetrics; + +import org.junit.jupiter.api.Test; + +import java.net.InetSocketAddress; +import java.net.Socket; +import java.time.Duration; +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests idle connection closure through the Kafka listener's actual Netty pipeline. */ +class KafkaProtocolPluginITCase { + + @Test + void testKafkaConnectionClosesAfterConfiguredNettyIdleTimeout() throws Exception { + Configuration conf = new Configuration(); + conf.set(ConfigOptions.KAFKA_ENABLED, true); + conf.set(ConfigOptions.NETTY_SERVER_NUM_WORKER_THREADS, 1); + conf.set(ConfigOptions.NETTY_SERVER_NUM_NETWORK_THREADS, 1); + conf.set(ConfigOptions.NETTY_CONNECTION_MAX_IDLE_TIME, Duration.ofSeconds(1)); + conf.setString("kafka.connection.max-idle-time", "1 min"); + MetricGroup metricGroup = NOPMetricsGroup.newInstance(); + try (NettyServer server = + new NettyServer( + conf, + Collections.singletonList(new Endpoint("localhost", 0, "KAFKA")), + new TestingTabletGatewayService(), + metricGroup, + RequestsMetrics.createCoordinatorServerRequestMetrics( + metricGroup)); + Socket socket = new Socket()) { + server.start(); + Endpoint endpoint = server.getBindEndpoints().get(0); + socket.connect(new InetSocketAddress(endpoint.getHost(), endpoint.getPort()), 10_000); + socket.setSoTimeout(10_000); + + assertThat(socket.getInputStream().read()).isEqualTo(-1); + } + } +} diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaProtocolPluginTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaProtocolPluginTest.java new file mode 100644 index 00000000000..dd270dbd563 --- /dev/null +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaProtocolPluginTest.java @@ -0,0 +1,108 @@ +/* + * 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.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.rpc.netty.server.RequestChannel; +import org.apache.fluss.shaded.netty4.io.netty.channel.embedded.EmbeddedChannel; +import org.apache.fluss.shaded.netty4.io.netty.channel.socket.SocketChannel; +import org.apache.fluss.shaded.netty4.io.netty.handler.timeout.IdleStateEvent; +import org.apache.fluss.shaded.netty4.io.netty.handler.timeout.IdleStateHandler; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.time.Duration; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** Tests Kafka listener idle-time configuration and idle-event handling. */ +class KafkaProtocolPluginTest { + + @Test + void testDefaultNettyIdleTimeout() throws Exception { + EmbeddedChannel channel = createChannel(new Configuration()); + try { + IdleStateHandler idleHandler = channel.pipeline().get(IdleStateHandler.class); + assertThat(idleHandler.getAllIdleTimeInMillis()) + .isEqualTo(Duration.ofMinutes(10).toMillis()); + assertThat(idleHandler.getReaderIdleTimeInMillis()).isZero(); + assertThat(idleHandler.getWriterIdleTimeInMillis()).isZero(); + } finally { + channel.finishAndReleaseAll(); + } + } + + @ParameterizedTest + @ValueSource(longs = {0, 7}) + void testConfiguredNettyIdleTimeoutOverridesLegacyKafkaOption(long timeoutSeconds) + throws Exception { + Configuration conf = new Configuration(); + conf.set(ConfigOptions.NETTY_CONNECTION_MAX_IDLE_TIME, Duration.ofSeconds(timeoutSeconds)); + conf.setString("kafka.connection.max-idle-time", "2 s"); + EmbeddedChannel channel = createChannel(conf); + try { + IdleStateHandler idleHandler = channel.pipeline().get(IdleStateHandler.class); + assertThat(idleHandler.getAllIdleTimeInMillis()) + .isEqualTo(Duration.ofSeconds(timeoutSeconds).toMillis()); + assertThat(idleHandler.getReaderIdleTimeInMillis()).isZero(); + assertThat(idleHandler.getWriterIdleTimeInMillis()).isZero(); + } finally { + channel.finishAndReleaseAll(); + } + } + + @Test + void testOnlyAllIdleClosesConnection() throws Exception { + EmbeddedChannel channel = createChannel(new Configuration()); + try { + channel.pipeline().fireUserEventTriggered(IdleStateEvent.FIRST_READER_IDLE_STATE_EVENT); + assertThat(channel.isActive()).isTrue(); + channel.pipeline().fireUserEventTriggered(IdleStateEvent.FIRST_WRITER_IDLE_STATE_EVENT); + assertThat(channel.isActive()).isTrue(); + channel.pipeline().fireUserEventTriggered(IdleStateEvent.FIRST_ALL_IDLE_STATE_EVENT); + assertThat(channel.isActive()).isFalse(); + } finally { + channel.finishAndReleaseAll(); + } + } + + private static EmbeddedChannel createChannel(Configuration conf) throws Exception { + EmbeddedChannel channel = new EmbeddedChannel(); + try { + // Adapt the socket type while installing the real pipeline on the embedded event loop. + SocketChannel socket = mock(SocketChannel.class); + when(socket.pipeline()).thenReturn(channel.pipeline()); + KafkaProtocolPlugin plugin = new KafkaProtocolPlugin(); + plugin.setup(conf); + KafkaChannelInitializer initializer = + (KafkaChannelInitializer) + plugin.createChannelHandler( + new RequestChannel[] {new RequestChannel(1)}, "KAFKA"); + initializer.initChannel(socket); + return channel; + } catch (Exception | Error failure) { + channel.finishAndReleaseAll(); + throw failure; + } + } +} From 3848096e9a51ca564f161be9fa4b3612ba33292e Mon Sep 17 00:00:00 2001 From: Yang Guo Date: Mon, 12 Oct 2026 16:40:53 +0800 Subject: [PATCH 4/6] [kafka] Remove obsolete Kafka idle timeout option Remove the unused Kafka-specific idle timeout configuration and its documentation. Document the shared Netty option and update listener tests to configure it directly. Co-Authored-By: Codex AI-Model: gpt-6 AI-Contributed/Feature: 11/11 AI-Contributed/UT: 5/5 --- .../main/java/org/apache/fluss/config/ConfigOptions.java | 7 ------- .../org/apache/fluss/kafka/KafkaProtocolPluginITCase.java | 1 - .../org/apache/fluss/kafka/KafkaProtocolPluginTest.java | 4 +--- website/docs/maintenance/configuration.md | 4 +++- 4 files changed, 4 insertions(+), 12 deletions(-) diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java index e497ea07039..b926c74c411 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java @@ -2903,13 +2903,6 @@ public class ConfigOptions { .withDescription( "The database for fluss kafka. The default database is `kafka`."); - public static final ConfigOption KAFKA_CONNECTION_MAX_IDLE_TIME = - key("kafka.connection.max-idle-time") - .durationType() - .defaultValue(Duration.ofSeconds(60)) - .withDescription( - "Close kafka idle connections after the given time specified by this config."); - /** * Compaction style for Fluss's kv, which is same to rocksdb's, but help use avoid including * rocksdb dependency when only need include this common module. diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaProtocolPluginITCase.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaProtocolPluginITCase.java index 8cf4835148f..a10afcb372c 100644 --- a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaProtocolPluginITCase.java +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaProtocolPluginITCase.java @@ -45,7 +45,6 @@ void testKafkaConnectionClosesAfterConfiguredNettyIdleTimeout() throws Exception conf.set(ConfigOptions.NETTY_SERVER_NUM_WORKER_THREADS, 1); conf.set(ConfigOptions.NETTY_SERVER_NUM_NETWORK_THREADS, 1); conf.set(ConfigOptions.NETTY_CONNECTION_MAX_IDLE_TIME, Duration.ofSeconds(1)); - conf.setString("kafka.connection.max-idle-time", "1 min"); MetricGroup metricGroup = NOPMetricsGroup.newInstance(); try (NettyServer server = new NettyServer( diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaProtocolPluginTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaProtocolPluginTest.java index dd270dbd563..3d0ded2756b 100644 --- a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaProtocolPluginTest.java +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaProtocolPluginTest.java @@ -54,11 +54,9 @@ void testDefaultNettyIdleTimeout() throws Exception { @ParameterizedTest @ValueSource(longs = {0, 7}) - void testConfiguredNettyIdleTimeoutOverridesLegacyKafkaOption(long timeoutSeconds) - throws Exception { + void testConfiguredNettyIdleTimeout(long timeoutSeconds) throws Exception { Configuration conf = new Configuration(); conf.set(ConfigOptions.NETTY_CONNECTION_MAX_IDLE_TIME, Duration.ofSeconds(timeoutSeconds)); - conf.setString("kafka.connection.max-idle-time", "2 s"); EmbeddedChannel channel = createChannel(conf); try { IdleStateHandler idleHandler = channel.pipeline().get(IdleStateHandler.class); diff --git a/website/docs/maintenance/configuration.md b/website/docs/maintenance/configuration.md index 3e7e986d03c..ce76beca9a1 100644 --- a/website/docs/maintenance/configuration.md +++ b/website/docs/maintenance/configuration.md @@ -255,4 +255,6 @@ Kafka protocol compatibility is still in development. | kafka.enabled | Boolean | false | Whether enable Fluss Kafka. Disabled by default. When this option is set to true, the Fluss Kafka will be enabled. | | kafka.listener.names | String | KAFKA | The listener names for Kafka wire protocol communication. Support multiple listener names, separated by comma. | | kafka.database | String | kafka | The database for Fluss Kafka. The default database is `kafka`. | -| kafka.connection.max-idle-time | Duration | 60s | Close kafka idle connections after the given time specified by this config. | + +Kafka listeners share `netty.connection.max-idle-time` with native Fluss RPC listeners (see [Netty](#netty)). +The default is `10min`; connections close when neither reads nor writes occur for the configured duration. From 9008e9fc6325e9025a1bd77fd566710b6c992dca Mon Sep 17 00:00:00 2001 From: Yang Guo Date: Mon, 12 Oct 2026 23:14:13 +0800 Subject: [PATCH 5/6] [kafka] Make API registry immutable and harden error mapping Construct the registry from a defensive copy of the handler collection and require explicit listener names when creating Kafka requests. Return failed futures when synchronous error mapping fails, preserving buffer ownership. Add registry immutability and synchronous error-mapping regression coverage. Co-Authored-By: Codex AI-Model: gpt-6 AI-Contributed/Feature: 54/54 AI-Contributed/UT: 152/152 --- .../org/apache/fluss/kafka/KafkaRequest.java | 11 --- .../kafka/dispatcher/KafkaApiRegistry.java | 36 ++++--- .../dispatcher/KafkaRequestDispatcher.java | 7 +- .../fluss/kafka/KafkaRequestHandlerTest.java | 1 + .../apache/fluss/kafka/KafkaRequestTest.java | 1 + .../dispatcher/KafkaApiRegistryTest.java | 94 +++++++++++++------ .../KafkaRequestDispatcherTest.java | 56 +++++++++-- 7 files changed, 138 insertions(+), 68 deletions(-) diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaRequest.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaRequest.java index 0d2799a7a18..fe31c295e2a 100644 --- a/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaRequest.java +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaRequest.java @@ -55,17 +55,6 @@ public class KafkaRequest implements RpcRequest { private final AtomicBoolean bufferReleased = new AtomicBoolean(); private volatile boolean cancelled = false; - protected KafkaRequest( - ApiKeys apiKey, - short apiVersion, - RequestHeader header, - AbstractRequest request, - ByteBuf buffer, - ChannelHandlerContext ctx, - CompletableFuture future) { - this(apiKey, apiVersion, header, request, "UNKNOWN", buffer, ctx, future); - } - protected KafkaRequest( ApiKeys apiKey, short apiVersion, diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaApiRegistry.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaApiRegistry.java index b4a1dfd8ea3..7b7f0f848b4 100644 --- a/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaApiRegistry.java +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaApiRegistry.java @@ -22,6 +22,7 @@ import org.apache.kafka.common.protocol.ApiKeys; import java.util.ArrayList; +import java.util.Collection; import java.util.Collections; import java.util.Comparator; import java.util.HashMap; @@ -30,30 +31,27 @@ import static org.apache.fluss.utils.Preconditions.checkArgument; import static org.apache.fluss.utils.Preconditions.checkNotNull; -import static org.apache.fluss.utils.Preconditions.checkState; -/** Registry and single source of truth for Kafka APIs exposed by one server. */ +/** Immutable registry and single source of truth for Kafka APIs exposed by one server. */ @Internal public final class KafkaApiRegistry { - private final Map> handlers = new HashMap<>(); - private boolean frozen; + private final Map> handlers; - /** Creates an empty API registry. */ - public KafkaApiRegistry() {} - - /** Registers a handler. Registrations are rejected after {@link #freeze()} is called. */ - public void register(KafkaApiHandler handler) { - checkNotNull(handler); - checkState(!frozen, "Kafka API registry is already frozen."); - ApiKeys apiKey = handler.apiSpec().apiKey(); - checkArgument(!handlers.containsKey(apiKey), "Kafka API %s is already registered.", apiKey); - handlers.put(apiKey, handler); - } - - /** Prevents further registrations. */ - public void freeze() { - frozen = true; + /** + * Creates a registry from a snapshot of the supplied handlers, rejecting duplicate API keys. + */ + public KafkaApiRegistry(Collection> handlers) { + Map> registeredHandlers = new HashMap<>(); + for (KafkaApiHandler handler : checkNotNull(handlers)) { + ApiKeys apiKey = checkNotNull(handler).apiSpec().apiKey(); + checkArgument( + !registeredHandlers.containsKey(apiKey), + "Kafka API %s is already registered.", + apiKey); + registeredHandlers.put(apiKey, handler); + } + this.handlers = Collections.unmodifiableMap(registeredHandlers); } /** Returns a routable handler, or {@code null} when the API is not exposed by this server. */ diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcher.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcher.java index 235303f5f0e..b12c13171dc 100644 --- a/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcher.java +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcher.java @@ -21,6 +21,7 @@ import org.apache.fluss.kafka.KafkaRequest; import org.apache.fluss.kafka.KafkaRequestContext; import org.apache.fluss.kafka.error.KafkaErrorMapper; +import org.apache.fluss.utils.concurrent.FutureUtils; import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.requests.AbstractRequest; @@ -110,6 +111,10 @@ private static CompletableFuture invoke( private CompletableFuture completedErrorResponse( AbstractRequest request, Throwable failure) { - return CompletableFuture.completedFuture(errorMapper.toResponse(request, failure)); + try { + return CompletableFuture.completedFuture(errorMapper.toResponse(request, failure)); + } catch (Throwable mappingFailure) { + return FutureUtils.completedExceptionally(mappingFailure); + } } } diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaRequestHandlerTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaRequestHandlerTest.java index 8613d83df32..bf8e23b4356 100644 --- a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaRequestHandlerTest.java +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaRequestHandlerTest.java @@ -114,6 +114,7 @@ private static KafkaRequest newRequest( apiVersion, header, requestBody, + "KAFKA", requestBuffer, context, new CompletableFuture<>()); diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaRequestTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaRequestTest.java index 2aa9caf9ba9..fa1dfa90d09 100644 --- a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaRequestTest.java +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaRequestTest.java @@ -45,6 +45,7 @@ public void testReleaseBufferIsIdempotent() { version, new RequestHeader(ApiKeys.API_VERSIONS, version, "client-id", 1), new ApiVersionsRequest.Builder().build(version), + "KAFKA", buffer, mock(ChannelHandlerContext.class), new CompletableFuture<>()); diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaApiRegistryTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaApiRegistryTest.java index 84a889c16bd..4cf00860cc9 100644 --- a/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaApiRegistryTest.java +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaApiRegistryTest.java @@ -20,10 +20,14 @@ import org.apache.fluss.kafka.KafkaRequestContext; import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.requests.AbstractRequest; import org.apache.kafka.common.requests.AbstractResponse; -import org.apache.kafka.common.requests.ApiVersionsRequest; import org.junit.jupiter.api.Test; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; import java.util.concurrent.CompletableFuture; import static org.assertj.core.api.Assertions.assertThat; @@ -33,39 +37,75 @@ public class KafkaApiRegistryTest { @Test - public void testRejectDuplicateRegistrationAndRegistrationAfterFreeze() { - KafkaApiRegistry registry = brokerRegistry(); - TestingApiVersionsHandler handler = new TestingApiVersionsHandler(true); - registry.register(handler); - - assertThatThrownBy(() -> registry.register(handler)) + public void testRejectDuplicateApiKeys() { + assertThatThrownBy( + () -> + new KafkaApiRegistry( + Arrays.asList( + new TestingApiHandler(ApiKeys.API_VERSIONS, true), + new TestingApiHandler( + ApiKeys.API_VERSIONS, false)))) .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("already registered"); + } + + @Test + public void testEmptyRegistry() { + KafkaApiRegistry registry = new KafkaApiRegistry(Collections.emptyList()); + + assertThat(registry.lookup(ApiKeys.API_VERSIONS)).isNull(); + assertThat(registry.advertisedApiSpecs()).isEmpty(); + } - registry.freeze(); - assertThatThrownBy(() -> registry.register(handler)) - .isInstanceOf(IllegalStateException.class) - .hasMessageContaining("already frozen"); + @Test + public void testChangesToInputCollectionDoNotAffectRegistry() { + TestingApiHandler handler = new TestingApiHandler(ApiKeys.API_VERSIONS, true); + List> handlers = new ArrayList<>(); + handlers.add(handler); + KafkaApiRegistry registry = new KafkaApiRegistry(handlers); + + handlers.clear(); + handlers.add(new TestingApiHandler(ApiKeys.METADATA, true)); + + assertThat(registry.lookup(ApiKeys.API_VERSIONS)).isSameAs(handler); + assertThat(registry.lookup(ApiKeys.METADATA)).isNull(); + assertThat(registry.advertisedApiSpecs()).containsExactly(handler.apiSpec()); + } + + @Test + public void testAdvertisedSpecsAreSortedAndUnmodifiable() { + TestingApiHandler apiVersions = new TestingApiHandler(ApiKeys.API_VERSIONS, true); + TestingApiHandler metadata = new TestingApiHandler(ApiKeys.METADATA, true); + KafkaApiRegistry registry = new KafkaApiRegistry(Arrays.asList(apiVersions, metadata)); + + List specs = registry.advertisedApiSpecs(); + + assertThat(specs).containsExactly(metadata.apiSpec(), apiVersions.apiSpec()); + assertThatThrownBy(specs::clear).isInstanceOf(UnsupportedOperationException.class); + assertThat(registry.advertisedApiSpecs()) + .containsExactly(metadata.apiSpec(), apiVersions.apiSpec()); } @Test public void testOnlyAdvertiseEnabledHandlers() { - KafkaApiRegistry registry = brokerRegistry(); - registry.register(new TestingApiVersionsHandler(true)); + KafkaApiRegistry registry = + new KafkaApiRegistry( + Collections.singletonList( + new TestingApiHandler(ApiKeys.API_VERSIONS, true))); assertThat(registry.advertisedApiSpecs()).hasSize(1); - KafkaApiRegistry hiddenRegistry = brokerRegistry(); - hiddenRegistry.register(new TestingApiVersionsHandler(false)); + KafkaApiRegistry hiddenRegistry = + new KafkaApiRegistry( + Collections.singletonList( + new TestingApiHandler(ApiKeys.API_VERSIONS, false))); assertThat(hiddenRegistry.advertisedApiSpecs()).isEmpty(); assertThat(hiddenRegistry.lookup(ApiKeys.API_VERSIONS)).isNull(); } @Test public void testAdvertisedSpecIsSameSpecUsedForRouting() { - KafkaApiRegistry registry = brokerRegistry(); - TestingApiVersionsHandler handler = new TestingApiVersionsHandler(true); - registry.register(handler); - registry.freeze(); + TestingApiHandler handler = new TestingApiHandler(ApiKeys.API_VERSIONS, true); + KafkaApiRegistry registry = new KafkaApiRegistry(Collections.singletonList(handler)); KafkaApiSpec advertisedSpec = registry.advertisedApiSpecs().get(0); KafkaApiHandler routedHandler = registry.lookup(ApiKeys.API_VERSIONS); @@ -95,22 +135,14 @@ public void testRejectInvalidVersionRange() { .isInstanceOf(IllegalArgumentException.class); } - private static KafkaApiRegistry brokerRegistry() { - return new KafkaApiRegistry(); - } - - private static final class TestingApiVersionsHandler - implements KafkaApiHandler { + private static final class TestingApiHandler implements KafkaApiHandler { private final KafkaApiSpec spec; - private TestingApiVersionsHandler(boolean advertised) { + private TestingApiHandler(ApiKeys apiKey, boolean advertised) { this.spec = new KafkaApiSpec( - ApiKeys.API_VERSIONS, - ApiKeys.API_VERSIONS.oldestVersion(), - ApiKeys.API_VERSIONS.latestVersion(), - advertised); + apiKey, apiKey.oldestVersion(), apiKey.latestVersion(), advertised); } @Override @@ -120,7 +152,7 @@ public KafkaApiSpec apiSpec() { @Override public CompletableFuture handle( - KafkaRequestContext context, ApiVersionsRequest request) { + KafkaRequestContext context, AbstractRequest request) { throw new UnsupportedOperationException(); } } diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcherTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcherTest.java index a1e244bf310..d2bbbcc25ca 100644 --- a/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcherTest.java +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaRequestDispatcherTest.java @@ -34,6 +34,7 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; +import java.util.Collections; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.atomic.AtomicReference; @@ -50,8 +51,7 @@ class KafkaRequestDispatcherTest { @Test void testUnregisteredApiReturnsUnsupportedVersion() { - KafkaApiRegistry registry = new KafkaApiRegistry(); - registry.freeze(); + KafkaApiRegistry registry = new KafkaApiRegistry(Collections.emptyList()); AbstractResponse response = new KafkaRequestDispatcher(registry, new KafkaErrorMapper()) @@ -159,11 +159,55 @@ void testErrorMappingFailureCompletesDispatcherFutureExceptionally( assertThatThrownBy(result::join).hasCause(mappingFailure); } + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testSynchronousErrorMappingFailureCompletesFutureExceptionally(boolean nullFuture) { + KafkaRequestDispatcher dispatcher = + dispatcher( + (context, request) -> { + if (nullFuture) { + return null; + } + throw new InvalidRequestException("invalid request"); + }); + + assertErrorMappingFailureCompletesFutureExceptionally(dispatcher, request((short) 0)); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testRejectedRequestErrorMappingFailureCompletesFutureExceptionally(boolean registeredApi) { + KafkaRequestDispatcher dispatcher = + registeredApi + ? dispatcher( + (context, request) -> { + throw new AssertionError( + "Unsupported versions must not be dispatched."); + }) + : new KafkaRequestDispatcher( + new KafkaApiRegistry(Collections.emptyList()), + new KafkaErrorMapper()); + + assertErrorMappingFailureCompletesFutureExceptionally(dispatcher, request((short) 1)); + } + + private static void assertErrorMappingFailureCompletesFutureExceptionally( + KafkaRequestDispatcher dispatcher, KafkaRequest request) { + ApiVersionsRequest requestBody = mock(ApiVersionsRequest.class); + IllegalStateException mappingFailure = new IllegalStateException("error mapping failure"); + when(requestBody.getErrorResponse(any(Throwable.class))).thenThrow(mappingFailure); + when(request.request()).thenReturn(requestBody); + + CompletableFuture result = dispatcher.dispatch(request); + + assertThat(result).isCompletedExceptionally(); + assertThatThrownBy(result::join).hasCause(mappingFailure); + } + private static KafkaRequestDispatcher dispatcher( BiFunction> action) { - KafkaApiRegistry registry = new KafkaApiRegistry(); - registry.register( + KafkaApiHandler handler = new KafkaApiHandler() { @Override public KafkaApiSpec apiSpec() { @@ -175,8 +219,8 @@ public CompletableFuture handle( KafkaRequestContext context, ApiVersionsRequest request) { return action.apply(context, request); } - }); - registry.freeze(); + }; + KafkaApiRegistry registry = new KafkaApiRegistry(Collections.singletonList(handler)); return new KafkaRequestDispatcher(registry, new KafkaErrorMapper()); } From a52a81f4d03be36c83a5be73e4c4bc4260590904 Mon Sep 17 00:00:00 2001 From: Yang Guo Date: Tue, 13 Oct 2026 17:02:25 +0800 Subject: [PATCH 6/6] [test] Trim Kafka framework tests and cover version fallback Consolidate registry coverage and retain only disabled and real connection idle-timeout scenarios. Add decoder coverage for unsupported ApiVersions requests, including the version-zero error response and buffer release. Validation: 27 Kafka cases passed on Java 11; Checkstyle, Spotless and diff checks passed. The decoder regression test fails with the old parser. Co-Authored-By: Codex AI-Model: gpt-6 AI-Contributed/Feature: 0/0 AI-Contributed/UT: 178/178 --- .../fluss/kafka/KafkaCommandDecoderTest.java | 54 ++++++++++++++++ .../fluss/kafka/KafkaProtocolPluginTest.java | 60 +++-------------- .../dispatcher/KafkaApiRegistryTest.java | 64 +++++-------------- 3 files changed, 77 insertions(+), 101 deletions(-) diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaCommandDecoderTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaCommandDecoderTest.java index a2b8a0dc60b..f4a255b013d 100644 --- a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaCommandDecoderTest.java +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaCommandDecoderTest.java @@ -17,6 +17,7 @@ package org.apache.fluss.kafka; +import org.apache.fluss.rpc.TestingTabletGatewayService; import org.apache.fluss.rpc.netty.server.RequestChannel; import org.apache.fluss.shaded.netty4.io.netty.buffer.ByteBuf; import org.apache.fluss.shaded.netty4.io.netty.buffer.Unpooled; @@ -27,6 +28,7 @@ import org.apache.kafka.common.message.ProduceRequestData; import org.apache.kafka.common.message.ProduceResponseData; import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.protocol.Errors; import org.apache.kafka.common.requests.AbstractRequest; import org.apache.kafka.common.requests.ApiVersionsRequest; import org.apache.kafka.common.requests.ApiVersionsResponse; @@ -44,6 +46,58 @@ /** Tests response ordering and ownership in {@link KafkaCommandDecoder}. */ public class KafkaCommandDecoderTest { + @Test + public void testUnsupportedApiVersionsRequestReturnsVersionZeroErrorResponse() { + RequestChannel requestChannel = new RequestChannel(1); + EmbeddedChannel channel = + new EmbeddedChannel( + new KafkaCommandDecoder( + new RequestChannel[] {requestChannel}, "KAFKA_EXTERNAL")); + short unsupportedVersion = (short) (ApiKeys.API_VERSIONS.latestVersion() + 1); + RequestHeader header = + new RequestHeader(ApiKeys.API_VERSIONS, unsupportedVersion, "client", 42); + // The unsupported request body must not be decoded. A version-zero body is empty. + ByteBuf requestBuffer = + serialize(header, new ApiVersionsRequest.Builder().build((short) 0)); + KafkaRequest request = null; + ByteBuf responseBuffer = null; + try { + channel.writeInbound(requestBuffer); + request = (KafkaRequest) requestChannel.pollRequest(1000); + assertThat(request).isNotNull(); + assertThat(request.apiVersion()).isEqualTo(unsupportedVersion); + assertThat(request.listenerName()).isEqualTo("KAFKA_EXTERNAL"); + ApiVersionsRequest parsedRequest = request.request(); + assertThat(parsedRequest.hasUnsupportedRequestVersion()).isTrue(); + assertThat(parsedRequest.version()).isZero(); + + new KafkaRequestHandler(new TestingTabletGatewayService()).processRequest(request); + channel.runPendingTasks(); + + responseBuffer = channel.readOutbound(); + assertThat(responseBuffer).isNotNull(); + ByteBuffer responseBytes = responseBuffer.nioBuffer(); + assertThat(ResponseHeader.parse(responseBytes, (short) 0).correlationId()) + .isEqualTo(42); + ApiVersionsResponse response = ApiVersionsResponse.parse(responseBytes, (short) 0); + assertThat(response.data().errorCode()).isEqualTo(Errors.UNSUPPORTED_VERSION.code()); + assertThat(responseBytes.hasRemaining()).isFalse(); + assertThat(requestBuffer.refCnt()).isZero(); + assertThat(channel.isActive()).isTrue(); + } finally { + if (responseBuffer != null) { + responseBuffer.release(); + } + if (request != null) { + request.releaseBuffer(); + } + channel.finishAndReleaseAll(); + if (requestBuffer.refCnt() > 0) { + requestBuffer.release(); + } + } + } + @Test public void testAcksZeroSuppressesResponseAndUnblocksFollowingResponse() { RequestChannel requestChannel = new RequestChannel(100); diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaProtocolPluginTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaProtocolPluginTest.java index 3d0ded2756b..021bc67068b 100644 --- a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaProtocolPluginTest.java +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaProtocolPluginTest.java @@ -22,12 +22,9 @@ import org.apache.fluss.rpc.netty.server.RequestChannel; import org.apache.fluss.shaded.netty4.io.netty.channel.embedded.EmbeddedChannel; import org.apache.fluss.shaded.netty4.io.netty.channel.socket.SocketChannel; -import org.apache.fluss.shaded.netty4.io.netty.handler.timeout.IdleStateEvent; import org.apache.fluss.shaded.netty4.io.netty.handler.timeout.IdleStateHandler; import org.junit.jupiter.api.Test; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.ValueSource; import java.time.Duration; @@ -35,56 +32,13 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; -/** Tests Kafka listener idle-time configuration and idle-event handling. */ +/** Tests disabling the Kafka listener's idle timeout. */ class KafkaProtocolPluginTest { @Test - void testDefaultNettyIdleTimeout() throws Exception { - EmbeddedChannel channel = createChannel(new Configuration()); - try { - IdleStateHandler idleHandler = channel.pipeline().get(IdleStateHandler.class); - assertThat(idleHandler.getAllIdleTimeInMillis()) - .isEqualTo(Duration.ofMinutes(10).toMillis()); - assertThat(idleHandler.getReaderIdleTimeInMillis()).isZero(); - assertThat(idleHandler.getWriterIdleTimeInMillis()).isZero(); - } finally { - channel.finishAndReleaseAll(); - } - } - - @ParameterizedTest - @ValueSource(longs = {0, 7}) - void testConfiguredNettyIdleTimeout(long timeoutSeconds) throws Exception { + void testZeroDisablesNettyIdleTimeout() throws Exception { Configuration conf = new Configuration(); - conf.set(ConfigOptions.NETTY_CONNECTION_MAX_IDLE_TIME, Duration.ofSeconds(timeoutSeconds)); - EmbeddedChannel channel = createChannel(conf); - try { - IdleStateHandler idleHandler = channel.pipeline().get(IdleStateHandler.class); - assertThat(idleHandler.getAllIdleTimeInMillis()) - .isEqualTo(Duration.ofSeconds(timeoutSeconds).toMillis()); - assertThat(idleHandler.getReaderIdleTimeInMillis()).isZero(); - assertThat(idleHandler.getWriterIdleTimeInMillis()).isZero(); - } finally { - channel.finishAndReleaseAll(); - } - } - - @Test - void testOnlyAllIdleClosesConnection() throws Exception { - EmbeddedChannel channel = createChannel(new Configuration()); - try { - channel.pipeline().fireUserEventTriggered(IdleStateEvent.FIRST_READER_IDLE_STATE_EVENT); - assertThat(channel.isActive()).isTrue(); - channel.pipeline().fireUserEventTriggered(IdleStateEvent.FIRST_WRITER_IDLE_STATE_EVENT); - assertThat(channel.isActive()).isTrue(); - channel.pipeline().fireUserEventTriggered(IdleStateEvent.FIRST_ALL_IDLE_STATE_EVENT); - assertThat(channel.isActive()).isFalse(); - } finally { - channel.finishAndReleaseAll(); - } - } - - private static EmbeddedChannel createChannel(Configuration conf) throws Exception { + conf.set(ConfigOptions.NETTY_CONNECTION_MAX_IDLE_TIME, Duration.ZERO); EmbeddedChannel channel = new EmbeddedChannel(); try { // Adapt the socket type while installing the real pipeline on the embedded event loop. @@ -97,10 +51,12 @@ private static EmbeddedChannel createChannel(Configuration conf) throws Exceptio plugin.createChannelHandler( new RequestChannel[] {new RequestChannel(1)}, "KAFKA"); initializer.initChannel(socket); - return channel; - } catch (Exception | Error failure) { + IdleStateHandler idleHandler = channel.pipeline().get(IdleStateHandler.class); + assertThat(idleHandler.getAllIdleTimeInMillis()).isZero(); + assertThat(idleHandler.getReaderIdleTimeInMillis()).isZero(); + assertThat(idleHandler.getWriterIdleTimeInMillis()).isZero(); + } finally { channel.finishAndReleaseAll(); - throw failure; } } } diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaApiRegistryTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaApiRegistryTest.java index 4cf00860cc9..5c49a89e7ab 100644 --- a/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaApiRegistryTest.java +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/dispatcher/KafkaApiRegistryTest.java @@ -26,7 +26,6 @@ import java.util.ArrayList; import java.util.Arrays; -import java.util.Collections; import java.util.List; import java.util.concurrent.CompletableFuture; @@ -49,14 +48,6 @@ public void testRejectDuplicateApiKeys() { .hasMessageContaining("already registered"); } - @Test - public void testEmptyRegistry() { - KafkaApiRegistry registry = new KafkaApiRegistry(Collections.emptyList()); - - assertThat(registry.lookup(ApiKeys.API_VERSIONS)).isNull(); - assertThat(registry.advertisedApiSpecs()).isEmpty(); - } - @Test public void testChangesToInputCollectionDoNotAffectRegistry() { TestingApiHandler handler = new TestingApiHandler(ApiKeys.API_VERSIONS, true); @@ -73,52 +64,27 @@ public void testChangesToInputCollectionDoNotAffectRegistry() { } @Test - public void testAdvertisedSpecsAreSortedAndUnmodifiable() { + public void testOnlyEnabledApisAreAdvertisedAndRoutable() { TestingApiHandler apiVersions = new TestingApiHandler(ApiKeys.API_VERSIONS, true); TestingApiHandler metadata = new TestingApiHandler(ApiKeys.METADATA, true); - KafkaApiRegistry registry = new KafkaApiRegistry(Arrays.asList(apiVersions, metadata)); - - List specs = registry.advertisedApiSpecs(); - - assertThat(specs).containsExactly(metadata.apiSpec(), apiVersions.apiSpec()); - assertThatThrownBy(specs::clear).isInstanceOf(UnsupportedOperationException.class); - assertThat(registry.advertisedApiSpecs()) - .containsExactly(metadata.apiSpec(), apiVersions.apiSpec()); - } - - @Test - public void testOnlyAdvertiseEnabledHandlers() { + TestingApiHandler hidden = new TestingApiHandler(ApiKeys.PRODUCE, false); KafkaApiRegistry registry = - new KafkaApiRegistry( - Collections.singletonList( - new TestingApiHandler(ApiKeys.API_VERSIONS, true))); - assertThat(registry.advertisedApiSpecs()).hasSize(1); - - KafkaApiRegistry hiddenRegistry = - new KafkaApiRegistry( - Collections.singletonList( - new TestingApiHandler(ApiKeys.API_VERSIONS, false))); - assertThat(hiddenRegistry.advertisedApiSpecs()).isEmpty(); - assertThat(hiddenRegistry.lookup(ApiKeys.API_VERSIONS)).isNull(); - } - - @Test - public void testAdvertisedSpecIsSameSpecUsedForRouting() { - TestingApiHandler handler = new TestingApiHandler(ApiKeys.API_VERSIONS, true); - KafkaApiRegistry registry = new KafkaApiRegistry(Collections.singletonList(handler)); + new KafkaApiRegistry(Arrays.asList(apiVersions, hidden, metadata)); - KafkaApiSpec advertisedSpec = registry.advertisedApiSpecs().get(0); - KafkaApiHandler routedHandler = registry.lookup(ApiKeys.API_VERSIONS); + List specs = registry.advertisedApiSpecs(); - assertThat(routedHandler).isSameAs(handler); - assertThat(routedHandler.apiSpec()).isSameAs(advertisedSpec); - for (short version : ApiKeys.API_VERSIONS.allVersions()) { - assertThat(advertisedSpec.supportsVersion(version)).isTrue(); + assertThat(specs) + .extracting(KafkaApiSpec::apiKey) + .containsExactly(ApiKeys.METADATA, ApiKeys.API_VERSIONS); + assertThat(registry.lookup(ApiKeys.PRODUCE)).isNull(); + for (KafkaApiSpec spec : specs) { + KafkaApiHandler handler = registry.lookup(spec.apiKey()); + assertThat(handler).isNotNull(); + assertThat(handler.apiSpec().apiKey()).isEqualTo(spec.apiKey()); + assertThat(spec.minVersion()).isEqualTo(handler.apiSpec().minVersion()); + assertThat(spec.maxVersion()).isEqualTo(handler.apiSpec().maxVersion()); } - assertThat( - advertisedSpec.supportsVersion( - (short) (ApiKeys.API_VERSIONS.latestVersion() + 1))) - .isFalse(); + assertThatThrownBy(specs::clear).isInstanceOf(UnsupportedOperationException.class); } @Test