From 49b95cef3e0e876cff3e6ef9683d0a55bfcaa25a Mon Sep 17 00:00:00 2001 From: Yang Guo Date: Tue, 8 Sep 2026 11:29:44 +0800 Subject: [PATCH 1/2] [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 0ea02c70ac875b97728f910e2135e2bb6a53a8b5 Mon Sep 17 00:00:00 2001 From: Yang Guo Date: Tue, 8 Sep 2026 11:31:07 +0800 Subject: [PATCH 2/2] [kafka] Serve ApiVersions from registered capabilities Route requests through the dispatcher and advertise only implemented APIs. Return version-aware errors for unsupported APIs and invalid requests. Validated with mvn -o -pl fluss-kafka verify (23 unit tests and 1 IT). Co-Authored-By: Codex AI-Model: gpt-6 AI-Contributed/Feature: 313/313 AI-Contributed/UT: 137/137 --- .../fluss/kafka/KafkaProtocolPlugin.java | 3 +- .../fluss/kafka/KafkaRequestHandler.java | 232 ++---------------- .../api/versions/ApiVersionsHandler.java | 78 ++++++ .../fluss/kafka/KafkaRequestHandlerTest.java | 137 ++++++++--- 4 files changed, 207 insertions(+), 243 deletions(-) create mode 100644 fluss-kafka/src/main/java/org/apache/fluss/kafka/api/versions/ApiVersionsHandler.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..939c35a2d95 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 @@ -66,7 +66,6 @@ public RequestHandler createRequestHandler(RpcGatewayService service) { "Kafka protocol endpoints can only be enabled on TabletServers, but the service is " + service.getClass().getSimpleName()); } - TabletServerGateway gateway = (TabletServerGateway) service; - return new KafkaRequestHandler(gateway); + return new KafkaRequestHandler(); } } diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaRequestHandler.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaRequestHandler.java index 73555093ff0..2df16a8bd7f 100644 --- a/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaRequestHandler.java +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/KafkaRequestHandler.java @@ -17,27 +17,24 @@ package org.apache.fluss.kafka; -import org.apache.fluss.rpc.gateway.TabletServerGateway; +import org.apache.fluss.kafka.api.versions.ApiVersionsHandler; +import org.apache.fluss.kafka.dispatcher.KafkaApiRegistry; +import org.apache.fluss.kafka.dispatcher.KafkaRequestDispatcher; +import org.apache.fluss.kafka.error.KafkaErrorMapper; import org.apache.fluss.rpc.netty.server.RequestHandler; import org.apache.fluss.rpc.protocol.RequestType; -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.record.RecordBatch; -import org.apache.kafka.common.requests.AbstractRequest; -import org.apache.kafka.common.requests.AbstractResponse; -import org.apache.kafka.common.requests.ApiVersionsResponse; - -/** Kafka protocol implementation for request handler. */ +/** Entry point that dispatches Kafka protocol requests to registered API handlers. */ public class KafkaRequestHandler implements RequestHandler { - // TODO: we may need a new abstraction between TabletService and ReplicaManager to avoid - // affecting Fluss protocol when supporting compatibility with Kafka. - private final TabletServerGateway gateway; + private final KafkaRequestDispatcher dispatcher; - public KafkaRequestHandler(TabletServerGateway gateway) { - this.gateway = gateway; + /** Creates a Kafka request handler with the implemented server capabilities. */ + public KafkaRequestHandler() { + KafkaApiRegistry registry = new KafkaApiRegistry(); + registry.register(new ApiVersionsHandler(registry)); + registry.freeze(); + this.dispatcher = new KafkaRequestDispatcher(registry, new KafkaErrorMapper()); } @Override @@ -47,200 +44,15 @@ public RequestType requestType() { @Override public void processRequest(KafkaRequest request) { - // See kafka.server.KafkaApis#handle - switch (request.apiKey()) { - case API_VERSIONS: - handleApiVersionsRequest(request); - break; - case METADATA: - handleMetadataRequest(request); - break; - case PRODUCE: - handleProducerRequest(request); - break; - case FIND_COORDINATOR: - handleFindCoordinatorRequest(request); - break; - case LIST_OFFSETS: - handleListOffsetRequest(request); - break; - case OFFSET_FETCH: - handleOffsetFetchRequest(request); - break; - case OFFSET_COMMIT: - handleOffsetCommitRequest(request); - break; - case FETCH: - handleFetchRequest(request); - break; - case JOIN_GROUP: - handleJoinGroupRequest(request); - break; - case SYNC_GROUP: - handleSyncGroupRequest(request); - break; - case HEARTBEAT: - handleHeartbeatRequest(request); - break; - case LEAVE_GROUP: - handleLeaveGroupRequest(request); - break; - case DESCRIBE_GROUPS: - handleDescribeGroupsRequest(request); - break; - case LIST_GROUPS: - handleListGroupsRequest(request); - break; - case DELETE_GROUPS: - handleDeleteGroupsRequest(request); - break; - case SASL_HANDSHAKE: - handleSaslHandshakeRequest(request); - break; - case SASL_AUTHENTICATE: - handleSaslAuthenticateRequest(request); - break; - case CREATE_TOPICS: - handleCreateTopicsRequest(request); - break; - case INIT_PRODUCER_ID: - handleInitProducerIdRequest(request); - break; - case ADD_PARTITIONS_TO_TXN: - handleAddPartitionsToTxnRequest(request); - break; - case ADD_OFFSETS_TO_TXN: - handleAddOffsetsToTxnRequest(request); - break; - case TXN_OFFSET_COMMIT: - handleTxnOffsetCommitRequest(request); - break; - case END_TXN: - handleEndTxnRequest(request); - break; - case WRITE_TXN_MARKERS: - handleWriteTxnMarkersRequest(request); - break; - case DESCRIBE_CONFIGS: - handleDescribeConfigsRequest(request); - break; - case ALTER_CONFIGS: - handleAlterConfigsRequest(request); - break; - case DELETE_TOPICS: - handleDeleteTopicsRequest(request); - break; - case DELETE_RECORDS: - handleDeleteRecordsRequest(request); - break; - case OFFSET_DELETE: - handleOffsetDeleteRequest(request); - break; - case CREATE_PARTITIONS: - handleCreatePartitionsRequest(request); - break; - case DESCRIBE_CLUSTER: - handleDescribeClusterRequest(request); - break; - default: - handleUnsupportedRequest(request); - } - } - - private void handleUnsupportedRequest(KafkaRequest request) { - String message = String.format("Unsupported request with api key %s", request.apiKey()); - AbstractRequest abstractRequest = request.request(); - AbstractResponse response = - abstractRequest.getErrorResponse(new UnsupportedOperationException(message)); - request.complete(response); - } - - void handleApiVersionsRequest(KafkaRequest request) { - short apiVersion = request.apiVersion(); - if (!ApiKeys.API_VERSIONS.isVersionSupported(apiVersion)) { - request.fail(Errors.UNSUPPORTED_VERSION.exception()); - return; - } - ApiVersionsResponseData data = new ApiVersionsResponseData(); - for (ApiKeys apiKey : ApiKeys.values()) { - if (apiKey.minRequiredInterBrokerMagic <= RecordBatch.CURRENT_MAGIC_VALUE) { - ApiVersionsResponseData.ApiVersion apiVersionData = - new ApiVersionsResponseData.ApiVersion() - .setApiKey(apiKey.id) - .setMinVersion(apiKey.oldestVersion()) - .setMaxVersion(apiKey.latestVersion()); - if (apiKey.equals(ApiKeys.METADATA)) { - // Not support TopicId - short v = apiKey.latestVersion() > 11 ? 11 : apiKey.latestVersion(); - apiVersionData.setMaxVersion(v); - } else if (apiKey.equals(ApiKeys.FETCH)) { - // Not support TopicId - short v = apiKey.latestVersion() > 12 ? 12 : apiKey.latestVersion(); - apiVersionData.setMaxVersion(v); - } - data.apiKeys().add(apiVersionData); - } - } - request.complete(new ApiVersionsResponse(data)); + dispatcher + .dispatch(request) + .whenComplete( + (response, failure) -> { + if (failure == null) { + request.complete(response); + } else { + request.fail(failure); + } + }); } - - void handleProducerRequest(KafkaRequest request) {} - - void handleMetadataRequest(KafkaRequest request) {} - - void handleFindCoordinatorRequest(KafkaRequest request) {} - - void handleListOffsetRequest(KafkaRequest request) {} - - void handleOffsetFetchRequest(KafkaRequest request) {} - - void handleOffsetCommitRequest(KafkaRequest request) {} - - void handleFetchRequest(KafkaRequest request) {} - - void handleJoinGroupRequest(KafkaRequest request) {} - - void handleSyncGroupRequest(KafkaRequest request) {} - - void handleHeartbeatRequest(KafkaRequest request) {} - - void handleLeaveGroupRequest(KafkaRequest request) {} - - void handleDescribeGroupsRequest(KafkaRequest request) {} - - void handleListGroupsRequest(KafkaRequest request) {} - - void handleDeleteGroupsRequest(KafkaRequest request) {} - - void handleSaslHandshakeRequest(KafkaRequest request) {} - - void handleSaslAuthenticateRequest(KafkaRequest request) {} - - void handleCreateTopicsRequest(KafkaRequest request) {} - - void handleInitProducerIdRequest(KafkaRequest request) {} - - void handleAddPartitionsToTxnRequest(KafkaRequest request) {} - - void handleAddOffsetsToTxnRequest(KafkaRequest request) {} - - void handleTxnOffsetCommitRequest(KafkaRequest request) {} - - void handleEndTxnRequest(KafkaRequest request) {} - - void handleWriteTxnMarkersRequest(KafkaRequest request) {} - - void handleDescribeConfigsRequest(KafkaRequest request) {} - - void handleAlterConfigsRequest(KafkaRequest request) {} - - void handleDeleteTopicsRequest(KafkaRequest request) {} - - void handleDeleteRecordsRequest(KafkaRequest request) {} - - void handleOffsetDeleteRequest(KafkaRequest request) {} - - void handleCreatePartitionsRequest(KafkaRequest request) {} - - void handleDescribeClusterRequest(KafkaRequest request) {} } diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/api/versions/ApiVersionsHandler.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/api/versions/ApiVersionsHandler.java new file mode 100644 index 00000000000..c38d7bc6cb2 --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/api/versions/ApiVersionsHandler.java @@ -0,0 +1,78 @@ +/* + * 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.api.versions; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.kafka.KafkaRequestContext; +import org.apache.fluss.kafka.dispatcher.KafkaApiHandler; +import org.apache.fluss.kafka.dispatcher.KafkaApiRegistry; +import org.apache.fluss.kafka.dispatcher.KafkaApiSpec; + +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 java.util.concurrent.CompletableFuture; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** Implements ApiVersions from the capabilities actually registered on this server. */ +@Internal +public final class ApiVersionsHandler implements KafkaApiHandler { + + private static final KafkaApiSpec API_SPEC = + new KafkaApiSpec( + ApiKeys.API_VERSIONS, + ApiKeys.API_VERSIONS.oldestVersion(), + ApiKeys.API_VERSIONS.latestVersion(), + true); + + private final KafkaApiRegistry registry; + + /** Creates an ApiVersions handler backed by the server capability registry. */ + public ApiVersionsHandler(KafkaApiRegistry registry) { + this.registry = checkNotNull(registry); + } + + @Override + public KafkaApiSpec apiSpec() { + return API_SPEC; + } + + @Override + public CompletableFuture handle( + KafkaRequestContext context, ApiVersionsRequest request) { + if (!request.isValid()) { + return CompletableFuture.completedFuture( + request.getErrorResponse(Errors.INVALID_REQUEST.exception())); + } + ApiVersionsResponseData data = new ApiVersionsResponseData(); + for (KafkaApiSpec spec : registry.advertisedApiSpecs()) { + data.apiKeys() + .add( + new ApiVersionsResponseData.ApiVersion() + .setApiKey(spec.apiKey().id) + .setMinVersion(spec.minVersion()) + .setMaxVersion(spec.maxVersion())); + } + return CompletableFuture.completedFuture(new ApiVersionsResponse(data)); + } +} 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..e6b8e961128 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 @@ -17,24 +17,32 @@ package org.apache.fluss.kafka; -import org.apache.fluss.rpc.TestingTabletGatewayService; import org.apache.fluss.shaded.netty4.io.netty.buffer.ByteBuf; import org.apache.fluss.shaded.netty4.io.netty.buffer.ByteBufAllocator; import org.apache.fluss.shaded.netty4.io.netty.channel.ChannelHandlerContext; +import org.apache.kafka.common.message.ApiVersionsRequestData; +import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersion; +import org.apache.kafka.common.message.CreateTopicsRequestData; 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; +import org.apache.kafka.common.requests.CreateTopicsRequest; +import org.apache.kafka.common.requests.CreateTopicsResponse; 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.Collections; import java.util.Map; import java.util.concurrent.CompletableFuture; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.tuple; /** Tests for {@link KafkaRequestHandler}. */ public class KafkaRequestHandlerTest { @@ -53,7 +61,7 @@ public void testKafkaApiVersionsNotSupported() { new RequestHeader(ApiKeys.API_VERSIONS, latestVersion, "client-id", 0), apiVersionsRequest, ctx); - handler.handleApiVersionsRequest(request); + handler.processRequest(request); ApiVersionsResponse response = (ApiVersionsResponse) parseResponse(request); Map errorCounts = response.errorCounts(); @@ -61,44 +69,111 @@ public void testKafkaApiVersionsNotSupported() { assertThat(1).isEqualTo(errorCounts.get(Errors.UNSUPPORTED_VERSION)); } + @ParameterizedTest + @ValueSource(shorts = {0, 1, 2, 3, 4}) + public void testKafkaApiVersionsRequest(short version) { + KafkaRequestHandler handler = createKafkaRequestHandler(); + ApiVersionsResponse response = requestApiVersions(handler, version); + + assertSuccessfulResponseDefaults(response); + assertBrokerCapabilities(response); + } + + private static ApiVersionsResponse requestApiVersions( + KafkaRequestHandler handler, short version) { + ApiVersionsRequest apiVersionsRequest = new ApiVersionsRequest.Builder().build(version); + ChannelHandlerContext ctx = new TestingChannelHandlerContext(); + KafkaRequest request = + newRequest( + ApiKeys.API_VERSIONS, + version, + new RequestHeader(ApiKeys.API_VERSIONS, version, "client-id", 0), + apiVersionsRequest, + ctx); + handler.processRequest(request); + + return parseApiVersionsResponse(request); + } + + private static ApiVersionsResponse parseApiVersionsResponse(KafkaRequest request) { + return (ApiVersionsResponse) parseResponse(request); + } + + private static void assertSuccessfulResponseDefaults(ApiVersionsResponse response) { + assertThat(response.errorCounts()) + .containsExactlyEntriesOf(Collections.singletonMap(Errors.NONE, 1)); + assertThat(response.data().throttleTimeMs()).isZero(); + assertThat(response.data().supportedFeatures()).isEmpty(); + assertThat(response.data().finalizedFeaturesEpoch()).isEqualTo(-1L); + assertThat(response.data().finalizedFeatures()).isEmpty(); + assertThat(response.data().zkMigrationReady()).isFalse(); + } + + private static void assertBrokerCapabilities(ApiVersionsResponse response) { + assertThat(response.data().apiKeys()) + .extracting(ApiVersion::apiKey, ApiVersion::minVersion, ApiVersion::maxVersion) + .containsExactly( + tuple( + ApiKeys.API_VERSIONS.id, + ApiKeys.API_VERSIONS.oldestVersion(), + ApiKeys.API_VERSIONS.latestVersion())); + } + @Test - public void testKafkaApiVersionsRequest() { + public void testInvalidApiVersionsRequest() { KafkaRequestHandler handler = createKafkaRequestHandler(); short latestVersion = ApiKeys.API_VERSIONS.latestVersion(); - ApiVersionsRequest apiVersionsRequest = - new ApiVersionsRequest.Builder().build(latestVersion); - ChannelHandlerContext ctx = new TestingChannelHandlerContext(); + ApiVersionsRequest requestBody = + new ApiVersionsRequest.Builder( + new ApiVersionsRequestData() + .setClientSoftwareName("invalid client name") + .setClientSoftwareVersion("1.0"), + latestVersion, + latestVersion) + .build(latestVersion); KafkaRequest request = newRequest( ApiKeys.API_VERSIONS, latestVersion, new RequestHeader(ApiKeys.API_VERSIONS, latestVersion, "client-id", 0), - apiVersionsRequest, - ctx); - handler.handleApiVersionsRequest(request); + requestBody, + new TestingChannelHandlerContext()); + + handler.processRequest(request); ApiVersionsResponse response = (ApiVersionsResponse) parseResponse(request); - Map errorCounts = response.errorCounts(); - assertThat(1).isEqualTo(errorCounts.size()); - assertThat(1).isEqualTo(errorCounts.get(Errors.NONE)); - response.data() - .apiKeys() - .forEach( - apiVersion -> { - if (ApiKeys.METADATA.id == apiVersion.apiKey()) { - assertThat((short) 11) - .isGreaterThanOrEqualTo(apiVersion.maxVersion()); - } else if (ApiKeys.FETCH.id == apiVersion.apiKey()) { - assertThat((short) 12) - .isGreaterThanOrEqualTo(apiVersion.maxVersion()); - } else { - ApiKeys apiKeys = ApiKeys.forId(apiVersion.apiKey()); - assertThat(apiVersion.minVersion()) - .isEqualTo(apiKeys.oldestVersion()); - assertThat(apiVersion.maxVersion()) - .isEqualTo(apiKeys.latestVersion()); - } - }); + assertThat(response.errorCounts()).containsEntry(Errors.INVALID_REQUEST, 1); + } + + @Test + public void testUnregisteredApiIsNotRouted() { + KafkaRequestHandler handler = createKafkaRequestHandler(); + short version = ApiKeys.CREATE_TOPICS.latestVersion(); + CreateTopicsRequestData requestData = + new CreateTopicsRequestData() + .setTimeoutMs(1000) + .setTopics( + new CreateTopicsRequestData.CreatableTopicCollection( + Collections.singletonList( + new CreateTopicsRequestData.CreatableTopic() + .setName("topic") + .setNumPartitions(1) + .setReplicationFactor((short) 1)) + .iterator())); + CreateTopicsRequest requestBody = + new CreateTopicsRequest.Builder(requestData).build(version); + KafkaRequest request = + newRequest( + ApiKeys.CREATE_TOPICS, + version, + new RequestHeader(ApiKeys.CREATE_TOPICS, version, "client-id", 0), + requestBody, + new TestingChannelHandlerContext()); + + handler.processRequest(request); + + CreateTopicsResponse response = (CreateTopicsResponse) parseResponse(request); + assertThat(response.errorCounts()).containsEntry(Errors.UNSUPPORTED_VERSION, 1); } private static KafkaRequest newRequest( @@ -133,6 +208,6 @@ private static AbstractResponse parseResponse(KafkaRequest request) { } private static KafkaRequestHandler createKafkaRequestHandler() { - return new KafkaRequestHandler(new TestingTabletGatewayService()); + return new KafkaRequestHandler(); } }