diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java index 3fdb77970af..8c3a727bb3f 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java @@ -2645,6 +2645,22 @@ public class ConfigOptions { .withDescription( "The database for fluss kafka. The default database is `kafka`."); + public static final ConfigOption KAFKA_DEFAULT_KEY_FORMAT = + key("kafka.default.key.format") + .stringType() + .defaultValue("raw") + .withDescription( + "The default format for Kafka record keys when a CreateTopics request does not specify fluss.key.format. " + + "Supported formats are raw and string."); + + public static final ConfigOption KAFKA_DEFAULT_VALUE_FORMAT = + key("kafka.default.value.format") + .stringType() + .defaultValue("raw") + .withDescription( + "The default format for Kafka record values when a CreateTopics request does not specify fluss.value.format. " + + "Supported formats are raw and string."); + public static final ConfigOption KAFKA_CONNECTION_MAX_IDLE_TIME = key("kafka.connection.max-idle-time") .durationType() diff --git a/fluss-kafka/pom.xml b/fluss-kafka/pom.xml index 124001ca7b5..042a720cb02 100644 --- a/fluss-kafka/pom.xml +++ b/fluss-kafka/pom.xml @@ -64,11 +64,25 @@ + + org.apache.curator + curator-test + ${curator.version} + test + + org.apache.fluss fluss-test-utils + + org.apache.fluss + fluss-client + ${project.version} + test + + org.apache.fluss fluss-common @@ -92,4 +106,4 @@ test - \ No newline at end of file + 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..1dcf1cca90a 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 because Kafka record sets can reference its memory + // asynchronously. Release the decoder's ownership on every path; the request releases + // its retained reference after response handling or cancellation. + ReferenceCountUtil.release(buffer); } } @@ -184,19 +185,39 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E } private static KafkaRequest parseRequest( - ChannelHandlerContext ctx, CompletableFuture 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..65d2f8d7af6 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 @@ -19,7 +19,9 @@ import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; +import org.apache.fluss.kafka.format.KafkaDataFormat; import org.apache.fluss.rpc.RpcGatewayService; +import org.apache.fluss.rpc.gateway.AdminGatewayProvider; import org.apache.fluss.rpc.gateway.TabletServerGateway; import org.apache.fluss.rpc.netty.server.RequestChannel; import org.apache.fluss.rpc.netty.server.RequestHandler; @@ -53,6 +55,7 @@ public ChannelHandler createChannelHandler( RequestChannel[] requestChannels, String listenerName) { return new KafkaChannelInitializer( requestChannels, + listenerName, conf.get(ConfigOptions.KAFKA_CONNECTION_MAX_IDLE_TIME).getSeconds(), (int) conf.get(ConfigOptions.NETTY_SERVER_MAX_REQUEST_SIZE).getBytes(), conf.getBoolean(ConfigOptions.NETTY_CLIENT_ALLOCATOR_HEAP_BUFFER_FIRST)); @@ -66,6 +69,15 @@ public RequestHandler createRequestHandler(RpcGatewayService service) { + service.getClass().getSimpleName()); } TabletServerGateway gateway = (TabletServerGateway) service; - return new KafkaRequestHandler(gateway); + if (service instanceof AdminGatewayProvider) { + return new KafkaRequestHandler( + service, + gateway, + ((AdminGatewayProvider) service).getAdminGateway(), + conf.get(ConfigOptions.KAFKA_DATABASE), + KafkaDataFormat.parse(conf.get(ConfigOptions.KAFKA_DEFAULT_KEY_FORMAT)), + KafkaDataFormat.parse(conf.get(ConfigOptions.KAFKA_DEFAULT_VALUE_FORMAT))); + } + return new KafkaRequestHandler(service, gateway, conf.get(ConfigOptions.KAFKA_DATABASE)); } } 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..20d8bf01898 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 @@ -46,6 +46,7 @@ public class KafkaRequest implements RpcRequest { private final long requestId = ID_GENERATOR.getAndIncrement(); private final RequestHeader header; private final AbstractRequest request; + private final String listenerName; private final ByteBuf buffer; private final ChannelHandlerContext ctx; private final long startTimeMs; @@ -60,10 +61,23 @@ protected KafkaRequest( ByteBuf buffer, ChannelHandlerContext ctx, CompletableFuture 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(); @@ -100,6 +114,10 @@ public T request() { return (T) request; } + public String listenerName() { + return listenerName; + } + public ChannelHandlerContext ctx() { return ctx; } 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..bbfa4fb30bc 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,87 @@ package org.apache.fluss.kafka; +import org.apache.fluss.kafka.api.admin.CreateTopicsHandler; +import org.apache.fluss.kafka.api.admin.DeleteTopicsHandler; +import org.apache.fluss.kafka.api.metadata.MetadataHandler; +import org.apache.fluss.kafka.api.versions.ApiVersionsHandler; +import org.apache.fluss.kafka.backend.admin.GatewayKafkaTopicAdminBackend; +import org.apache.fluss.kafka.backend.metadata.GatewayKafkaMetadataBackend; +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.kafka.format.KafkaDataFormat; +import org.apache.fluss.rpc.RpcGatewayService; +import org.apache.fluss.rpc.gateway.AdminGateway; import org.apache.fluss.rpc.gateway.TabletServerGateway; 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; +import static org.apache.fluss.utils.Preconditions.checkNotNull; -/** 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; + + /** Creates a Kafka request handler with the capabilities provided by a TabletServer. */ + public KafkaRequestHandler( + RpcGatewayService service, TabletServerGateway gateway, String kafkaDatabase) { + checkNotNull(service); + checkNotNull(gateway); + checkNotNull(kafkaDatabase); + KafkaApiRegistry registry = new KafkaApiRegistry(); + registry.register(new ApiVersionsHandler(registry)); + registry.register( + new MetadataHandler( + new GatewayKafkaMetadataBackend(service, gateway, kafkaDatabase))); + registry.freeze(); + this.dispatcher = new KafkaRequestDispatcher(registry, new KafkaErrorMapper()); + } + + /** Creates a Kafka request handler including topic lifecycle capabilities. */ + public KafkaRequestHandler( + RpcGatewayService service, + TabletServerGateway gateway, + AdminGateway adminGateway, + String kafkaDatabase) { + this( + service, + gateway, + adminGateway, + kafkaDatabase, + KafkaDataFormat.RAW, + KafkaDataFormat.RAW); + } - public KafkaRequestHandler(TabletServerGateway gateway) { - this.gateway = gateway; + /** + * Creates a Kafka request handler including topic lifecycle and default format capabilities. + */ + public KafkaRequestHandler( + RpcGatewayService service, + TabletServerGateway gateway, + AdminGateway adminGateway, + String kafkaDatabase, + KafkaDataFormat defaultKeyFormat, + KafkaDataFormat defaultValueFormat) { + checkNotNull(service); + checkNotNull(gateway); + checkNotNull(adminGateway); + checkNotNull(kafkaDatabase); + checkNotNull(defaultKeyFormat); + checkNotNull(defaultValueFormat); + KafkaApiRegistry registry = new KafkaApiRegistry(); + registry.register(new ApiVersionsHandler(registry)); + registry.register( + new MetadataHandler( + new GatewayKafkaMetadataBackend(service, gateway, kafkaDatabase), true)); + GatewayKafkaTopicAdminBackend topicAdminBackend = + new GatewayKafkaTopicAdminBackend(service, adminGateway, kafkaDatabase); + registry.register( + new CreateTopicsHandler(topicAdminBackend, defaultKeyFormat, defaultValueFormat)); + registry.register(new DeleteTopicsHandler(topicAdminBackend)); + registry.freeze(); + this.dispatcher = new KafkaRequestDispatcher(registry, new KafkaErrorMapper()); } @Override @@ -47,200 +107,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); + dispatcher + .dispatch(request) + .whenComplete( + (response, failure) -> { + if (failure == null) { + request.complete(response); + } else { + request.fail(failure); + } + }); } - - 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)); - } - - 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/admin/CreateTopicsHandler.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/api/admin/CreateTopicsHandler.java new file mode 100644 index 00000000000..095420e72ec --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/api/admin/CreateTopicsHandler.java @@ -0,0 +1,218 @@ +/* + * 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.admin; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.kafka.KafkaRequestContext; +import org.apache.fluss.kafka.backend.admin.KafkaTopicAdminBackend; +import org.apache.fluss.kafka.backend.admin.KafkaTopicAdminBackend.CreateTopic; +import org.apache.fluss.kafka.backend.admin.KafkaTopicAdminBackend.TopicResult; +import org.apache.fluss.kafka.dispatcher.KafkaApiHandler; +import org.apache.fluss.kafka.dispatcher.KafkaApiSpec; +import org.apache.fluss.kafka.format.KafkaDataFormat; + +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.internals.Topic; +import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic; +import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfig; +import org.apache.kafka.common.message.CreateTopicsResponseData; +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.CreateTopicsRequest; +import org.apache.kafka.common.requests.CreateTopicsResponse; + +import javax.annotation.Nullable; + +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** Implements Kafka CreateTopics by creating fixed-schema Arrow log tables in Fluss. */ +@Internal +public final class CreateTopicsHandler implements KafkaApiHandler { + + private static final KafkaApiSpec API_SPEC = + new KafkaApiSpec( + ApiKeys.CREATE_TOPICS, + ApiKeys.CREATE_TOPICS.oldestVersion(), + ApiKeys.CREATE_TOPICS.latestVersion(), + true); + + private final KafkaTopicAdminBackend backend; + private final KafkaDataFormat defaultKeyFormat; + private final KafkaDataFormat defaultValueFormat; + + /** Creates a CreateTopics handler. */ + public CreateTopicsHandler(KafkaTopicAdminBackend backend) { + this(backend, KafkaDataFormat.RAW, KafkaDataFormat.RAW); + } + + /** Creates a CreateTopics handler with formats used when a request omits format configs. */ + public CreateTopicsHandler( + KafkaTopicAdminBackend backend, + KafkaDataFormat defaultKeyFormat, + KafkaDataFormat defaultValueFormat) { + this.backend = checkNotNull(backend); + this.defaultKeyFormat = checkNotNull(defaultKeyFormat); + this.defaultValueFormat = checkNotNull(defaultValueFormat); + } + + @Override + public KafkaApiSpec apiSpec() { + return API_SPEC; + } + + @Override + public CompletableFuture handle( + KafkaRequestContext context, CreateTopicsRequest request) { + List validTopics = new ArrayList<>(); + Map localResults = new LinkedHashMap<>(); + for (CreatableTopic topic : request.data().topics()) { + TopicResult invalid = validate(topic); + if (invalid == null) { + try { + FormatConfig formats = parseFormats(topic); + validTopics.add( + new CreateTopic( + topic.name(), + topic.numPartitions(), + topic.replicationFactor(), + formats.keyFormat, + formats.valueFormat)); + } catch (IllegalArgumentException e) { + localResults.put( + topic.name(), invalid(topic, Errors.INVALID_CONFIG, e.getMessage())); + } + } else { + localResults.put(topic.name(), invalid); + } + } + + return backend.createTopics( + validTopics, + request.data().validateOnly(), + context.listenerName(), + clientAddress(context.remoteAddress())) + .thenApply(results -> toResponse(request, localResults, results)); + } + + private static @Nullable TopicResult validate(CreatableTopic topic) { + if (!Topic.isValid(topic.name())) { + return invalid(topic, Errors.INVALID_TOPIC_EXCEPTION, "Invalid Kafka topic name."); + } + if (topic.numPartitions() <= 0) { + return invalid( + topic, + Errors.INVALID_PARTITIONS, + "A positive partition count is required for a Fluss topic table."); + } + if (topic.replicationFactor() == 0 || topic.replicationFactor() < -1) { + return invalid(topic, Errors.INVALID_REPLICATION_FACTOR, "Invalid replication factor."); + } + if (!topic.assignments().isEmpty()) { + return invalid( + topic, + Errors.INVALID_REPLICA_ASSIGNMENT, + "Explicit Kafka replica assignments are not supported by Fluss."); + } + return null; + } + + private FormatConfig parseFormats(CreatableTopic topic) { + KafkaDataFormat keyFormat = defaultKeyFormat; + KafkaDataFormat valueFormat = defaultValueFormat; + Map configs = new LinkedHashMap<>(); + for (CreatableTopicConfig config : topic.configs()) { + if (configs.containsKey(config.name())) { + throw new IllegalArgumentException( + "Duplicate Kafka topic config '" + config.name() + "'."); + } + configs.put(config.name(), config.value()); + } + for (Map.Entry config : configs.entrySet()) { + if (KafkaDataFormat.KEY_FORMAT_CONFIG.equals(config.getKey())) { + keyFormat = KafkaDataFormat.parse(config.getValue()); + } else if (KafkaDataFormat.VALUE_FORMAT_CONFIG.equals(config.getKey())) { + valueFormat = KafkaDataFormat.parse(config.getValue()); + } else { + throw new IllegalArgumentException( + "Unsupported Kafka topic config '" + config.getKey() + "'."); + } + } + return new FormatConfig(keyFormat, valueFormat); + } + + private static TopicResult invalid(CreatableTopic topic, Errors error, String message) { + return new TopicResult( + topic.name(), + Uuid.ZERO_UUID, + error, + message, + topic.numPartitions(), + topic.replicationFactor()); + } + + private static CreateTopicsResponse toResponse( + CreateTopicsRequest request, + Map localResults, + List backendResults) { + Map results = new LinkedHashMap<>(localResults); + for (TopicResult result : backendResults) { + results.put(result.name(), result); + } + CreateTopicsResponseData response = new CreateTopicsResponseData().setThrottleTimeMs(0); + for (CreatableTopic topic : request.data().topics()) { + TopicResult result = results.get(topic.name()); + response.topics() + .add( + new CreateTopicsResponseData.CreatableTopicResult() + .setName(result.name()) + .setTopicId(result.topicId()) + .setErrorCode(result.error().code()) + .setErrorMessage(result.errorMessage()) + .setNumPartitions(result.numPartitions()) + .setReplicationFactor(result.replicationFactor())); + } + return new CreateTopicsResponse(response); + } + + private static @Nullable InetAddress clientAddress(SocketAddress remoteAddress) { + if (remoteAddress instanceof InetSocketAddress) { + return ((InetSocketAddress) remoteAddress).getAddress(); + } + return null; + } + + private static final class FormatConfig { + private final KafkaDataFormat keyFormat; + private final KafkaDataFormat valueFormat; + + private FormatConfig(KafkaDataFormat keyFormat, KafkaDataFormat valueFormat) { + this.keyFormat = keyFormat; + this.valueFormat = valueFormat; + } + } +} diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/api/admin/DeleteTopicsHandler.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/api/admin/DeleteTopicsHandler.java new file mode 100644 index 00000000000..6a2bf1d6eae --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/api/admin/DeleteTopicsHandler.java @@ -0,0 +1,109 @@ +/* + * 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.admin; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.kafka.KafkaRequestContext; +import org.apache.fluss.kafka.backend.admin.KafkaTopicAdminBackend; +import org.apache.fluss.kafka.backend.admin.KafkaTopicAdminBackend.DeleteTopic; +import org.apache.fluss.kafka.backend.admin.KafkaTopicAdminBackend.TopicResult; +import org.apache.fluss.kafka.dispatcher.KafkaApiHandler; +import org.apache.fluss.kafka.dispatcher.KafkaApiSpec; + +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.message.DeleteTopicsRequestData.DeleteTopicState; +import org.apache.kafka.common.message.DeleteTopicsResponseData; +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.requests.AbstractResponse; +import org.apache.kafka.common.requests.DeleteTopicsRequest; +import org.apache.kafka.common.requests.DeleteTopicsResponse; + +import javax.annotation.Nullable; + +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** Implements Kafka DeleteTopics by deleting the corresponding Fluss tables. */ +@Internal +public final class DeleteTopicsHandler implements KafkaApiHandler { + + private static final short TOPIC_ID_VERSION = 6; + private static final KafkaApiSpec API_SPEC = + new KafkaApiSpec( + ApiKeys.DELETE_TOPICS, + ApiKeys.DELETE_TOPICS.oldestVersion(), + ApiKeys.DELETE_TOPICS.latestVersion(), + true); + + private final KafkaTopicAdminBackend backend; + + /** Creates a DeleteTopics handler. */ + public DeleteTopicsHandler(KafkaTopicAdminBackend backend) { + this.backend = checkNotNull(backend); + } + + @Override + public KafkaApiSpec apiSpec() { + return API_SPEC; + } + + @Override + public CompletableFuture handle( + KafkaRequestContext context, DeleteTopicsRequest request) { + List topics = new ArrayList<>(); + if (request.version() < TOPIC_ID_VERSION) { + for (String topicName : request.data().topicNames()) { + topics.add(new DeleteTopic(topicName, Uuid.ZERO_UUID)); + } + } else { + for (DeleteTopicState topic : request.data().topics()) { + topics.add(new DeleteTopic(topic.name(), topic.topicId())); + } + } + return backend.deleteTopics( + topics, context.listenerName(), clientAddress(context.remoteAddress())) + .thenApply(DeleteTopicsHandler::toResponse); + } + + private static DeleteTopicsResponse toResponse(List results) { + DeleteTopicsResponseData response = new DeleteTopicsResponseData().setThrottleTimeMs(0); + for (TopicResult result : results) { + response.responses() + .add( + new DeleteTopicsResponseData.DeletableTopicResult() + .setName(result.name()) + .setTopicId(result.topicId()) + .setErrorCode(result.error().code()) + .setErrorMessage(result.errorMessage())); + } + return new DeleteTopicsResponse(response); + } + + private static @Nullable InetAddress clientAddress(SocketAddress remoteAddress) { + if (remoteAddress instanceof InetSocketAddress) { + return ((InetSocketAddress) remoteAddress).getAddress(); + } + return null; + } +} diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/api/metadata/MetadataHandler.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/api/metadata/MetadataHandler.java new file mode 100644 index 00000000000..bcfc2655d9c --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/api/metadata/MetadataHandler.java @@ -0,0 +1,198 @@ +/* + * 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.metadata; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.kafka.KafkaRequestContext; +import org.apache.fluss.kafka.backend.metadata.KafkaClusterMetadata; +import org.apache.fluss.kafka.backend.metadata.KafkaClusterMetadata.Broker; +import org.apache.fluss.kafka.backend.metadata.KafkaClusterMetadata.Partition; +import org.apache.fluss.kafka.backend.metadata.KafkaClusterMetadata.TopicError; +import org.apache.fluss.kafka.backend.metadata.KafkaMetadataBackend; +import org.apache.fluss.kafka.backend.metadata.KafkaMetadataQuery; +import org.apache.fluss.kafka.backend.metadata.KafkaMetadataQuery.TopicReference; +import org.apache.fluss.kafka.dispatcher.KafkaApiHandler; +import org.apache.fluss.kafka.dispatcher.KafkaApiSpec; + +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.internals.Topic; +import org.apache.kafka.common.message.MetadataRequestData.MetadataRequestTopic; +import org.apache.kafka.common.message.MetadataResponseData; +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.MetadataRequest; +import org.apache.kafka.common.requests.MetadataResponse; + +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CompletableFuture; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** Implements Kafka Metadata versions 0 through 11 using a narrow Fluss metadata backend. */ +@Internal +public final class MetadataHandler implements KafkaApiHandler { + + private static final short MAX_SUPPORTED_VERSION = 11; + private static final KafkaApiSpec API_SPEC = + new KafkaApiSpec( + ApiKeys.METADATA, + ApiKeys.METADATA.oldestVersion(), + (short) Math.min(ApiKeys.METADATA.latestVersion(), MAX_SUPPORTED_VERSION), + true); + + private final KafkaMetadataBackend backend; + private final boolean controllerAvailable; + + /** Creates a Metadata handler. */ + public MetadataHandler(KafkaMetadataBackend backend) { + this(backend, false); + } + + /** Creates a Metadata handler and optionally exposes a Kafka-reachable controller. */ + public MetadataHandler(KafkaMetadataBackend backend, boolean controllerAvailable) { + this.backend = checkNotNull(backend); + this.controllerAvailable = controllerAvailable; + } + + @Override + public KafkaApiSpec apiSpec() { + return API_SPEC; + } + + @Override + public CompletableFuture handle( + KafkaRequestContext context, MetadataRequest request) { + List validTopics = new ArrayList<>(); + List invalidTopics = new ArrayList<>(); + if (!request.isAllTopics()) { + for (MetadataRequestTopic topic : request.data().topics()) { + if (topic.name() != null && !Topic.isValid(topic.name())) { + invalidTopics.add( + new KafkaClusterMetadata.Topic( + topic.name(), + topic.topicId(), + TopicError.INVALID_TOPIC, + Collections.emptyList())); + } else { + // Kafka added the topic ID fields in v10, but ID-based Metadata lookup was not + // implemented until v12. This handler intentionally stops at v11. + validTopics.add(new TopicReference(topic.name(), Uuid.ZERO_UUID)); + } + } + } + + KafkaMetadataQuery query = + new KafkaMetadataQuery( + request.isAllTopics(), + validTopics, + context.listenerName(), + clientAddress(context.remoteAddress())); + return backend.getMetadata(query) + .thenApply( + metadata -> { + List topics = + new ArrayList<>(metadata.topics()); + topics.addAll(invalidTopics); + return toResponse( + request.version(), + new KafkaClusterMetadata(metadata.brokers(), topics), + controllerAvailable); + }); + } + + private static MetadataResponse toResponse( + short version, KafkaClusterMetadata metadata, boolean controllerAvailable) { + int controllerId = + controllerAvailable && !metadata.brokers().isEmpty() + ? metadata.brokers().get(0).id() + : MetadataResponse.NO_CONTROLLER_ID; + MetadataResponseData data = + new MetadataResponseData() + .setThrottleTimeMs(0) + .setControllerId(controllerId) + .setClusterAuthorizedOperations( + MetadataResponse.AUTHORIZED_OPERATIONS_OMITTED); + for (Broker broker : metadata.brokers()) { + MetadataResponseData.MetadataResponseBroker responseBroker = + new MetadataResponseData.MetadataResponseBroker() + .setNodeId(broker.id()) + .setHost(broker.host()) + .setPort(broker.port()); + if (broker.rack() != null) { + responseBroker.setRack(broker.rack()); + } + data.brokers().add(responseBroker); + } + for (KafkaClusterMetadata.Topic topic : metadata.topics()) { + MetadataResponseData.MetadataResponseTopic responseTopic = + new MetadataResponseData.MetadataResponseTopic() + .setName(topic.name()) + .setTopicId(topic.topicId()) + .setErrorCode(toKafkaError(topic.error()).code()) + .setIsInternal(topic.name() != null && Topic.isInternal(topic.name())) + .setTopicAuthorizedOperations( + MetadataResponse.AUTHORIZED_OPERATIONS_OMITTED); + for (Partition partition : topic.partitions()) { + responseTopic + .partitions() + .add( + new MetadataResponseData.MetadataResponsePartition() + .setErrorCode( + partition.leaderAvailable() + ? Errors.NONE.code() + : Errors.LEADER_NOT_AVAILABLE.code()) + .setPartitionIndex(partition.partitionId()) + .setLeaderId(partition.leaderId()) + .setLeaderEpoch(partition.leaderEpoch()) + .setReplicaNodes(partition.replicas()) + .setIsrNodes(partition.isr()) + .setOfflineReplicas(partition.offlineReplicas())); + } + data.topics().add(responseTopic); + } + return new MetadataResponse(data, version); + } + + private static Errors toKafkaError(TopicError error) { + switch (error) { + case NONE: + return Errors.NONE; + case UNKNOWN_TOPIC_OR_PARTITION: + return Errors.UNKNOWN_TOPIC_OR_PARTITION; + case UNKNOWN_TOPIC_ID: + return Errors.UNKNOWN_TOPIC_ID; + case INVALID_TOPIC: + return Errors.INVALID_TOPIC_EXCEPTION; + default: + throw new IllegalArgumentException("Unsupported metadata error " + error); + } + } + + private static InetAddress clientAddress(SocketAddress remoteAddress) { + if (remoteAddress instanceof InetSocketAddress) { + return ((InetSocketAddress) remoteAddress).getAddress(); + } + return null; + } +} 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/backend/admin/GatewayKafkaTopicAdminBackend.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/backend/admin/GatewayKafkaTopicAdminBackend.java new file mode 100644 index 00000000000..6fe0321835a --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/backend/admin/GatewayKafkaTopicAdminBackend.java @@ -0,0 +1,268 @@ +/* + * 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.backend.admin; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.kafka.format.KafkaDataFormat; +import org.apache.fluss.kafka.mapping.KafkaTopicMapper; +import org.apache.fluss.metadata.LogFormat; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.rpc.RpcGatewayService; +import org.apache.fluss.rpc.gateway.AdminGateway; +import org.apache.fluss.rpc.messages.CreateTableRequest; +import org.apache.fluss.rpc.messages.DropTableRequest; +import org.apache.fluss.rpc.messages.GetTableInfoRequest; +import org.apache.fluss.rpc.netty.server.Session; +import org.apache.fluss.security.acl.FlussPrincipal; +import org.apache.fluss.types.DataType; +import org.apache.fluss.types.DataTypes; + +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.protocol.Errors; + +import javax.annotation.Nullable; + +import java.net.InetAddress; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** Uses the TabletServer's existing coordinator gateway for Kafka topic administration. */ +@Internal +public final class GatewayKafkaTopicAdminBackend implements KafkaTopicAdminBackend { + + private final RpcGatewayService service; + private final AdminGateway gateway; + private final String databaseName; + private final KafkaTopicMapper topicMapper; + + /** Creates a topic backend backed by the Fluss coordinator admin gateway. */ + public GatewayKafkaTopicAdminBackend( + RpcGatewayService service, AdminGateway gateway, String databaseName) { + this.service = checkNotNull(service); + this.gateway = checkNotNull(gateway); + this.databaseName = checkNotNull(databaseName); + this.topicMapper = new KafkaTopicMapper(databaseName); + } + + @Override + public CompletableFuture> createTopics( + List topics, + boolean validateOnly, + String listenerName, + @Nullable InetAddress clientAddress) { + List> futures = new ArrayList<>(); + for (CreateTopic topic : topics) { + futures.add(createTopic(topic, validateOnly, listenerName, clientAddress)); + } + return collect(futures); + } + + @Override + public CompletableFuture> deleteTopics( + List topics, String listenerName, @Nullable InetAddress clientAddress) { + List> futures = new ArrayList<>(); + for (DeleteTopic topic : topics) { + futures.add(deleteTopic(topic, listenerName, clientAddress)); + } + return collect(futures); + } + + private CompletableFuture createTopic( + CreateTopic topic, + boolean validateOnly, + String listenerName, + @Nullable InetAddress clientAddress) { + TableDescriptor descriptor = createDescriptor(topic); + if (validateOnly) { + return CompletableFuture.completedFuture(success(topic, Uuid.ZERO_UUID)); + } + + CreateTableRequest request = new CreateTableRequest(); + request.setTableJson(descriptor.toJsonBytes()) + .setIgnoreIfExists(false) + .setTablePath() + .setDatabaseName(databaseName) + .setTableName(topic.name()); + setCurrentSession(listenerName, clientAddress); + return gateway.createTable(request) + .thenCompose(ignored -> getCreatedTopic(topic, listenerName, clientAddress)) + .exceptionally(failure -> failed(topic.name(), failure)); + } + + private CompletableFuture getCreatedTopic( + CreateTopic topic, String listenerName, @Nullable InetAddress clientAddress) { + GetTableInfoRequest request = new GetTableInfoRequest(); + request.setTablePath().setDatabaseName(databaseName).setTableName(topic.name()); + setCurrentSession(listenerName, clientAddress); + return gateway.getTableInfo(request) + .thenApply( + response -> success(topic, topicMapper.toTopicId(response.getTableId()))); + } + + private CompletableFuture deleteTopic( + DeleteTopic topic, String listenerName, @Nullable InetAddress clientAddress) { + if (topic.name() == null) { + return CompletableFuture.completedFuture( + new TopicResult( + null, + topic.topicId(), + Errors.UNKNOWN_TOPIC_ID, + "Deleting a Fluss table by Kafka topic id is not supported.", + -1, + (short) -1)); + } + DropTableRequest request = new DropTableRequest(); + request.setIgnoreIfNotExists(false) + .setTablePath() + .setDatabaseName(databaseName) + .setTableName(topic.name()); + setCurrentSession(listenerName, clientAddress); + return gateway.dropTable(request) + .thenApply( + ignored -> + new TopicResult( + topic.name(), + topic.topicId(), + Errors.NONE, + null, + -1, + (short) -1)) + .exceptionally(failure -> failed(topic.name(), topic.topicId(), failure)); + } + + private static TableDescriptor createDescriptor(CreateTopic topic) { + TableDescriptor.Builder builder = + TableDescriptor.builder() + .schema( + Schema.newBuilder() + .column("record_key", dataType(topic.keyFormat())) + .column("payload", dataType(topic.valueFormat())) + .column( + "event_time", + DataTypes.TIMESTAMP_LTZ(3).copy(false)) + .column( + "headers", + DataTypes.ARRAY( + DataTypes.ROW( + DataTypes.FIELD( + "name", + DataTypes.STRING() + .copy(false)), + DataTypes.FIELD( + "value", + DataTypes.BYTES())))) + .build()) + .distributedBy(topic.numPartitions()) + .property(ConfigOptions.TABLE_LOG_FORMAT, LogFormat.ARROW) + .customProperty( + KafkaDataFormat.KEY_FORMAT_CONFIG, topic.keyFormat().value()) + .customProperty( + KafkaDataFormat.VALUE_FORMAT_CONFIG, topic.valueFormat().value()); + if (topic.replicationFactor() > 0) { + builder.property( + ConfigOptions.TABLE_REPLICATION_FACTOR, (int) topic.replicationFactor()); + } + return builder.build(); + } + + private static DataType dataType(KafkaDataFormat format) { + return format == KafkaDataFormat.RAW ? DataTypes.BYTES() : DataTypes.STRING(); + } + + private static TopicResult success(CreateTopic topic, Uuid topicId) { + return new TopicResult( + topic.name(), + topicId, + Errors.NONE, + null, + topic.numPartitions(), + topic.replicationFactor()); + } + + private static TopicResult failed(String topicName, Throwable failure) { + return failed(topicName, Uuid.ZERO_UUID, failure); + } + + private static TopicResult failed(String topicName, Uuid topicId, Throwable failure) { + Throwable cause = unwrap(failure); + return new TopicResult( + topicName, topicId, toKafkaError(cause), cause.getMessage(), -1, (short) -1); + } + + private static Errors toKafkaError(Throwable failure) { + org.apache.fluss.rpc.protocol.Errors error = + org.apache.fluss.rpc.protocol.Errors.forException(failure); + switch (error) { + case TABLE_ALREADY_EXIST: + return Errors.TOPIC_ALREADY_EXISTS; + case TABLE_NOT_EXIST: + case UNKNOWN_TABLE_OR_BUCKET_EXCEPTION: + return Errors.UNKNOWN_TOPIC_OR_PARTITION; + case INVALID_TABLE_EXCEPTION: + return Errors.INVALID_REQUEST; + case INVALID_REPLICATION_FACTOR: + return Errors.INVALID_REPLICATION_FACTOR; + case BUCKET_MAX_NUM_EXCEPTION: + return Errors.INVALID_PARTITIONS; + case AUTHORIZATION_EXCEPTION: + return Errors.TOPIC_AUTHORIZATION_FAILED; + case DELETION_DISABLED_EXCEPTION: + return Errors.TOPIC_DELETION_DISABLED; + case REQUEST_TIME_OUT: + return Errors.REQUEST_TIMED_OUT; + case NOT_COORDINATOR_LEADER_EXCEPTION: + return Errors.NOT_CONTROLLER; + default: + return Errors.UNKNOWN_SERVER_ERROR; + } + } + + private void setCurrentSession(String listenerName, @Nullable InetAddress clientAddress) { + service.setCurrentSession( + new Session( + (short) 0, listenerName, false, clientAddress, FlussPrincipal.ANONYMOUS)); + } + + private static CompletableFuture> collect( + List> futures) { + CompletableFuture all = + CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])); + return all.thenApply( + ignored -> { + List results = new ArrayList<>(); + for (CompletableFuture future : futures) { + results.add(future.join()); + } + return results; + }); + } + + private static Throwable unwrap(Throwable failure) { + Throwable current = failure; + while (current instanceof CompletionException && current.getCause() != null) { + current = current.getCause(); + } + return current; + } +} diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/backend/admin/KafkaTopicAdminBackend.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/backend/admin/KafkaTopicAdminBackend.java new file mode 100644 index 00000000000..653d03367c4 --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/backend/admin/KafkaTopicAdminBackend.java @@ -0,0 +1,172 @@ +/* + * 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.backend.admin; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.kafka.format.KafkaDataFormat; + +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.protocol.Errors; + +import javax.annotation.Nullable; + +import java.net.InetAddress; +import java.util.List; +import java.util.concurrent.CompletableFuture; + +/** Backend contract for mapping Kafka topic lifecycle operations to Fluss tables. */ +@Internal +public interface KafkaTopicAdminBackend { + + /** Creates or validates the requested topics. */ + CompletableFuture> createTopics( + List topics, + boolean validateOnly, + String listenerName, + @Nullable InetAddress clientAddress); + + /** Deletes the requested topics. */ + CompletableFuture> deleteTopics( + List topics, String listenerName, @Nullable InetAddress clientAddress); + + /** A validated request to create one topic. */ + final class CreateTopic { + private final String name; + private final int numPartitions; + private final short replicationFactor; + private final KafkaDataFormat keyFormat; + private final KafkaDataFormat valueFormat; + + /** Creates a topic specification. */ + public CreateTopic( + String name, + int numPartitions, + short replicationFactor, + KafkaDataFormat keyFormat, + KafkaDataFormat valueFormat) { + this.name = name; + this.numPartitions = numPartitions; + this.replicationFactor = replicationFactor; + this.keyFormat = keyFormat; + this.valueFormat = valueFormat; + } + + /** Returns the Kafka topic name. */ + public String name() { + return name; + } + + /** Returns the requested partition count. */ + public int numPartitions() { + return numPartitions; + } + + /** Returns the requested replication factor, or {@code -1} for the Fluss default. */ + public short replicationFactor() { + return replicationFactor; + } + + /** Returns the interpretation of Kafka record keys. */ + public KafkaDataFormat keyFormat() { + return keyFormat; + } + + /** Returns the interpretation of Kafka record values. */ + public KafkaDataFormat valueFormat() { + return valueFormat; + } + } + + /** A request to delete one topic by name or Kafka topic id. */ + final class DeleteTopic { + private final @Nullable String name; + private final Uuid topicId; + + /** Creates a topic deletion reference. */ + public DeleteTopic(@Nullable String name, Uuid topicId) { + this.name = name; + this.topicId = topicId; + } + + /** Returns the topic name, if supplied. */ + public @Nullable String name() { + return name; + } + + /** Returns the Kafka topic id, or {@link Uuid#ZERO_UUID}. */ + public Uuid topicId() { + return topicId; + } + } + + /** Result of one topic lifecycle operation. */ + final class TopicResult { + private final @Nullable String name; + private final Uuid topicId; + private final Errors error; + private final @Nullable String errorMessage; + private final int numPartitions; + private final short replicationFactor; + + /** Creates a topic operation result. */ + public TopicResult( + @Nullable String name, + Uuid topicId, + Errors error, + @Nullable String errorMessage, + int numPartitions, + short replicationFactor) { + this.name = name; + this.topicId = topicId; + this.error = error; + this.errorMessage = errorMessage; + this.numPartitions = numPartitions; + this.replicationFactor = replicationFactor; + } + + /** Returns the topic name, if known. */ + public @Nullable String name() { + return name; + } + + /** Returns the Kafka topic id, if known. */ + public Uuid topicId() { + return topicId; + } + + /** Returns the Kafka protocol error. */ + public Errors error() { + return error; + } + + /** Returns the optional error detail. */ + public @Nullable String errorMessage() { + return errorMessage; + } + + /** Returns the created partition count. */ + public int numPartitions() { + return numPartitions; + } + + /** Returns the created replication factor. */ + public short replicationFactor() { + return replicationFactor; + } + } +} diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/backend/metadata/GatewayKafkaMetadataBackend.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/backend/metadata/GatewayKafkaMetadataBackend.java new file mode 100644 index 00000000000..087eb6a8d46 --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/backend/metadata/GatewayKafkaMetadataBackend.java @@ -0,0 +1,281 @@ +/* + * 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.backend.metadata; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.kafka.backend.metadata.KafkaClusterMetadata.Broker; +import org.apache.fluss.kafka.backend.metadata.KafkaClusterMetadata.Partition; +import org.apache.fluss.kafka.backend.metadata.KafkaClusterMetadata.Topic; +import org.apache.fluss.kafka.backend.metadata.KafkaClusterMetadata.TopicError; +import org.apache.fluss.kafka.backend.metadata.KafkaMetadataQuery.TopicReference; +import org.apache.fluss.kafka.mapping.KafkaTopicMapper; +import org.apache.fluss.rpc.RpcGatewayService; +import org.apache.fluss.rpc.gateway.TabletServerGateway; +import org.apache.fluss.rpc.messages.ListTablesRequest; +import org.apache.fluss.rpc.messages.MetadataRequest; +import org.apache.fluss.rpc.messages.MetadataResponse; +import org.apache.fluss.rpc.messages.PbBucketMetadata; +import org.apache.fluss.rpc.messages.PbServerNode; +import org.apache.fluss.rpc.messages.PbTableMetadata; +import org.apache.fluss.rpc.messages.PbTablePath; +import org.apache.fluss.rpc.netty.server.Session; +import org.apache.fluss.security.acl.FlussPrincipal; + +import org.apache.kafka.common.Uuid; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** Adapts the existing Fluss metadata RPC to the Kafka Metadata backend contract. */ +@Internal +public final class GatewayKafkaMetadataBackend implements KafkaMetadataBackend { + + private static final Logger LOG = LoggerFactory.getLogger(GatewayKafkaMetadataBackend.class); + + private final RpcGatewayService service; + private final TabletServerGateway gateway; + private final String databaseName; + private final KafkaTopicMapper topicMapper; + + /** Creates a metadata backend backed by the local TabletServer gateway. */ + public GatewayKafkaMetadataBackend( + RpcGatewayService service, TabletServerGateway gateway, String databaseName) { + this.service = checkNotNull(service); + this.gateway = checkNotNull(gateway); + this.databaseName = checkNotNull(databaseName); + this.topicMapper = new KafkaTopicMapper(databaseName); + } + + @Override + public CompletableFuture getMetadata(KafkaMetadataQuery query) { + if (query.allTopics() || containsTopicId(query.topics())) { + setCurrentSession(query); + return gateway.listTables(new ListTablesRequest().setDatabaseName(databaseName)) + .thenCompose( + response -> + requestFlussMetadata( + query, + new LinkedHashSet<>(response.getTableNamesList()))); + } + + Set topicNames = new LinkedHashSet<>(); + for (TopicReference topic : query.topics()) { + if (topic.topicName() != null) { + topicNames.add(topic.topicName()); + } + } + return requestFlussMetadata(query, topicNames); + } + + private CompletableFuture requestFlussMetadata( + KafkaMetadataQuery query, Set topicNames) { + return requestFlussMetadata(query, topicNames, true); + } + + private CompletableFuture requestFlussMetadata( + KafkaMetadataQuery query, Set topicNames, boolean refreshAndRetry) { + MetadataRequest request = new MetadataRequest(); + for (String topicName : topicNames) { + request.addAllTablePaths( + Collections.singletonList( + new PbTablePath() + .setDatabaseName(databaseName) + .setTableName(topicName))); + } + setCurrentSession(query); + try { + return gateway.metadata(request) + .handle( + (response, failure) -> + failure == null + ? CompletableFuture.completedFuture( + toKafkaMetadata(query, response)) + : recoverMetadataFailure( + query, failure, refreshAndRetry)) + .thenCompose(future -> future); + } catch (Throwable failure) { + return recoverMetadataFailure(query, failure, refreshAndRetry); + } + } + + private CompletableFuture recoverMetadataFailure( + KafkaMetadataQuery query, Throwable failure, boolean refreshAndRetry) { + if (refreshAndRetry) { + return currentTopicNames(query) + .thenCompose(currentNames -> requestFlussMetadata(query, currentNames, false)); + } + LOG.warn("Failed to load Kafka metadata from Fluss.", unwrap(failure)); + CompletableFuture failed = new CompletableFuture<>(); + failed.completeExceptionally(unwrap(failure)); + return failed; + } + + private CompletableFuture> currentTopicNames(KafkaMetadataQuery query) { + setCurrentSession(query); + return gateway.listTables(new ListTablesRequest().setDatabaseName(databaseName)) + .thenApply( + response -> { + Set currentNames = + new LinkedHashSet<>(response.getTableNamesList()); + if (!query.allTopics() && !containsTopicId(query.topics())) { + Set requestedNames = new HashSet<>(); + for (TopicReference topic : query.topics()) { + if (topic.topicName() != null) { + requestedNames.add(topic.topicName()); + } + } + currentNames.retainAll(requestedNames); + } + return currentNames; + }); + } + + private KafkaClusterMetadata toKafkaMetadata( + KafkaMetadataQuery query, MetadataResponse response) { + List brokers = new ArrayList<>(); + Set aliveBrokerIds = new HashSet<>(); + for (PbServerNode server : response.getTabletServersList()) { + brokers.add( + new Broker( + server.getNodeId(), + server.getHost(), + server.getPort(), + server.hasRack() ? server.getRack() : null)); + aliveBrokerIds.add(server.getNodeId()); + } + Collections.sort(brokers, Comparator.comparingInt(Broker::id)); + + Map topicsByName = new HashMap<>(); + Map topicsById = new HashMap<>(); + for (PbTableMetadata table : response.getTableMetadatasList()) { + if (!databaseName.equals(table.getTablePath().getDatabaseName())) { + continue; + } + Topic topic = toKafkaTopic(table, aliveBrokerIds); + topicsByName.put(topic.name(), topic); + topicsById.put(topic.topicId(), topic); + } + + List topics = new ArrayList<>(); + if (query.allTopics()) { + topics.addAll(topicsByName.values()); + Collections.sort(topics, Comparator.comparing(Topic::name)); + } else { + for (TopicReference reference : query.topics()) { + Topic topic = + reference.hasTopicId() + ? topicsById.get(reference.topicId()) + : topicsByName.get(reference.topicName()); + if (topic != null && matches(reference, topic)) { + topics.add(topic); + } else { + topics.add(missingTopic(reference)); + } + } + } + return new KafkaClusterMetadata(brokers, topics); + } + + private Topic toKafkaTopic(PbTableMetadata table, Set aliveBrokerIds) { + List partitions = new ArrayList<>(); + for (PbBucketMetadata bucket : table.getBucketMetadatasList()) { + List replicas = new ArrayList<>(); + List isr = new ArrayList<>(); + List offlineReplicas = new ArrayList<>(); + for (int replicaId : bucket.getReplicaIds()) { + replicas.add(replicaId); + if (aliveBrokerIds.contains(replicaId)) { + isr.add(replicaId); + } else { + offlineReplicas.add(replicaId); + } + } + boolean leaderAvailable = + bucket.hasLeaderId() && aliveBrokerIds.contains(bucket.getLeaderId()); + partitions.add( + new Partition( + bucket.getBucketId(), + leaderAvailable ? bucket.getLeaderId() : -1, + bucket.hasLeaderEpoch() ? bucket.getLeaderEpoch() : -1, + replicas, + isr, + offlineReplicas, + leaderAvailable)); + } + Collections.sort(partitions, Comparator.comparingInt(Partition::partitionId)); + return new Topic( + table.getTablePath().getTableName(), + topicMapper.toTopicId(table.getTableId()), + TopicError.NONE, + partitions); + } + + private static Topic missingTopic(TopicReference reference) { + TopicError error = + reference.hasTopicId() + ? TopicError.UNKNOWN_TOPIC_ID + : TopicError.UNKNOWN_TOPIC_OR_PARTITION; + return new Topic( + reference.topicName(), reference.topicId(), error, Collections.emptyList()); + } + + private static boolean matches(TopicReference reference, Topic topic) { + return (reference.topicName() == null || reference.topicName().equals(topic.name())) + && (!reference.hasTopicId() || reference.topicId().equals(topic.topicId())); + } + + private void setCurrentSession(KafkaMetadataQuery query) { + service.setCurrentSession( + new Session( + (short) 0, + query.listenerName(), + false, + query.clientAddress(), + FlussPrincipal.ANONYMOUS)); + } + + private static boolean containsTopicId(List topics) { + for (TopicReference topic : topics) { + if (topic.hasTopicId()) { + return true; + } + } + return false; + } + + private static Throwable unwrap(Throwable failure) { + Throwable current = failure; + while (current instanceof CompletionException && current.getCause() != null) { + current = current.getCause(); + } + return current; + } +} diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/backend/metadata/KafkaClusterMetadata.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/backend/metadata/KafkaClusterMetadata.java new file mode 100644 index 00000000000..bbc634af6f0 --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/backend/metadata/KafkaClusterMetadata.java @@ -0,0 +1,210 @@ +/* + * 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.backend.metadata; + +import org.apache.fluss.annotation.Internal; + +import org.apache.kafka.common.Uuid; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** Kafka-domain cluster metadata returned by a Fluss metadata backend. */ +@Internal +public final class KafkaClusterMetadata { + + private final List brokers; + private final List topics; + + /** Creates cluster metadata. */ + public KafkaClusterMetadata(List brokers, List topics) { + this.brokers = immutableCopy(brokers); + this.topics = immutableCopy(topics); + } + + /** Returns Kafka-reachable brokers. */ + public List brokers() { + return brokers; + } + + /** Returns topic metadata and topic-level errors. */ + public List topics() { + return topics; + } + + private static List immutableCopy(List values) { + return Collections.unmodifiableList(new ArrayList<>(checkNotNull(values))); + } + + /** Kafka-reachable broker information. */ + @Internal + public static final class Broker { + + private final int id; + private final String host; + private final int port; + private final @Nullable String rack; + + /** Creates broker information. */ + public Broker(int id, String host, int port, @Nullable String rack) { + this.id = id; + this.host = checkNotNull(host); + this.port = port; + this.rack = rack; + } + + /** Returns the Kafka broker ID. */ + public int id() { + return id; + } + + /** Returns the Kafka listener host. */ + public String host() { + return host; + } + + /** Returns the Kafka listener port. */ + public int port() { + return port; + } + + /** Returns the broker rack, if configured. */ + public @Nullable String rack() { + return rack; + } + } + + /** Topic-level error independent of a Kafka response schema version. */ + @Internal + public enum TopicError { + NONE, + UNKNOWN_TOPIC_OR_PARTITION, + UNKNOWN_TOPIC_ID, + INVALID_TOPIC + } + + /** Metadata for one Kafka topic. */ + @Internal + public static final class Topic { + + private final @Nullable String name; + private final Uuid topicId; + private final TopicError error; + private final List partitions; + + /** Creates topic metadata. */ + public Topic( + @Nullable String name, Uuid topicId, TopicError error, List partitions) { + this.name = name; + this.topicId = checkNotNull(topicId); + this.error = checkNotNull(error); + this.partitions = immutableCopy(partitions); + } + + /** Returns the Kafka topic name, if known. */ + public @Nullable String name() { + return name; + } + + /** Returns the stable Kafka topic ID. */ + public Uuid topicId() { + return topicId; + } + + /** Returns the topic-level domain error. */ + public TopicError error() { + return error; + } + + /** Returns the topic partitions. */ + public List partitions() { + return partitions; + } + } + + /** Metadata for one Kafka partition backed by a Fluss bucket. */ + @Internal + public static final class Partition { + + private final int partitionId; + private final int leaderId; + private final int leaderEpoch; + private final List replicas; + private final List isr; + private final List offlineReplicas; + private final boolean leaderAvailable; + + /** Creates partition metadata. */ + public Partition( + int partitionId, + int leaderId, + int leaderEpoch, + List replicas, + List isr, + List offlineReplicas, + boolean leaderAvailable) { + this.partitionId = partitionId; + this.leaderId = leaderId; + this.leaderEpoch = leaderEpoch; + this.replicas = immutableCopy(replicas); + this.isr = immutableCopy(isr); + this.offlineReplicas = immutableCopy(offlineReplicas); + this.leaderAvailable = leaderAvailable; + } + + /** Returns the Kafka partition ID. */ + public int partitionId() { + return partitionId; + } + + /** Returns the current leader ID, or {@code -1} when unavailable. */ + public int leaderId() { + return leaderId; + } + + /** Returns the leader epoch, or {@code -1} when unavailable. */ + public int leaderEpoch() { + return leaderEpoch; + } + + /** Returns assigned replica IDs. */ + public List replicas() { + return replicas; + } + + /** Returns replica IDs currently visible as in-sync. */ + public List isr() { + return isr; + } + + /** Returns assigned replicas whose TabletServers are unavailable. */ + public List offlineReplicas() { + return offlineReplicas; + } + + /** Returns whether the partition has a reachable leader. */ + public boolean leaderAvailable() { + return leaderAvailable; + } + } +} diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/backend/metadata/KafkaMetadataBackend.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/backend/metadata/KafkaMetadataBackend.java new file mode 100644 index 00000000000..abb02a920ed --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/backend/metadata/KafkaMetadataBackend.java @@ -0,0 +1,30 @@ +/* + * 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.backend.metadata; + +import org.apache.fluss.annotation.Internal; + +import java.util.concurrent.CompletableFuture; + +/** Narrow backend used by the Kafka Metadata API. */ +@Internal +public interface KafkaMetadataBackend { + + /** Resolves Kafka-domain metadata asynchronously. */ + CompletableFuture getMetadata(KafkaMetadataQuery query); +} diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/backend/metadata/KafkaMetadataQuery.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/backend/metadata/KafkaMetadataQuery.java new file mode 100644 index 00000000000..01e21fa4440 --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/backend/metadata/KafkaMetadataQuery.java @@ -0,0 +1,102 @@ +/* + * 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.backend.metadata; + +import org.apache.fluss.annotation.Internal; + +import org.apache.kafka.common.Uuid; + +import javax.annotation.Nullable; + +import java.net.InetAddress; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** Domain query used by the Metadata API to access the Fluss adapter layer. */ +@Internal +public final class KafkaMetadataQuery { + + private final boolean allTopics; + private final List topics; + private final String listenerName; + private final @Nullable InetAddress clientAddress; + + /** Creates a metadata query. */ + public KafkaMetadataQuery( + boolean allTopics, + List topics, + String listenerName, + @Nullable InetAddress clientAddress) { + this.allTopics = allTopics; + this.topics = Collections.unmodifiableList(new ArrayList<>(checkNotNull(topics))); + this.listenerName = checkNotNull(listenerName); + this.clientAddress = clientAddress; + } + + /** Returns whether all Kafka topics should be returned. */ + public boolean allTopics() { + return allTopics; + } + + /** Returns the explicitly requested topic identities. */ + public List topics() { + return topics; + } + + /** Returns the Kafka listener used by the client connection. */ + public String listenerName() { + return listenerName; + } + + /** Returns the client address when it is available. */ + public @Nullable InetAddress clientAddress() { + return clientAddress; + } + + /** Kafka topic name and ID supplied by a Metadata request. */ + @Internal + public static final class TopicReference { + + private final @Nullable String topicName; + private final Uuid topicId; + + /** Creates a topic reference. */ + public TopicReference(@Nullable String topicName, Uuid topicId) { + this.topicName = topicName; + this.topicId = checkNotNull(topicId); + } + + /** Returns the requested topic name, if present. */ + public @Nullable String topicName() { + return topicName; + } + + /** Returns the requested topic ID, or {@link Uuid#ZERO_UUID} when absent. */ + public Uuid topicId() { + return topicId; + } + + /** Returns whether this reference identifies a topic by ID. */ + public boolean hasTopicId() { + return !Uuid.ZERO_UUID.equals(topicId); + } + } +} 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/main/java/org/apache/fluss/kafka/format/KafkaDataFormat.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/KafkaDataFormat.java new file mode 100644 index 00000000000..0581abdfe36 --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/format/KafkaDataFormat.java @@ -0,0 +1,61 @@ +/* + * 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.format; + +import org.apache.fluss.annotation.Internal; + +import java.util.Locale; + +/** Supported interpretations of Kafka record key and value bytes. */ +@Internal +public enum KafkaDataFormat { + RAW("raw"), + STRING("string"); + + /** Kafka topic config and Fluss custom property controlling the record key format. */ + public static final String KEY_FORMAT_CONFIG = "fluss.key.format"; + + /** Kafka topic config and Fluss custom property controlling the record value format. */ + public static final String VALUE_FORMAT_CONFIG = "fluss.value.format"; + + private final String value; + + KafkaDataFormat(String value) { + this.value = value; + } + + /** Parses a topic config value. */ + public static KafkaDataFormat parse(String value) { + if (value == null) { + throw new IllegalArgumentException("Kafka data format must not be null."); + } + String normalized = value.trim().toLowerCase(Locale.ROOT); + for (KafkaDataFormat format : values()) { + if (format.value.equals(normalized)) { + return format; + } + } + throw new IllegalArgumentException( + "Unsupported Kafka data format '" + value + "'. Expected raw or string."); + } + + /** Returns the persisted topic config value. */ + public String value() { + return value; + } +} diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/mapping/KafkaTopicMapper.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/mapping/KafkaTopicMapper.java new file mode 100644 index 00000000000..6b75bdfd0d1 --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/mapping/KafkaTopicMapper.java @@ -0,0 +1,65 @@ +/* + * 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.mapping; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.metadata.TablePath; + +import org.apache.kafka.common.Uuid; + +import static org.apache.fluss.utils.Preconditions.checkArgument; +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** Maps Kafka topic identities to tables in the configured Fluss Kafka database. */ +@Internal +public final class KafkaTopicMapper { + + // ASCII "Fluss" followed by zero bytes. A dedicated namespace avoids Kafka-reserved UUIDs. + private static final long TOPIC_ID_NAMESPACE = 0x466c757373000000L; + + private final String databaseName; + + /** Creates a topic mapper for one Fluss database. */ + public KafkaTopicMapper(String databaseName) { + this.databaseName = checkNotNull(databaseName); + } + + /** Maps a Kafka topic name to its Fluss table path. */ + public TablePath toTablePath(String topicName) { + return TablePath.of(databaseName, topicName); + } + + /** Maps a Fluss table ID to a stable Kafka topic ID. */ + public Uuid toTopicId(long tableId) { + checkArgument(tableId >= 0, "Table ID must be non-negative, but was %s.", tableId); + return new Uuid(TOPIC_ID_NAMESPACE, tableId); + } + + /** Returns whether a Kafka topic ID can represent a Fluss table ID. */ + public boolean isMappedTopicId(Uuid topicId) { + return topicId != null + && topicId.getMostSignificantBits() == TOPIC_ID_NAMESPACE + && topicId.getLeastSignificantBits() >= 0L; + } + + /** Extracts the Fluss table ID encoded in a Kafka topic ID. */ + public long toTableId(Uuid topicId) { + checkArgument(isMappedTopicId(topicId), "Topic ID %s is not a Fluss topic ID.", topicId); + return topicId.getLeastSignificantBits(); + } +} 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/KafkaConfigsTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaConfigsTest.java index 7f502494f2e..1c52120fae4 100644 --- a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaConfigsTest.java +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaConfigsTest.java @@ -37,12 +37,18 @@ public void testFromMap() throws Exception { map.put(ConfigOptions.KAFKA_ENABLED.key(), "true"); map.put(ConfigOptions.KAFKA_LISTENER_NAMES.key(), "kafka,kafka_sasl"); map.put(ConfigOptions.KAFKA_DATABASE.key(), "fluss"); + map.put(ConfigOptions.KAFKA_DEFAULT_KEY_FORMAT.key(), "string"); + map.put(ConfigOptions.KAFKA_DEFAULT_VALUE_FORMAT.key(), "string"); Configuration configuration = Configuration.fromMap(map); assertThat(configuration.getBoolean(ConfigOptions.KAFKA_ENABLED)).isTrue(); assertThat(configuration.get(ConfigOptions.KAFKA_LISTENER_NAMES)) .isEqualTo(Arrays.asList("kafka", "kafka_sasl")); assertThat(configuration.getString(ConfigOptions.KAFKA_DATABASE)).isEqualTo("fluss"); + assertThat(configuration.getString(ConfigOptions.KAFKA_DEFAULT_KEY_FORMAT)) + .isEqualTo("string"); + assertThat(configuration.getString(ConfigOptions.KAFKA_DEFAULT_VALUE_FORMAT)) + .isEqualTo("string"); } @Test @@ -52,5 +58,9 @@ public void testFromDefault() throws Exception { assertThat(configuration.get(ConfigOptions.KAFKA_LISTENER_NAMES)) .isEqualTo(Collections.singletonList("KAFKA")); assertThat(configuration.getString(ConfigOptions.KAFKA_DATABASE)).isEqualTo("kafka"); + assertThat(configuration.getString(ConfigOptions.KAFKA_DEFAULT_KEY_FORMAT)) + .isEqualTo("raw"); + assertThat(configuration.getString(ConfigOptions.KAFKA_DEFAULT_VALUE_FORMAT)) + .isEqualTo("raw"); } } diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaMetadataHandlerTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaMetadataHandlerTest.java new file mode 100644 index 00000000000..f7e7dd9bd47 --- /dev/null +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaMetadataHandlerTest.java @@ -0,0 +1,372 @@ +/* + * 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.exception.TableNotExistException; +import org.apache.fluss.rpc.TestingTabletGatewayService; +import org.apache.fluss.rpc.messages.ListTablesRequest; +import org.apache.fluss.rpc.messages.ListTablesResponse; +import org.apache.fluss.rpc.messages.PbBucketMetadata; +import org.apache.fluss.rpc.messages.PbServerNode; +import org.apache.fluss.rpc.messages.PbTableMetadata; +import org.apache.fluss.rpc.messages.PbTablePath; +import org.apache.fluss.shaded.netty4.io.netty.buffer.ByteBuf; +import org.apache.fluss.shaded.netty4.io.netty.buffer.ByteBufAllocator; + +import org.apache.kafka.common.Node; +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.message.MetadataRequestData; +import org.apache.kafka.common.message.MetadataRequestData.MetadataRequestTopic; +import org.apache.kafka.common.message.MetadataResponseData.MetadataResponsePartition; +import org.apache.kafka.common.message.MetadataResponseData.MetadataResponseTopic; +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.protocol.types.RawTaggedField; +import org.apache.kafka.common.requests.AbstractResponse; +import org.apache.kafka.common.requests.MetadataRequest; +import org.apache.kafka.common.requests.MetadataResponse; +import org.apache.kafka.common.requests.RequestHeader; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Protocol compatibility tests for the Kafka Metadata API. */ +public class KafkaMetadataHandlerTest { + + private static final Uuid TOPIC_ID = new Uuid(0x466c757373000000L, 123L); + + @Test + public void testNamedTopicForEverySupportedVersion() { + TestingMetadataGatewayService service = new TestingMetadataGatewayService(); + for (short version = ApiKeys.METADATA.oldestVersion(); version <= 11; version++) { + MetadataRequest request = + new MetadataRequest( + new MetadataRequestData() + .setTopics( + MetadataRequest.convertToMetadataRequestTopic( + Collections.singletonList("topic"))), + version); + if (version >= 9) { + request.data() + .unknownTaggedFields() + .add(new RawTaggedField(100, new byte[] {1, 2, 3})); + request.data() + .topics() + .get(0) + .unknownTaggedFields() + .add(new RawTaggedField(101, new byte[] {4, 5, 6})); + } + MetadataResponse response = handle(service, request, version); + + assertThat(response.brokers()).hasSize(2); + assertThat(response.controller()).isNull(); + Node broker = response.brokers().iterator().next(); + assertThat(broker.host()).isEqualTo("broker-1"); + assertThat(broker.port()).isEqualTo(9092); + assertThat(broker.rack()).isEqualTo(version >= 1 ? "rack-a" : null); + MetadataResponseTopic topic = response.data().topics().find("topic"); + assertThat(topic.errorCode()).isEqualTo(Errors.NONE.code()); + assertThat(topic.partitions()).hasSize(2); + assertThat(topic.topicId()).isEqualTo(version >= 10 ? TOPIC_ID : Uuid.ZERO_UUID); + MetadataResponsePartition partition = topic.partitions().get(0); + assertThat(partition.partitionIndex()).isZero(); + assertThat(partition.leaderId()).isEqualTo(1); + assertThat(partition.replicaNodes()).containsExactly(1, 2); + assertThat(partition.isrNodes()).containsExactly(1, 2); + assertThat(partition.offlineReplicas()).isEmpty(); + } + assertThat(service.lastListenerName).isEqualTo("KAFKA"); + } + + @Test + public void testAllTopicsForEverySupportedVersion() { + TestingMetadataGatewayService service = new TestingMetadataGatewayService(); + for (short version = ApiKeys.METADATA.oldestVersion(); version <= 11; version++) { + MetadataRequest request = allTopicsRequest(version); + if (version >= 9) { + request.data() + .unknownTaggedFields() + .add(new RawTaggedField(102, new byte[] {7, 8, 9})); + } + + MetadataResponse response = handle(service, request, version); + + assertThat(response.data().topics()) + .extracting(MetadataResponseTopic::name) + .containsExactly("other", "topic"); + } + } + + @Test + public void testAllTopicsAndMissingAndInvalidTopic() { + TestingMetadataGatewayService service = new TestingMetadataGatewayService(); + + MetadataResponse allTopics = + handle(service, MetadataRequest.Builder.allTopics().build((short) 9), (short) 9); + assertThat(allTopics.data().topics()) + .extracting(MetadataResponseTopic::name) + .containsExactlyInAnyOrder("other", "topic"); + + MetadataRequest requestedTopics = + new MetadataRequest.Builder(Arrays.asList("missing", "invalid topic"), false) + .build((short) 9); + MetadataResponse errors = handle(service, requestedTopics, (short) 9); + assertThat(errors.errors()) + .containsEntry("missing", Errors.UNKNOWN_TOPIC_OR_PARTITION) + .containsEntry("invalid topic", Errors.INVALID_TOPIC_EXCEPTION); + } + + @Test + public void testV10AndV11IgnoreRequestTopicIdAndLookupByName() { + TestingMetadataGatewayService service = new TestingMetadataGatewayService(); + for (short version = 10; version <= 11; version++) { + MetadataRequest request = + new MetadataRequest( + new MetadataRequestData() + .setTopics( + Collections.singletonList( + new MetadataRequestTopic() + .setName("topic") + .setTopicId( + new Uuid( + 0x466c757373000000L, + 999L)))), + version); + + MetadataResponse response = handle(service, request, version); + + assertThat(response.errorCounts()).containsOnlyKeys(Errors.NONE); + assertThat(response.data().topics().find("topic").topicId()).isEqualTo(TOPIC_ID); + } + } + + @Test + public void testTopicIdentityAcrossDeleteAndRecreate() { + TestingMetadataGatewayService service = new TestingMetadataGatewayService(); + + MetadataResponse initial = handle(service, namedTopicRequest("topic"), (short) 11); + assertThat(initial.data().topics().find("topic").topicId()).isEqualTo(TOPIC_ID); + + service.removeTable("topic"); + MetadataResponse deleted = handle(service, namedTopicRequest("topic"), (short) 11); + assertThat(deleted.errorCounts()) + .containsExactlyEntriesOf( + Collections.singletonMap(Errors.UNKNOWN_TOPIC_OR_PARTITION, 1)); + + service.putTable("topic", 223L); + Uuid recreatedTopicId = new Uuid(0x466c757373000000L, 223L); + MetadataResponse recreatedByName = + handle( + service, + new MetadataRequest.Builder(Collections.singletonList("topic"), false) + .build((short) 11), + (short) 11); + assertThat(recreatedByName.data().topics().find("topic").topicId()) + .isEqualTo(recreatedTopicId) + .isNotEqualTo(TOPIC_ID); + } + + @Test + public void testDeleteRaceBecomesUnknownTopicResult() { + TestingMetadataGatewayService service = new TestingMetadataGatewayService(); + service.removeTable("topic"); + service.failNextMetadataAsMissing = true; + + MetadataResponse response = handle(service, namedTopicRequest("topic"), (short) 11); + + assertThat(response.errorCounts()) + .containsExactlyEntriesOf( + Collections.singletonMap(Errors.UNKNOWN_TOPIC_OR_PARTITION, 1)); + } + + @Test + public void testUnavailableLeaderUsesPartitionError() { + TestingMetadataGatewayService service = new TestingMetadataGatewayService(); + service.topicLeaderAvailable = false; + MetadataRequest request = + new MetadataRequest.Builder(Collections.singletonList("topic"), false) + .build((short) 11); + + MetadataResponse response = handle(service, request, (short) 11); + + MetadataResponseTopic topic = response.data().topics().find("topic"); + assertThat(topic.errorCode()).isEqualTo(Errors.NONE.code()); + MetadataResponsePartition partition = topic.partitions().get(0); + assertThat(partition.errorCode()).isEqualTo(Errors.LEADER_NOT_AVAILABLE.code()); + assertThat(partition.leaderId()).isEqualTo(-1); + assertThat(partition.replicaNodes()).containsExactly(1, 2, 3); + assertThat(partition.isrNodes()).containsExactly(1, 2); + assertThat(partition.offlineReplicas()).containsExactly(3); + } + + @Test + public void testUnexpectedGatewayFailureUsesRequestErrorResponse() { + TestingMetadataGatewayService service = new TestingMetadataGatewayService(); + service.failMetadata = true; + MetadataRequest request = + new MetadataRequest.Builder(Collections.singletonList("topic"), false) + .build((short) 11); + + MetadataResponse response = handle(service, request, (short) 11); + + assertThat(response.errorCounts()) + .containsExactlyEntriesOf(Collections.singletonMap(Errors.UNKNOWN_SERVER_ERROR, 1)); + assertThat(response.brokers()).isEmpty(); + } + + private static MetadataRequest namedTopicRequest(String topicName) { + return new MetadataRequest( + new MetadataRequestData() + .setTopics( + Collections.singletonList( + new MetadataRequestTopic() + .setName(topicName) + .setTopicId(Uuid.ZERO_UUID))), + (short) 11); + } + + private static MetadataRequest allTopicsRequest(short version) { + MetadataRequestData data = new MetadataRequestData(); + data.setTopics(version == 0 ? Collections.emptyList() : null); + return new MetadataRequest(data, version); + } + + private static MetadataResponse handle( + TestingMetadataGatewayService service, MetadataRequest requestBody, short version) { + KafkaRequestHandler handler = new KafkaRequestHandler(service, service, "kafka"); + KafkaRequest request = + new KafkaRequest( + ApiKeys.METADATA, + version, + new RequestHeader(ApiKeys.METADATA, version, "client-id", 1), + requestBody, + "KAFKA", + ByteBufAllocator.DEFAULT.buffer(), + new TestingChannelHandlerContext(), + new CompletableFuture<>()); + handler.processRequest(request); + ByteBuf responseBuffer = request.responseBuffer(); + try { + return (MetadataResponse) + AbstractResponse.parseResponse(responseBuffer.nioBuffer(), request.header()); + } finally { + responseBuffer.release(); + } + } + + private static final class TestingMetadataGatewayService extends TestingTabletGatewayService { + + private final Map tables = new LinkedHashMap<>(); + private String lastListenerName; + private boolean topicLeaderAvailable = true; + private boolean failMetadata; + private boolean failNextMetadataAsMissing; + + private TestingMetadataGatewayService() { + tables.put("topic", 123L); + tables.put("other", 124L); + } + + @Override + public CompletableFuture listTables(ListTablesRequest request) { + assertThat(request.getDatabaseName()).isEqualTo("kafka"); + return CompletableFuture.completedFuture( + new ListTablesResponse().addAllTableNames(new ArrayList<>(tables.keySet()))); + } + + @Override + public CompletableFuture metadata( + org.apache.fluss.rpc.messages.MetadataRequest request) { + lastListenerName = currentListenerName(); + if (failMetadata) { + CompletableFuture failure = + new CompletableFuture<>(); + failure.completeExceptionally(new IllegalStateException("metadata unavailable")); + return failure; + } + if (failNextMetadataAsMissing) { + failNextMetadataAsMissing = false; + throw new TableNotExistException("table was deleted"); + } + List topics = new ArrayList<>(); + for (PbTablePath tablePath : request.getTablePathsList()) { + Long tableId = tables.get(tablePath.getTableName()); + if (tableId != null) { + topics.add( + tableMetadata( + tablePath.getTableName(), + tableId, + !"topic".equals(tablePath.getTableName()) + || topicLeaderAvailable)); + } + } + return CompletableFuture.completedFuture( + new org.apache.fluss.rpc.messages.MetadataResponse() + .addAllTabletServers( + Arrays.asList( + new PbServerNode() + .setNodeId(1) + .setHost("broker-1") + .setPort(9092) + .setRack("rack-a"), + new PbServerNode() + .setNodeId(2) + .setHost("broker-2") + .setPort(9093))) + .addAllTableMetadatas(topics)); + } + + private void putTable(String topic, long tableId) { + tables.put(topic, tableId); + } + + private void removeTable(String topic) { + tables.remove(topic); + } + + private static PbTableMetadata tableMetadata( + String topic, long tableId, boolean leaderAvailable) { + return new PbTableMetadata() + .setTablePath(new PbTablePath().setDatabaseName("kafka").setTableName(topic)) + .setTableId(tableId) + .addAllBucketMetadatas( + Arrays.asList( + new PbBucketMetadata() + .setBucketId(0) + .setLeaderId(leaderAvailable ? 1 : 3) + .setLeaderEpoch(5) + .setReplicaIds( + leaderAvailable + ? new int[] {1, 2} + : new int[] {1, 2, 3}), + new PbBucketMetadata() + .setBucketId(1) + .setLeaderId(2) + .setLeaderEpoch(6) + .setReplicaIds(new int[] {1, 2}))); + } + } +} 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..1f3032c219c 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 @@ -18,22 +18,33 @@ package org.apache.fluss.kafka; import org.apache.fluss.rpc.TestingTabletGatewayService; +import org.apache.fluss.rpc.gateway.AdminGateway; 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.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; +import static org.mockito.Mockito.mock; /** Tests for {@link KafkaRequestHandler}. */ public class KafkaRequestHandlerTest { @@ -54,7 +65,7 @@ public void testKafkaApiVersionsNotSupported() { ByteBufAllocator.DEFAULT.buffer(), ctx, new CompletableFuture<>()); - handler.handleApiVersionsRequest(request); + handler.processRequest(request); ByteBuf responseBuffer = request.responseBuffer(); ApiVersionsResponse response = @@ -66,53 +77,171 @@ 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); + } + @Test - public void testKafkaApiVersionsRequest() { + public void testAdminCapabilitiesAreAdvertisedWhenCoordinatorGatewayIsAvailable() { + TestingTabletGatewayService service = new TestingTabletGatewayService(); + KafkaRequestHandler handler = + new KafkaRequestHandler(service, service, mock(AdminGateway.class), "kafka"); + short version = ApiKeys.API_VERSIONS.latestVersion(); + ApiVersionsRequest requestBody = new ApiVersionsRequest.Builder().build(version); + KafkaRequest request = + new KafkaRequest( + ApiKeys.API_VERSIONS, + version, + new RequestHeader(ApiKeys.API_VERSIONS, version, "client-id", 0), + requestBody, + ByteBufAllocator.DEFAULT.buffer(), + new TestingChannelHandlerContext(), + new CompletableFuture<>()); + + handler.processRequest(request); + + ApiVersionsResponse response = parseApiVersionsResponse(request); + assertSuccessfulResponseDefaults(response); + assertThat(response.data().apiKeys()) + .extracting(ApiVersion::apiKey, ApiVersion::minVersion, ApiVersion::maxVersion) + .containsExactly( + tuple(ApiKeys.METADATA.id, ApiKeys.METADATA.oldestVersion(), (short) 11), + tuple( + ApiKeys.API_VERSIONS.id, + ApiKeys.API_VERSIONS.oldestVersion(), + ApiKeys.API_VERSIONS.latestVersion()), + tuple( + ApiKeys.CREATE_TOPICS.id, + ApiKeys.CREATE_TOPICS.oldestVersion(), + ApiKeys.CREATE_TOPICS.latestVersion()), + tuple( + ApiKeys.DELETE_TOPICS.id, + ApiKeys.DELETE_TOPICS.oldestVersion(), + ApiKeys.DELETE_TOPICS.latestVersion())); + } + + private static ApiVersionsResponse requestApiVersions( + KafkaRequestHandler handler, short version) { + ApiVersionsRequest apiVersionsRequest = new ApiVersionsRequest.Builder().build(version); + ChannelHandlerContext ctx = new TestingChannelHandlerContext(); + KafkaRequest request = + new KafkaRequest( + ApiKeys.API_VERSIONS, + version, + new RequestHeader(ApiKeys.API_VERSIONS, version, "client-id", 0), + apiVersionsRequest, + ByteBufAllocator.DEFAULT.buffer(), + ctx, + new CompletableFuture<>()); + handler.processRequest(request); + + return parseApiVersionsResponse(request); + } + + private static ApiVersionsResponse parseApiVersionsResponse(KafkaRequest request) { + ByteBuf responseBuffer = request.responseBuffer(); + return (ApiVersionsResponse) + AbstractResponse.parseResponse(responseBuffer.nioBuffer(), request.header()); + } + + 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.METADATA.id, ApiKeys.METADATA.oldestVersion(), (short) 11), + tuple( + ApiKeys.API_VERSIONS.id, + ApiKeys.API_VERSIONS.oldestVersion(), + ApiKeys.API_VERSIONS.latestVersion())); + } + + @Test + 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( ApiKeys.API_VERSIONS, latestVersion, new RequestHeader(ApiKeys.API_VERSIONS, latestVersion, "client-id", 0), - apiVersionsRequest, + requestBody, ByteBufAllocator.DEFAULT.buffer(), - ctx, + new TestingChannelHandlerContext(), new CompletableFuture<>()); - handler.handleApiVersionsRequest(request); + + handler.processRequest(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()); - } - }); + 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 = + new KafkaRequest( + ApiKeys.CREATE_TOPICS, + version, + new RequestHeader(ApiKeys.CREATE_TOPICS, version, "client-id", 0), + requestBody, + ByteBufAllocator.DEFAULT.buffer(), + new TestingChannelHandlerContext(), + new CompletableFuture<>()); + + handler.processRequest(request); + + ByteBuf responseBuffer = request.responseBuffer(); + CreateTopicsResponse response = + (CreateTopicsResponse) + AbstractResponse.parseResponse( + responseBuffer.nioBuffer(), request.header()); + assertThat(response.errorCounts()).containsEntry(Errors.UNSUPPORTED_VERSION, 1); } private static KafkaRequestHandler createKafkaRequestHandler() { - return new KafkaRequestHandler(new TestingTabletGatewayService()); + TestingTabletGatewayService service = new TestingTabletGatewayService(); + return new KafkaRequestHandler(service, service, "kafka"); } } diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaTopicAdminHandlerTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaTopicAdminHandlerTest.java new file mode 100644 index 00000000000..c32b8cd0298 --- /dev/null +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaTopicAdminHandlerTest.java @@ -0,0 +1,301 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.kafka; + +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.exception.TableAlreadyExistException; +import org.apache.fluss.kafka.format.KafkaDataFormat; +import org.apache.fluss.metadata.LogFormat; +import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.rpc.TestingTabletGatewayService; +import org.apache.fluss.rpc.gateway.AdminGateway; +import org.apache.fluss.rpc.messages.CreateTableRequest; +import org.apache.fluss.rpc.messages.CreateTableResponse; +import org.apache.fluss.rpc.messages.DropTableRequest; +import org.apache.fluss.rpc.messages.DropTableResponse; +import org.apache.fluss.rpc.messages.GetTableInfoRequest; +import org.apache.fluss.rpc.messages.GetTableInfoResponse; +import org.apache.fluss.shaded.netty4.io.netty.buffer.ByteBuf; +import org.apache.fluss.shaded.netty4.io.netty.buffer.ByteBufAllocator; +import org.apache.fluss.types.DataTypes; + +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.message.CreateTopicsRequestData; +import org.apache.kafka.common.message.CreateTopicsResponseData; +import org.apache.kafka.common.message.DeleteTopicsRequestData; +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.CreateTopicsRequest; +import org.apache.kafka.common.requests.CreateTopicsResponse; +import org.apache.kafka.common.requests.DeleteTopicsRequest; +import org.apache.kafka.common.requests.DeleteTopicsResponse; +import org.apache.kafka.common.requests.RequestHeader; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.concurrent.CompletableFuture; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** Tests the Kafka topic lifecycle mapping to Fluss tables. */ +public class KafkaTopicAdminHandlerTest { + + @Test + public void testCreateTopicCreatesArrowTable() { + TestingTabletGatewayService service = new TestingTabletGatewayService(); + AdminGateway adminGateway = mock(AdminGateway.class); + when(adminGateway.createTable(any(CreateTableRequest.class))) + .thenReturn(CompletableFuture.completedFuture(new CreateTableResponse())); + when(adminGateway.getTableInfo(any(GetTableInfoRequest.class))) + .thenReturn( + CompletableFuture.completedFuture( + new GetTableInfoResponse().setTableId(123L))); + short version = ApiKeys.CREATE_TOPICS.latestVersion(); + CreateTopicsRequest requestBody = createTopicsRequest(version); + KafkaRequest request = kafkaRequest(ApiKeys.CREATE_TOPICS, requestBody, version); + + new KafkaRequestHandler(service, service, adminGateway, "kafka").processRequest(request); + + CreateTopicsResponse response = (CreateTopicsResponse) parseResponse(request); + CreateTopicsResponseData.CreatableTopicResult result = + response.data().topics().find("topic"); + assertThat(result.errorCode()).isEqualTo(Errors.NONE.code()); + assertThat(result.topicId()).isNotEqualTo(Uuid.ZERO_UUID); + ArgumentCaptor captor = + ArgumentCaptor.forClass(CreateTableRequest.class); + verify(adminGateway).createTable(captor.capture()); + CreateTableRequest flussRequest = captor.getValue(); + assertThat(flussRequest.getTablePath().getDatabaseName()).isEqualTo("kafka"); + assertThat(flussRequest.getTablePath().getTableName()).isEqualTo("topic"); + TableDescriptor descriptor = TableDescriptor.fromJsonBytes(flussRequest.getTableJson()); + assertThat(descriptor.getSchema().getColumnNames()) + .containsExactly("record_key", "payload", "event_time", "headers"); + assertThat(descriptor.getSchema().getRowType().getTypeAt(0)).isEqualTo(DataTypes.BYTES()); + assertThat(descriptor.getSchema().getRowType().getTypeAt(1)).isEqualTo(DataTypes.BYTES()); + assertThat(descriptor.getSchema().getRowType().getTypeAt(2)) + .isEqualTo(DataTypes.TIMESTAMP_LTZ(3).copy(false)); + assertThat(descriptor.getSchema().getRowType().getTypeAt(3)) + .isEqualTo( + DataTypes.ARRAY( + DataTypes.ROW( + DataTypes.FIELD("name", DataTypes.STRING().copy(false)), + DataTypes.FIELD("value", DataTypes.BYTES())))); + assertThat(descriptor.getTableDistribution().get().getBucketCount().get()).isEqualTo(3); + assertThat(descriptor.getProperties()) + .containsEntry(ConfigOptions.TABLE_LOG_FORMAT.key(), LogFormat.ARROW.toString()) + .containsEntry(ConfigOptions.TABLE_REPLICATION_FACTOR.key(), "2"); + assertThat(descriptor.getCustomProperties()) + .containsEntry(KafkaDataFormat.KEY_FORMAT_CONFIG, "raw") + .containsEntry(KafkaDataFormat.VALUE_FORMAT_CONFIG, "raw"); + } + + @Test + public void testCreateTopicSupportsIndependentStringFormats() { + TestingTabletGatewayService service = new TestingTabletGatewayService(); + AdminGateway adminGateway = mock(AdminGateway.class); + when(adminGateway.createTable(any(CreateTableRequest.class))) + .thenReturn(CompletableFuture.completedFuture(new CreateTableResponse())); + when(adminGateway.getTableInfo(any(GetTableInfoRequest.class))) + .thenReturn( + CompletableFuture.completedFuture( + new GetTableInfoResponse().setTableId(123L))); + Map configs = new LinkedHashMap<>(); + configs.put(KafkaDataFormat.KEY_FORMAT_CONFIG, "string"); + configs.put(KafkaDataFormat.VALUE_FORMAT_CONFIG, "raw"); + short version = ApiKeys.CREATE_TOPICS.latestVersion(); + KafkaRequest request = + kafkaRequest(ApiKeys.CREATE_TOPICS, createTopicsRequest(version, configs), version); + + new KafkaRequestHandler(service, service, adminGateway, "kafka").processRequest(request); + + assertThat(((CreateTopicsResponse) parseResponse(request)).errorCounts()) + .containsOnlyKeys(Errors.NONE); + ArgumentCaptor captor = + ArgumentCaptor.forClass(CreateTableRequest.class); + verify(adminGateway).createTable(captor.capture()); + TableDescriptor descriptor = + TableDescriptor.fromJsonBytes(captor.getValue().getTableJson()); + assertThat(descriptor.getSchema().getRowType().getTypeAt(0)).isEqualTo(DataTypes.STRING()); + assertThat(descriptor.getSchema().getRowType().getTypeAt(1)).isEqualTo(DataTypes.BYTES()); + assertThat(descriptor.getCustomProperties()).containsAllEntriesOf(configs); + } + + @Test + public void testCreateTopicUsesConfiguredDefaultFormats() { + TestingTabletGatewayService service = new TestingTabletGatewayService(); + AdminGateway adminGateway = mock(AdminGateway.class); + when(adminGateway.createTable(any(CreateTableRequest.class))) + .thenReturn(CompletableFuture.completedFuture(new CreateTableResponse())); + when(adminGateway.getTableInfo(any(GetTableInfoRequest.class))) + .thenReturn( + CompletableFuture.completedFuture( + new GetTableInfoResponse().setTableId(123L))); + short version = ApiKeys.CREATE_TOPICS.latestVersion(); + KafkaRequest request = + kafkaRequest(ApiKeys.CREATE_TOPICS, createTopicsRequest(version), version); + + new KafkaRequestHandler( + service, + service, + adminGateway, + "kafka", + KafkaDataFormat.STRING, + KafkaDataFormat.STRING) + .processRequest(request); + + assertThat(((CreateTopicsResponse) parseResponse(request)).errorCounts()) + .containsOnlyKeys(Errors.NONE); + ArgumentCaptor captor = + ArgumentCaptor.forClass(CreateTableRequest.class); + verify(adminGateway).createTable(captor.capture()); + TableDescriptor descriptor = + TableDescriptor.fromJsonBytes(captor.getValue().getTableJson()); + assertThat(descriptor.getSchema().getRowType().getTypeAt(0)).isEqualTo(DataTypes.STRING()); + assertThat(descriptor.getSchema().getRowType().getTypeAt(1)).isEqualTo(DataTypes.STRING()); + assertThat(descriptor.getCustomProperties()) + .containsEntry(KafkaDataFormat.KEY_FORMAT_CONFIG, "string") + .containsEntry(KafkaDataFormat.VALUE_FORMAT_CONFIG, "string"); + } + + @Test + public void testCreateTopicRejectsInvalidFormat() { + TestingTabletGatewayService service = new TestingTabletGatewayService(); + AdminGateway adminGateway = mock(AdminGateway.class); + short version = ApiKeys.CREATE_TOPICS.latestVersion(); + KafkaRequest request = + kafkaRequest( + ApiKeys.CREATE_TOPICS, + createTopicsRequest( + version, + Collections.singletonMap( + KafkaDataFormat.VALUE_FORMAT_CONFIG, "json")), + version); + + new KafkaRequestHandler(service, service, adminGateway, "kafka").processRequest(request); + + CreateTopicsResponse response = (CreateTopicsResponse) parseResponse(request); + assertThat(response.errorCounts()).containsEntry(Errors.INVALID_CONFIG, 1); + assertThat(response.data().topics().find("topic").errorMessage()) + .contains("Expected raw or string"); + verify(adminGateway, never()).createTable(any(CreateTableRequest.class)); + } + + @Test + public void testCreateTopicMapsAlreadyExists() { + TestingTabletGatewayService service = new TestingTabletGatewayService(); + AdminGateway adminGateway = mock(AdminGateway.class); + CompletableFuture failure = new CompletableFuture<>(); + failure.completeExceptionally(new TableAlreadyExistException("already exists")); + when(adminGateway.createTable(any(CreateTableRequest.class))).thenReturn(failure); + short version = ApiKeys.CREATE_TOPICS.latestVersion(); + KafkaRequest request = + kafkaRequest(ApiKeys.CREATE_TOPICS, createTopicsRequest(version), version); + + new KafkaRequestHandler(service, service, adminGateway, "kafka").processRequest(request); + + CreateTopicsResponse response = (CreateTopicsResponse) parseResponse(request); + assertThat(response.errorCounts()).containsEntry(Errors.TOPIC_ALREADY_EXISTS, 1); + } + + @Test + public void testDeleteTopicDropsTable() { + TestingTabletGatewayService service = new TestingTabletGatewayService(); + AdminGateway adminGateway = mock(AdminGateway.class); + when(adminGateway.dropTable(any(DropTableRequest.class))) + .thenReturn(CompletableFuture.completedFuture(new DropTableResponse())); + short version = ApiKeys.DELETE_TOPICS.latestVersion(); + DeleteTopicsRequestData data = + new DeleteTopicsRequestData() + .setTimeoutMs(1000) + .setTopics( + Collections.singletonList( + new DeleteTopicsRequestData.DeleteTopicState() + .setName("topic") + .setTopicId(Uuid.ZERO_UUID))); + DeleteTopicsRequest requestBody = new DeleteTopicsRequest.Builder(data).build(version); + KafkaRequest request = kafkaRequest(ApiKeys.DELETE_TOPICS, requestBody, version); + + new KafkaRequestHandler(service, service, adminGateway, "kafka").processRequest(request); + + DeleteTopicsResponse response = (DeleteTopicsResponse) parseResponse(request); + assertThat(response.errorCounts()).containsOnlyKeys(Errors.NONE); + ArgumentCaptor captor = ArgumentCaptor.forClass(DropTableRequest.class); + verify(adminGateway).dropTable(captor.capture()); + assertThat(captor.getValue().getTablePath().getDatabaseName()).isEqualTo("kafka"); + assertThat(captor.getValue().getTablePath().getTableName()).isEqualTo("topic"); + } + + private static CreateTopicsRequest createTopicsRequest(short version) { + return createTopicsRequest(version, Collections.emptyMap()); + } + + private static CreateTopicsRequest createTopicsRequest( + short version, Map configs) { + CreateTopicsRequestData.CreatableTopic topic = + new CreateTopicsRequestData.CreatableTopic() + .setName("topic") + .setNumPartitions(3) + .setReplicationFactor((short) 2); + for (Map.Entry config : configs.entrySet()) { + topic.configs() + .add( + new CreateTopicsRequestData.CreatableTopicConfig() + .setName(config.getKey()) + .setValue(config.getValue())); + } + CreateTopicsRequestData data = + new CreateTopicsRequestData() + .setTimeoutMs(1000) + .setTopics( + new CreateTopicsRequestData.CreatableTopicCollection( + Collections.singletonList(topic).iterator())); + return new CreateTopicsRequest.Builder(data).build(version); + } + + private static KafkaRequest kafkaRequest( + ApiKeys apiKey, AbstractRequest requestBody, short version) { + return new KafkaRequest( + apiKey, + version, + new RequestHeader(apiKey, version, "client-id", 1), + requestBody, + "KAFKA", + ByteBufAllocator.DEFAULT.buffer(), + new TestingChannelHandlerContext(), + new CompletableFuture<>()); + } + + private static AbstractResponse parseResponse(KafkaRequest request) { + ByteBuf responseBuffer = request.responseBuffer(); + try { + return AbstractResponse.parseResponse(responseBuffer.nioBuffer(), request.header()); + } finally { + responseBuffer.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/mapping/KafkaTopicMapperTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/mapping/KafkaTopicMapperTest.java new file mode 100644 index 00000000000..a0a2036d4d5 --- /dev/null +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/mapping/KafkaTopicMapperTest.java @@ -0,0 +1,43 @@ +/* + * 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.mapping; + +import org.apache.kafka.common.Uuid; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link KafkaTopicMapper}. */ +public class KafkaTopicMapperTest { + + @Test + public void testTopicNameAndIdMapping() { + KafkaTopicMapper mapper = new KafkaTopicMapper("kafka"); + + assertThat(mapper.toTablePath("topic").toString()).isEqualTo("kafka.topic"); + Uuid topicId = mapper.toTopicId(123L); + assertThat(topicId).isNotIn(Uuid.ZERO_UUID, Uuid.ONE_UUID, Uuid.METADATA_TOPIC_ID); + assertThat(mapper.isMappedTopicId(topicId)).isTrue(); + assertThat(mapper.toTableId(topicId)).isEqualTo(123L); + + Uuid firstTableTopicId = mapper.toTopicId(0L); + assertThat(firstTableTopicId).isNotEqualTo(Uuid.ZERO_UUID); + assertThat(mapper.isMappedTopicId(firstTableTopicId)).isTrue(); + assertThat(mapper.toTableId(firstTableTopicId)).isZero(); + } +} diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminGatewayProvider.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminGatewayProvider.java new file mode 100644 index 00000000000..901060cfa1b --- /dev/null +++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminGatewayProvider.java @@ -0,0 +1,28 @@ +/* + * 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.rpc.gateway; + +import org.apache.fluss.annotation.Internal; + +/** Provides the admin gateway used by a server service for delegated metadata mutations. */ +@Internal +public interface AdminGatewayProvider { + + /** Returns the admin gateway available to the server service. */ + AdminGateway getAdminGateway(); +} 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 diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java index bd3ef49b35a..4230637bdab 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java @@ -37,6 +37,7 @@ import org.apache.fluss.rpc.entity.LookupResultForBucket; import org.apache.fluss.rpc.entity.PrefixLookupResultForBucket; import org.apache.fluss.rpc.entity.ResultForBucket; +import org.apache.fluss.rpc.gateway.AdminGatewayProvider; import org.apache.fluss.rpc.gateway.CoordinatorGateway; import org.apache.fluss.rpc.gateway.TabletServerGateway; import org.apache.fluss.rpc.messages.FetchLogRequest; @@ -160,7 +161,8 @@ import static org.apache.fluss.server.utils.ServerRpcMessageUtils.toPutKvDataForBuckets; /** An RPC Gateway service for tablet server. */ -public final class TabletService extends RpcServiceBase implements TabletServerGateway { +public final class TabletService extends RpcServiceBase + implements TabletServerGateway, AdminGatewayProvider { private final String serviceName; private final ReplicaManager replicaManager; @@ -209,6 +211,14 @@ public String name() { return serviceName; } + /** + * Returns the coordinator admin gateway used by this tablet service for internal forwarding. + */ + @Override + public CoordinatorGateway getAdminGateway() { + return coordinatorGateway; + } + @Override public void shutdown() {}