[kafka] Separate value decoding and preserve record metadata

fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java

@@ -19,8 +19,10 @@
 
 import org.apache.fluss.client.Connection;
 import org.apache.fluss.client.ConnectionFactory;
+import org.apache.fluss.client.FlussConnection;
 import org.apache.fluss.client.metadata.KvSnapshotMetadata;
 import org.apache.fluss.client.metadata.KvSnapshots;
+import org.apache.fluss.client.metadata.MetadataUpdater;
 import org.apache.fluss.client.metadata.TestingMetadataUpdater;
 import org.apache.fluss.client.table.Table;
 import org.apache.fluss.client.table.writer.AppendWriter;
@@ -1211,19 +1213,24 @@ void testKvSnapshotLeaseAfterCoordinatorServerRestart() throws Exception {
         assertThat(zkClient.getKvSnapshotLeaseMetadata(lease.leaseId())).isPresent();
 
         restartCoordinatorServer(zkClient);
-        FLUSS_CLUSTER_EXTENSION.waitUntilAllGatewayHasSameMetadata();
-
         lease.dropLease().get();
         assertThat(zkClient.getKvSnapshotLeaseMetadata(lease.leaseId())).isNotPresent();
     }
 
     private void restartCoordinatorServer(ZooKeeperClient zkClient) throws Exception {
+        MetadataUpdater metadataUpdater = ((FlussConnection) conn).getMetadataUpdater();
+        ServerNode cachedCoordinator = metadataUpdater.getCoordinatorServer();
         FLUSS_CLUSTER_EXTENSION.stopCoordinatorServer();
         waitUntil(
                 () -> !zkClient.getCoordinatorLeaderAddress().isPresent(),
                 Duration.ofMinutes(1),
                 "Coordinator server node still exists in ZooKeeper");
         FLUSS_CLUSTER_EXTENSION.startCoordinatorServer();
+        // Election completes before the new coordinator address reaches every tablet server.
+        // Wait for server metadata to converge so that the client's retry can discover it.
+        FLUSS_CLUSTER_EXTENSION.waitUntilAllGatewayHasSameMetadata();
+        // Keep the client cache stale to exercise the lease's metadata refresh and retry.
+        assertThat(metadataUpdater.getCoordinatorServer()).isEqualTo(cachedCoordinator);
     }
 
     @Test
@@ -2831,6 +2838,7 @@ void testKvSnapshotLeaseForLogTable() throws Exception {
         TableDescriptor logTableDescriptor =
                 TableDescriptor.builder().schema(logSchema).distributedBy(3).build();
         long tableId = createTable(logTablePath, logTableDescriptor, true);
+        FLUSS_CLUSTER_EXTENSION.waitUntilTableReady(tableId);
 
         // Create a KvSnapshotLease
         KvSnapshotLease lease = admin.createKvSnapshotLease("test-lease-log", 60000);

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

@@ -2269,6 +2269,38 @@ public class ConfigOptions {
                                     + "Downstream consumers must be upgraded to Fluss v1.0+ before enabling this option, "
                                     + "as older versions cannot parse the extended batch format.");
 
+    public static final ConfigOption<Boolean> TABLE_KAFKA_ENABLED =
+            key("table.kafka.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether the table is exposed as a Kafka topic. Enabling validates the "
+                                    + "existing schema and conversion options without adding columns or defaults.");
+
+    public static final ConfigOption<String> TABLE_KAFKA_CONVERT_VALUE_TYPE =
+            key("table.kafka.convert.value.type")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "Required when Kafka compatibility is enabled: binary, string, or parse_json. "
+                                    + "Binary and string target the declared _kafka_value column.");
+
+    public static final ConfigOption<String> TABLE_KAFKA_CONVERT_VALUE_LAYOUT =
+            key("table.kafka.convert.value.layout")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "JSON layout: flat (the read-time default for parse_json) or nested. "
+                                    + "Nested targets _kafka_value ROW. Not allowed for binary or string.");
+
+    public static final ConfigOption<String> TABLE_KAFKA_CONVERT_KEY_TYPE =
+            key("table.kafka.convert.key.type")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "Optional Kafka key conversion type. Only binary is supported and requires "
+                                    + "_kafka_key BYTES. Declaring that column implies binary without persisting a default.");
+
     // ------------------------------------------------------------------------
     //  ConfigOptions for Kv
     // ------------------------------------------------------------------------
