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..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 @@ -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)); @@ -65,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/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/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/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..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,23 +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 { @@ -46,73 +55,159 @@ 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<>()); - handler.handleApiVersionsRequest(request); + ctx); + handler.processRequest(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)); } + @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 = - new KafkaRequest( + newRequest( ApiKeys.API_VERSIONS, latestVersion, new RequestHeader(ApiKeys.API_VERSIONS, latestVersion, "client-id", 0), - apiVersionsRequest, - ByteBufAllocator.DEFAULT.buffer(), - ctx, - new CompletableFuture<>()); - handler.handleApiVersionsRequest(request); + requestBody, + new TestingChannelHandlerContext()); + + handler.processRequest(request); + + ApiVersionsResponse response = (ApiVersionsResponse) parseResponse(request); + 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( + 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(); - ApiVersionsResponse response = - (ApiVersionsResponse) - AbstractResponse.parseResponse( - responseBuffer.nioBuffer(), request.header()); - 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()); - } - }); + try { + return AbstractResponse.parseResponse(responseBuffer.nioBuffer(), request.header()); + } finally { + responseBuffer.release(); + } } private static KafkaRequestHandler createKafkaRequestHandler() { - return new KafkaRequestHandler(new TestingTabletGatewayService()); + return new KafkaRequestHandler(); } } 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