[kafka] Define DDL table mapping for Kafka compatibility

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

@@ -2831,6 +2831,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,44 @@ 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 to expose this table as a Kafka topic. Requires a non-partitioned "
+                                    + "log table using the Arrow log format.");
+
+    public static final ConfigOption<KafkaValueConversionType> TABLE_KAFKA_CONVERT_VALUE_TYPE =
+            key("table.kafka.convert.value.type")
+                    .enumType(KafkaValueConversionType.class)
+                    .defaultValue(KafkaValueConversionType.PARSE_JSON)
+                    .withDescription(
+                            "How to convert Kafka record values. 'binary' stores the raw bytes in "
+                                    + "_kafka_value (BYTES); 'string' decodes UTF-8 text into _kafka_value "
+                                    + "(STRING); 'parse_json' parses a JSON object and maps its fields "
+                                    + "according to table.kafka.convert.value.layout. Defaults to 'parse_json'.");
+
+    public static final ConfigOption<KafkaValueConversionLayout> TABLE_KAFKA_CONVERT_VALUE_LAYOUT =
+            key("table.kafka.convert.value.layout")
+                    .enumType(KafkaValueConversionLayout.class)
+                    .defaultValue(KafkaValueConversionLayout.FLAT)
+                    .withDescription(
+                            "How to map JSON object fields when table.kafka.convert.value.type is "
+                                    + "'parse_json'. 'flat' maps fields by name to top-level table columns; "
+                                    + "'nested' maps them to fields in the _kafka_value column, which must "
+                                    + "have ROW type. Defaults to 'flat'. Ignored for 'binary' and 'string' conversion.");
+
+    public static final ConfigOption<KafkaKeyConversionType> TABLE_KAFKA_CONVERT_KEY_TYPE =
+            key("table.kafka.convert.key.type")
+                    .enumType(KafkaKeyConversionType.class)
+                    .noDefaultValue()
+                    .withDescription(
+                            "How to convert Kafka record keys. Only 'binary' is supported, which stores "
+                                    + "raw key bytes in the _kafka_key column of type BYTES. If this column "
+                                    + "is declared, keys are stored even when this option is omitted. "
+                                    + "Without this column, keys are not stored and this option must be left unset.");
+
     // ------------------------------------------------------------------------
     //  ConfigOptions for Consumer Offsets
     // ------------------------------------------------------------------------
@@ -2947,12 +2985,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, the coordinator creates "
+                                    + "the configured database if missing, including 'fluss'. With Kafka disabled, "
+                                    + "native initialization creates 'fluss' 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/KafkaKeyConversionType.java

@@ -0,0 +1,27 @@
+/*
+ * 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.config;
+
+import org.apache.fluss.annotation.PublicEvolving;
+
+/** Conversion applied to Kafka record keys. */
+@PublicEvolving
+public enum KafkaKeyConversionType {
+    /** Store the key as bytes in the fixed key column. */
+    BINARY
+}

fluss-common/src/main/java/org/apache/fluss/config/KafkaValueConversionLayout.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.config;
+
+import org.apache.fluss.annotation.PublicEvolving;
+
+/** Mapping of JSON object members to Fluss table columns. */
+@PublicEvolving
+public enum KafkaValueConversionLayout {
+    /** Map members to top-level business columns. */
+    FLAT,
+
+    /** Map members to fields inside the fixed value ROW column. */
+    NESTED
+}

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

@@ -0,0 +1,33 @@
+/*
+ * 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.config;
+
+import org.apache.fluss.annotation.PublicEvolving;
+
+/** Conversion applied to Kafka record values. */
+@PublicEvolving
+public enum KafkaValueConversionType {
+    /** Store the value as bytes in the fixed value column. */
+    BINARY,
+
+    /** Decode the value as a UTF-8 string in the fixed value column. */
+    STRING,
+
+    /** Parse the JSON object using the table schema and configured value layout. */
+    PARSE_JSON
+}

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
+                + "}";
+    }
+}