@@ -2905,12 +2937,15 @@ public class ConfigOptions {
                     .withDescription(
                             "The listener names for Kafka wire protocol communication. Support multiple listener names, separated by comma.");
 
-    public static final ConfigOption<String> KAFKA_DATABASE =
-            key("kafka.database")
+    public static final ConfigOption<String> KAFKA_DEFAULT_DATABASE =
+            key("kafka.default.database")
                     .stringType()
-                    .defaultValue("kafka")
+                    .defaultValue("fluss")
                     .withDescription(
-                            "The database for fluss kafka. The default database is `kafka`.");
+                            "Default database for unqualified Kafka topic names. Fully qualified names may "
+                                    + "access other databases. When Kafka is enabled, a configured database other "
+                                    + "than 'fluss' is created if missing. The native 'fluss' database is created "
+                                    + "only when the cluster has no databases.");
 
     /**
      * Compaction style for Fluss's kv, which is same to rocksdb's, but help use avoid including

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

@@ -47,6 +47,10 @@ public class FlussConfigUtils {
         CLIENT_OPTIONS = extractConfigOptions("client.");
         ALTERABLE_TABLE_OPTIONS =
                 Arrays.asList(
+                        ConfigOptions.TABLE_KAFKA_ENABLED.key(),
+                        ConfigOptions.TABLE_KAFKA_CONVERT_VALUE_TYPE.key(),
+                        ConfigOptions.TABLE_KAFKA_CONVERT_VALUE_LAYOUT.key(),
+                        ConfigOptions.TABLE_KAFKA_CONVERT_KEY_TYPE.key(),
                         ConfigOptions.TABLE_DATALAKE_ENABLED.key(),
                         ConfigOptions.TABLE_DATALAKE_HISTORICAL_PARTITION_ENABLED.key(),
                         ConfigOptions.TABLE_DATALAKE_HISTORICAL_PARTITION_LOOKUP_MODE.key(),

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

@@ -54,6 +54,11 @@ public TableConfig(Configuration config) {
         this.config = config;
     }
 
+    /** Whether the table explicitly enables Kafka compatibility. */
+    public boolean isKafkaEnabled() {
+        return config.get(ConfigOptions.TABLE_KAFKA_ENABLED);
+    }
+
     /** Gets the replication factor of the table. */
     public int getReplicationFactor() {
         return config.get(ConfigOptions.TABLE_REPLICATION_FACTOR);

fluss-common/src/main/java/org/apache/fluss/metadata/KafkaFieldProjection.java

@@ -0,0 +1,123 @@
+/*
+ * 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.metadata;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.types.DataType;
+import org.apache.fluss.types.RowType;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/** Ordered physical Fluss fields populated by one Kafka record component. */
+@Internal
+public final class KafkaFieldProjection {
+
+    private final List<Integer> positions;
+    private final List<String> names;
+    private final List<DataType> dataTypes;
+
+    /** Creates a projection from physical row positions. */
+    public KafkaFieldProjection(RowType rowType, List<Integer> positions) {
+        checkNotNull(rowType);
+        checkNotNull(positions);
+        List<Integer> positionCopy = new ArrayList<>(positions.size());
+        List<String> projectedNames = new ArrayList<>(positions.size());
+        List<DataType> projectedTypes = new ArrayList<>(positions.size());
+        for (Integer position : positions) {
+            checkArgument(
+                    position != null && position >= 0 && position < rowType.getFieldCount(),
+                    "Invalid Kafka field projection position %s.",
+                    position);
+            positionCopy.add(position);
+            projectedNames.add(rowType.getFields().get(position).getName());
+            projectedTypes.add(rowType.getTypeAt(position));
+        }
+        this.positions = Collections.unmodifiableList(positionCopy);
+        this.names = Collections.unmodifiableList(projectedNames);
+        this.dataTypes = Collections.unmodifiableList(projectedTypes);
+    }
+
+    /** Returns the number of projected fields. */
+    public int size() {
+        return positions.size();
+    }
+
+    /** Returns whether this projection owns no fields. */
+    public boolean isEmpty() {
+        return positions.isEmpty();
+    }
+
+    /** Returns the physical row position at the projection position. */
+    public int positionAt(int projectionPosition) {
+        return positions.get(projectionPosition);
+    }
+
+    /** Returns the physical field name at the projection position. */
+    public String nameAt(int projectionPosition) {
+        return names.get(projectionPosition);
+    }
+
+    /** Returns the physical data type at the projection position. */
+    public DataType dataTypeAt(int projectionPosition) {
+        return dataTypes.get(projectionPosition);
+    }
+
+    /** Returns the projected physical positions. */
+    public List<Integer> positions() {
+        return positions;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (!(obj instanceof KafkaFieldProjection)) {
+            return false;
+        }
+        KafkaFieldProjection that = (KafkaFieldProjection) obj;
+        return Objects.equals(positions, that.positions)
+                && Objects.equals(names, that.names)
+                && Objects.equals(dataTypes, that.dataTypes);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(positions, names, dataTypes);
+    }
+
+    @Override
+    public String toString() {
+        return "KafkaFieldProjection{"
+                + "positions="
+                + positions
+                + ", "
+                + "names="
+                + names
+                + ", "
+                + "dataTypes="
+                + dataTypes
+                + "}";
+    }
+}