diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java index e67faf84b24..b39739c6b01 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java +++ b/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); 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 f02fdbe0066..8f4dad6d777 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 @@ -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 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 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 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 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 KAFKA_DATABASE = - key("kafka.database") + public static final ConfigOption 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 diff --git a/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java b/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java index 37d5cc9f88e..415103c46ea 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java +++ b/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(), diff --git a/fluss-common/src/main/java/org/apache/fluss/config/KafkaKeyConversionType.java b/fluss-common/src/main/java/org/apache/fluss/config/KafkaKeyConversionType.java new file mode 100644 index 00000000000..5c0ebbdc05d --- /dev/null +++ b/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 +} diff --git a/fluss-common/src/main/java/org/apache/fluss/config/KafkaValueConversionLayout.java b/fluss-common/src/main/java/org/apache/fluss/config/KafkaValueConversionLayout.java new file mode 100644 index 00000000000..e1cb4f3aadb --- /dev/null +++ b/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 +} diff --git a/fluss-common/src/main/java/org/apache/fluss/config/KafkaValueConversionType.java b/fluss-common/src/main/java/org/apache/fluss/config/KafkaValueConversionType.java new file mode 100644 index 00000000000..efa57982b7c --- /dev/null +++ b/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 +} diff --git a/fluss-common/src/main/java/org/apache/fluss/metadata/KafkaFieldProjection.java b/fluss-common/src/main/java/org/apache/fluss/metadata/KafkaFieldProjection.java new file mode 100644 index 00000000000..f8946107020 --- /dev/null +++ b/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 positions; + private final List names; + private final List dataTypes; + + /** Creates a projection from physical row positions. */ + public KafkaFieldProjection(RowType rowType, List positions) { + checkNotNull(rowType); + checkNotNull(positions); + List positionCopy = new ArrayList<>(positions.size()); + List projectedNames = new ArrayList<>(positions.size()); + List 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 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 + + "}"; + } +} diff --git a/fluss-common/src/main/java/org/apache/fluss/metadata/KafkaTableSchema.java b/fluss-common/src/main/java/org/apache/fluss/metadata/KafkaTableSchema.java new file mode 100644 index 00000000000..0005111a4d1 --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/metadata/KafkaTableSchema.java @@ -0,0 +1,449 @@ +/* + * 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.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.KafkaValueConversionLayout; +import org.apache.fluss.config.KafkaValueConversionType; +import org.apache.fluss.config.TableConfig; +import org.apache.fluss.exception.InvalidTableException; +import org.apache.fluss.types.ArrayType; +import org.apache.fluss.types.BytesType; +import org.apache.fluss.types.DataField; +import org.apache.fluss.types.DataType; +import org.apache.fluss.types.DataTypes; +import org.apache.fluss.types.MapType; +import org.apache.fluss.types.RowType; +import org.apache.fluss.types.StringType; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Objects; +import java.util.Set; + +/** + * Immutable Kafka record mapping, shared by native DDL validation and Kafka request handling. + * Resolving a contract never changes the schema or materializes configuration defaults. + */ +@Internal +public final class KafkaTableSchema { + + public static final String KEY_COLUMN = "_kafka_key"; + public static final String VALUE_COLUMN = "_kafka_value"; + public static final String HEADERS_COLUMN = "_kafka_headers"; + public static final String METADATA_COLUMN = "_kafka_metadata"; + public static final String RESCUE_COLUMN = "_kafka_rescue"; + + /** Maximum recursive depth below each top-level JSON business field. */ + public static final int MAX_JSON_SCHEMA_NESTING_DEPTH = 64; + + private static final Set RESERVED_NAMES = + Collections.unmodifiableSet( + new HashSet<>( + Arrays.asList( + KEY_COLUMN, + VALUE_COLUMN, + HEADERS_COLUMN, + METADATA_COLUMN, + RESCUE_COLUMN))); + + private final RowType rowType; + private final KafkaValueConversionType valueConversionType; + private final @Nullable KafkaValueConversionLayout valueConversionLayout; + private final KafkaFieldProjection valueProjection; + private final int keyPosition; + private final int headersPosition; + private final int metadataPosition; + private final int rescuePosition; + + private KafkaTableSchema( + RowType rowType, + KafkaValueConversionType valueConversionType, + @Nullable KafkaValueConversionLayout valueConversionLayout, + List valuePositions, + int keyPosition, + int headersPosition, + int metadataPosition, + int rescuePosition) { + this.rowType = rowType; + this.valueConversionType = valueConversionType; + this.valueConversionLayout = valueConversionLayout; + this.valueProjection = new KafkaFieldProjection(rowType, valuePositions); + this.keyPosition = keyPosition; + this.headersPosition = headersPosition; + this.metadataPosition = metadataPosition; + this.rescuePosition = rescuePosition; + } + + /** Returns only the persisted opt-in flag; schema compatibility does not imply enablement. */ + public static boolean isEnabled(TableDescriptor table) { + String value = table.getProperties().get(ConfigOptions.TABLE_KAFKA_ENABLED.key()); + if (value == null) { + return ConfigOptions.TABLE_KAFKA_ENABLED.defaultValue(); + } + if ("true".equalsIgnoreCase(value)) { + return true; + } + if ("false".equalsIgnoreCase(value)) { + return false; + } + throw new IllegalArgumentException( + "Invalid value for table.kafka.enabled: '" + + value + + "'. Expected true or false (case insensitive)."); + } + + /** Validates the Kafka contract only when the final table state enables it. */ + public static void validate(TableDescriptor table) { + if (isEnabled(table)) { + resolve(table); + } + } + + /** Checks whether the canonical topic name satisfies Fluss and Kafka name restrictions. */ + public static boolean isValidTablePath(TablePath path) { + // Fluss identifiers are already restricted to Kafka's ASCII character set, without dots. + return path != null + && path.isValid() + && TablePath.validatePrefix(path.getDatabaseName()) == null + && TablePath.validatePrefix(path.getTableName()) == null + && path.toString().length() <= 249; + } + + /** Validates the canonical topic name before persisting an enabled table. */ + public static void validateTablePath(TablePath path, TableDescriptor table) { + if (isEnabled(table) && !isValidTablePath(path)) { + throw invalid("Table path cannot be represented as a Kafka topic: " + path); + } + } + + /** Returns whether a name is reserved at the physical table's top level. */ + public static boolean isReservedName(String name) { + return RESERVED_NAMES.contains(name); + } + + /** Resolves an explicitly enabled table or rejects its invalid contract. */ + public static KafkaTableSchema resolve(TableDescriptor table) { + if (!isEnabled(table)) { + throw invalid("Kafka compatibility is disabled; table.kafka.enabled must be true."); + } + // The initial Kafka write path encodes Arrow batches and appends them to log tables. + // Other log formats require matching encoders before they can be enabled here. + if (table.hasPrimaryKey() + || table.isPartitioned() + || new TableConfig(Configuration.fromMap(table.getProperties())).getLogFormat() + != LogFormat.ARROW) { + throw invalid( + "Kafka requires a non-partitioned Arrow log table without a primary key."); + } + Configuration options = Configuration.fromMap(table.getProperties()); + KafkaValueConversionType valueConversionType = + options.get(ConfigOptions.TABLE_KAFKA_CONVERT_VALUE_TYPE); + KafkaValueConversionLayout valueConversionLayout = + valueConversionType == KafkaValueConversionType.PARSE_JSON + ? options.get(ConfigOptions.TABLE_KAFKA_CONVERT_VALUE_LAYOUT) + : null; + + RowType rowType = table.getSchema().getRowType(); + int keyPosition = rowType.getFieldIndex(KEY_COLUMN); + if (keyPosition >= 0 && !(rowType.getTypeAt(keyPosition) instanceof BytesType)) { + throw invalid("_kafka_key must be BYTES."); + } + if (options.getOptional(ConfigOptions.TABLE_KAFKA_CONVERT_KEY_TYPE).isPresent() + && keyPosition < 0) { + throw invalid("table.kafka.convert.key.type requires _kafka_key BYTES."); + } + int headersPosition = rowType.getFieldIndex(HEADERS_COLUMN); + if (headersPosition >= 0) { + DataType headers = rowType.getTypeAt(headersPosition); + if (!(headers instanceof MapType) + || !(((MapType) headers).getKeyType() instanceof StringType) + || !(((MapType) headers).getValueType() instanceof BytesType)) { + throw invalid("_kafka_headers must be MAP."); + } + } + int rescuePosition = rowType.getFieldIndex(RESCUE_COLUMN); + if (rescuePosition >= 0 + && (!(rowType.getTypeAt(rescuePosition) instanceof StringType) + || !rowType.getTypeAt(rescuePosition).isNullable())) { + throw invalid("_kafka_rescue must be nullable STRING."); + } + int metadataPosition = rowType.getFieldIndex(METADATA_COLUMN); + if (metadataPosition >= 0) { + validateMetadataType(rowType.getTypeAt(metadataPosition)); + } + + List valuePositions = new ArrayList<>(); + if (valueConversionLayout == KafkaValueConversionLayout.FLAT) { + if (rowType.getFieldIndex(VALUE_COLUMN) >= 0) { + throw invalid( + "_kafka_value is reserved for binary, string or nested JSON; " + + "it cannot be a flat JSON business field."); + } + for (int i = 0; i < rowType.getFieldCount(); i++) { + if (!isReservedName(rowType.getFields().get(i).getName())) { + valuePositions.add(i); + } + } + } else { + int valuePosition = rowType.getFieldIndex(VALUE_COLUMN); + if (valuePosition < 0) { + throw invalid("Missing required value column _kafka_value."); + } + DataType type = rowType.getTypeAt(valuePosition); + boolean valid = + valueConversionType == KafkaValueConversionType.BINARY + ? type instanceof BytesType + : valueConversionType == KafkaValueConversionType.STRING + ? type instanceof StringType + : type instanceof RowType; + if (!valid) { + throw invalid( + "_kafka_value must be " + + (valueConversionType == KafkaValueConversionType.BINARY + ? "BYTES" + : valueConversionType == KafkaValueConversionType.STRING + ? "STRING" + : "ROW") + + " for " + + valueConversionType + + "."); + } + for (String name : rowType.getFieldNames()) { + if (!isReservedName(name)) { + throw invalid( + "Column '" + + name + + "' has no value mapping; this conversion targets _kafka_value only."); + } + } + valuePositions.add(valuePosition); + } + if (valueConversionType == KafkaValueConversionType.PARSE_JSON) { + for (int position : valuePositions) { + DataType type = rowType.getTypeAt(position); + if (valueConversionLayout == KafkaValueConversionLayout.NESTED) { + // The physical value container represents the JSON object root, just like + // the table row in flat layout; it must not add a business nesting level. + for (DataType field : ((RowType) type).getChildren()) { + validateJsonType(field); + } + } else { + validateJsonType(type); + } + } + } + for (DataField field : rowType.getFields()) { + validateNullable(field.getType(), field.getName()); + } + return new KafkaTableSchema( + rowType, + valueConversionType, + valueConversionLayout, + valuePositions, + keyPosition, + headersPosition, + metadataPosition, + rescuePosition); + } + + /** + * Validates a top-level JSON business field without parser or protocol dependencies. DDL and + * decoder construction share this contract. All currently defined Fluss types are supported + * except MAP types with non-STRING keys (including nested maps) and schemas nested deeper than + * 64 levels. Each business field starts at depth zero; ROW fields, ARRAY elements and MAP + * values each advance the depth by one. Table resolution separately requires nullable fields; + * scalar value constraints are enforced when decoding individual records. The JSON object root + * (including the physical nested _kafka_value ROW) is not counted. + */ + public static void validateJsonType(DataType type) { + validateJsonType(type, 0); + } + + private static void validateJsonType(DataType type, int depth) { + if (depth > MAX_JSON_SCHEMA_NESTING_DEPTH) { + throw invalid( + "Kafka JSON schema exceeds the maximum nesting depth of " + + MAX_JSON_SCHEMA_NESTING_DEPTH + + "."); + } + switch (type.getTypeRoot()) { + case CHAR: + case STRING: + case BOOLEAN: + case BINARY: + case BYTES: + case DECIMAL: + case TINYINT: + case SMALLINT: + case INTEGER: + case BIGINT: + case FLOAT: + case DOUBLE: + case DATE: + case TIME_WITHOUT_TIME_ZONE: + case TIMESTAMP_WITHOUT_TIME_ZONE: + case TIMESTAMP_WITH_LOCAL_TIME_ZONE: + return; + case ARRAY: + validateJsonType(((ArrayType) type).getElementType(), depth + 1); + return; + case MAP: + MapType map = (MapType) type; + if (!(map.getKeyType() instanceof StringType)) { + throw invalid( + "Kafka JSON format only supports STRING map keys, but found " + + type.asSummaryString() + + "."); + } + validateJsonType(map.getValueType(), depth + 1); + return; + case ROW: + for (DataType field : ((RowType) type).getChildren()) { + validateJsonType(field, depth + 1); + } + return; + default: + throw invalid( + "Kafka JSON format does not support Fluss type " + + type.asSummaryString() + + "."); + } + } + + private static void validateNullable(DataType type, String path) { + if (!type.isNullable()) { + throw invalid("Kafka requires nullable fields, but '" + path + "' is NOT NULL."); + } + if (type instanceof RowType) { + for (DataField field : ((RowType) type).getFields()) { + validateNullable(field.getType(), path + "." + field.getName()); + } + } else if (type instanceof ArrayType) { + validateNullable(((ArrayType) type).getElementType(), path + "[]"); + } else if (type instanceof MapType) { + // Fluss requires non-null MAP keys. Only values can be nulled by field-level rescue. + validateNullable(((MapType) type).getValueType(), path + ""); + } + } + + private static void validateMetadataType(DataType type) { + String message = + "_kafka_metadata must be ROW."; + if (!(type instanceof RowType)) { + throw invalid(message); + } + RowType row = (RowType) type; + if (!row.getFieldNames().equals(Arrays.asList("timestamp", "timestamp_type", "partition")) + || !row.getTypeAt(0).copy(true).equals(DataTypes.TIMESTAMP_LTZ(3)) + || !row.getTypeAt(1).copy(true).equals(DataTypes.INT()) + || !row.getTypeAt(2).copy(true).equals(DataTypes.INT())) { + throw invalid(message); + } + } + + /** Returns the unchanged physical schema, including declared nullability. */ + public RowType rowType() { + return rowType; + } + + /** Returns the value conversion type. */ + public KafkaValueConversionType valueConversionType() { + return valueConversionType; + } + + /** Returns the JSON layout, or null for binary/string. */ + public @Nullable KafkaValueConversionLayout valueConversionLayout() { + return valueConversionLayout; + } + + /** Returns physical value targets; nested JSON projects its single ROW container. */ + public KafkaFieldProjection valueProjection() { + return valueProjection; + } + + /** Returns the key column position, or -1 when keys are not stored. */ + public int keyPosition() { + return keyPosition; + } + + /** + * Returns the headers column position, or -1 when headers are not stored. Decoding maps names + * to binary values, with the last occurrence winning for duplicate header names. + */ + public int headersPosition() { + return headersPosition; + } + + /** + * Returns the metadata column position, or -1 when absent. The ROW stores the input record + * timestamp, batch timestamp type and requested partition; log offsets are not row fields. + */ + public int metadataPosition() { + return metadataPosition; + } + + /** + * Returns the rescue column position, or -1 when absent. JSON decoders collect unknown and + * reserved members here; binary/string decoders write null to this optional column. + */ + public int rescuePosition() { + return rescuePosition; + } + + @Override + public boolean equals(Object other) { + if (!(other instanceof KafkaTableSchema)) { + return false; + } + KafkaTableSchema that = (KafkaTableSchema) other; + return rowType.equals(that.rowType) + && valueConversionType == that.valueConversionType + && valueConversionLayout == that.valueConversionLayout + && valueProjection.equals(that.valueProjection) + && keyPosition == that.keyPosition + && headersPosition == that.headersPosition + && metadataPosition == that.metadataPosition + && rescuePosition == that.rescuePosition; + } + + @Override + public int hashCode() { + return Objects.hash( + rowType, + valueConversionType, + valueConversionLayout, + valueProjection, + keyPosition, + headersPosition, + metadataPosition, + rescuePosition); + } + + private static InvalidTableException invalid(String message) { + return new InvalidTableException(message); + } +} diff --git a/fluss-common/src/test/java/org/apache/fluss/metadata/KafkaTableSchemaTest.java b/fluss-common/src/test/java/org/apache/fluss/metadata/KafkaTableSchemaTest.java new file mode 100644 index 00000000000..a8287eda6ee --- /dev/null +++ b/fluss-common/src/test/java/org/apache/fluss/metadata/KafkaTableSchemaTest.java @@ -0,0 +1,614 @@ +/* + * 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.config.ConfigOptions; +import org.apache.fluss.config.KafkaKeyConversionType; +import org.apache.fluss.config.KafkaValueConversionLayout; +import org.apache.fluss.config.KafkaValueConversionType; +import org.apache.fluss.exception.InvalidTableException; +import org.apache.fluss.types.DataType; +import org.apache.fluss.types.DataTypes; +import org.apache.fluss.types.RowType; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.Arrays; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests the shared DDL and Kafka conversion contract. */ +class KafkaTableSchemaTest { + + @Test + void testExplicitOptInAndDisabledTables() { + TableDescriptor disabled = + TableDescriptor.builder() + .schema( + Schema.newBuilder() + .column("_kafka_value", DataTypes.BYTES()) + .build()) + .property(ConfigOptions.TABLE_KAFKA_CONVERT_VALUE_TYPE.key(), "binary") + .build(); + assertThat(KafkaTableSchema.isEnabled(disabled)).isFalse(); + assertThatCode(() -> KafkaTableSchema.validate(disabled)).doesNotThrowAnyException(); + assertThatThrownBy(() -> KafkaTableSchema.resolve(disabled)) + .hasMessageContaining("disabled"); + TableDescriptor arbitrary = + TableDescriptor.builder(disabled) + .schema( + Schema.newBuilder() + .column("id", DataTypes.INT().copy(false)) + .primaryKey("id") + .build()) + .property( + ConfigOptions.TABLE_KAFKA_CONVERT_VALUE_TYPE.key(), "future-format") + .property(ConfigOptions.TABLE_KAFKA_ENABLED, false) + .build(); + assertThatCode(() -> KafkaTableSchema.validate(arbitrary)).doesNotThrowAnyException(); + } + + @Test + void testEnabledFlagParsing() { + TableDescriptor.Builder builder = enabled(valueSchema(), "binary"); + for (String value : Arrays.asList("true", "TRUE", "TrUe")) { + assertThat( + KafkaTableSchema.isEnabled( + builder.property("table.kafka.enabled", value).build())) + .isTrue(); + } + for (String value : Arrays.asList("false", "FALSE", "FaLsE")) { + assertThat( + KafkaTableSchema.isEnabled( + builder.property("table.kafka.enabled", value).build())) + .isFalse(); + } + for (String value : Arrays.asList("", "yes", "1", " true", "false ")) { + TableDescriptor table = builder.property("table.kafka.enabled", value).build(); + assertThatThrownBy(() -> KafkaTableSchema.isEnabled(table)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("table.kafka.enabled"); + } + } + + @ParameterizedTest + @ValueSource(strings = {"binary", "string"}) + void testFixedValueColumnWithOptionalSystemFields(String format) { + DataType valueType = format.equals("binary") ? DataTypes.BYTES() : DataTypes.STRING(); + assertInvalid( + enabled( + Schema.newBuilder() + .column("_kafka_value", valueType.copy(false)) + .build(), + format) + .build(), + "nullable fields"); + for (int mask = 0; mask < 16; mask++) { + Schema.Builder schema = Schema.newBuilder().column("_kafka_value", valueType); + if ((mask & 1) != 0) { + schema.column("_kafka_key", DataTypes.BYTES()); + } + if ((mask & 2) != 0) { + schema.column( + "_kafka_headers", DataTypes.MAP(DataTypes.STRING(), DataTypes.BYTES())); + } + if ((mask & 4) != 0) { + schema.column("_kafka_rescue", DataTypes.STRING()); + } + if ((mask & 8) != 0) { + schema.column("_kafka_metadata", metadataType()); + } + TableDescriptor table = enabled(schema.build(), format).build(); + KafkaTableSchema resolved = KafkaTableSchema.resolve(table); + assertThat(resolved.valueProjection().positions()).containsExactly(0); + assertThat(resolved.valueProjection().dataTypeAt(0).isNullable()).isTrue(); + assertThat(resolved.keyPosition() >= 0).isEqualTo((mask & 1) != 0); + assertThat(resolved.headersPosition() >= 0).isEqualTo((mask & 2) != 0); + assertThat(resolved.rescuePosition() >= 0).isEqualTo((mask & 4) != 0); + assertThat(resolved.metadataPosition() >= 0).isEqualTo((mask & 8) != 0); + assertThat(table.getProperties()) + .doesNotContainKeys( + "table.kafka.convert.key.type", "table.kafka.convert.value.layout"); + } + } + + @Test + void testDefaultJsonMappingPreservesNestedTypesAndExcludesSystemColumns() { + RowType nested = DataTypes.ROW(DataTypes.FIELD("_kafka_key", DataTypes.STRING())); + TableDescriptor table = + TableDescriptor.builder() + .schema( + Schema.newBuilder() + .column("id", DataTypes.BIGINT()) + .column("_kafka_rescue", DataTypes.STRING()) + .column("_kafka_metadata", metadataType()) + .column("payload", nested) + .column("_kafka_key", DataTypes.BYTES()) + .column( + "_kafka_headers", + DataTypes.MAP( + DataTypes.STRING(), DataTypes.BYTES())) + .column("custom_rescue", DataTypes.STRING()) + .build()) + .property(ConfigOptions.TABLE_KAFKA_ENABLED, true) + .build(); + KafkaTableSchema resolved = KafkaTableSchema.resolve(table); + assertThat(resolved.valueConversionType()).isEqualTo(KafkaValueConversionType.PARSE_JSON); + assertThat(resolved.valueConversionLayout()).isEqualTo(KafkaValueConversionLayout.FLAT); + assertThat(resolved.valueProjection().positions()).containsExactly(0, 3, 6); + assertThat(resolved.valueProjection().dataTypeAt(1)).isEqualTo(nested); + assertThat(resolved.rescuePosition()).isEqualTo(1); + assertThat(resolved.metadataPosition()).isEqualTo(2); + assertThat(table.getProperties()).containsOnlyKeys("table.kafka.enabled"); + } + + @Test + void testNestedJsonTargetsOneRow() { + RowType row = + DataTypes.ROW( + DataTypes.FIELD("id", DataTypes.STRING()), + DataTypes.FIELD( + "details", + DataTypes.ROW( + DataTypes.FIELD("_kafka_rescue", DataTypes.STRING())))); + TableDescriptor table = + enabled( + Schema.newBuilder() + .column("_kafka_rescue", DataTypes.STRING()) + .column("_kafka_metadata", metadataType()) + .column("_kafka_value", row) + .build(), + "parse_json") + .property(ConfigOptions.TABLE_KAFKA_CONVERT_VALUE_LAYOUT.key(), "nested") + .build(); + KafkaTableSchema resolved = KafkaTableSchema.resolve(table); + assertThat(resolved.valueConversionLayout()).isEqualTo(KafkaValueConversionLayout.NESTED); + assertThat(resolved.valueProjection().positions()).containsExactly(2); + assertThat(resolved.valueProjection().dataTypeAt(0)).isEqualTo(row); + assertThat(resolved.rescuePosition()).isZero(); + assertThat(resolved.metadataPosition()).isEqualTo(1); + } + + @ParameterizedTest + @ValueSource(strings = {"flat", "nested"}) + void testJsonRejectsNonStringMapKeysRecursively(String layout) { + for (DataType key : Arrays.asList(DataTypes.INT(), DataTypes.CHAR(10), DataTypes.BYTES())) { + DataType invalidMap = DataTypes.MAP(key, DataTypes.STRING()); + for (DataType type : + Arrays.asList( + invalidMap, + DataTypes.ARRAY(invalidMap), + DataTypes.ROW(DataTypes.FIELD("attributes", invalidMap)), + DataTypes.MAP(DataTypes.STRING(), DataTypes.ARRAY(invalidMap)))) { + TableDescriptor table = jsonTable(layout, type); + assertInvalid(table, "only supports STRING map keys"); + } + } + } + + @ParameterizedTest + @ValueSource(strings = {"flat", "nested"}) + void testJsonAcceptsAllScalarTypesInNestedContainers(String layout) { + for (DataType scalar : + Arrays.asList( + DataTypes.CHAR(10), + DataTypes.STRING(), + DataTypes.BOOLEAN(), + DataTypes.BINARY(4), + DataTypes.BYTES(), + DataTypes.DECIMAL(18, 2), + DataTypes.TINYINT(), + DataTypes.SMALLINT(), + DataTypes.INT(), + DataTypes.BIGINT(), + DataTypes.FLOAT(), + DataTypes.DOUBLE(), + DataTypes.DATE(), + DataTypes.TIME(3), + DataTypes.TIMESTAMP(9), + DataTypes.TIMESTAMP_LTZ(9))) { + DataType type = + DataTypes.ARRAY( + DataTypes.MAP( + DataTypes.STRING().copy(false), + DataTypes.ROW(DataTypes.FIELD("_kafka_key", scalar)))); + TableDescriptor table = jsonTable(layout, type); + assertThatCode(() -> KafkaTableSchema.validate(table)).doesNotThrowAnyException(); + } + } + + @ParameterizedTest + @ValueSource(strings = {"flat", "nested"}) + void testJsonBusinessNestingDepthBoundary(String layout) { + // Exercise every recursive edge, including map values (keys do not add a level). + DataType boundary = DataTypes.STRING(); + for (int depth = 0; depth < KafkaTableSchema.MAX_JSON_SCHEMA_NESTING_DEPTH; depth++) { + if (depth % 3 == 0) { + boundary = DataTypes.ARRAY(boundary); + } else if (depth % 3 == 1) { + boundary = DataTypes.MAP(DataTypes.STRING(), boundary); + } else { + boundary = DataTypes.ROW(DataTypes.FIELD("child", boundary)); + } + } + TableDescriptor valid = jsonTable(layout, boundary); + assertThatCode(() -> KafkaTableSchema.validate(valid)).doesNotThrowAnyException(); + TableDescriptor tooDeep = jsonTable(layout, DataTypes.ARRAY(boundary)); + assertInvalid(tooDeep, "maximum nesting depth of 64"); + } + + private static TableDescriptor jsonTable(String layout, DataType businessType) { + Schema schema = + "flat".equals(layout) + ? Schema.newBuilder().column("payload", businessType).build() + : Schema.newBuilder() + .column( + "_kafka_value", + DataTypes.ROW(DataTypes.FIELD("payload", businessType))) + .build(); + return enabled(schema, "parse_json") + .property(ConfigOptions.TABLE_KAFKA_CONVERT_VALUE_LAYOUT.key(), layout) + .build(); + } + + @ParameterizedTest + @ValueSource( + strings = { + "_kafka_key", + "_kafka_value", + "_kafka_headers", + "_kafka_metadata", + "_kafka_rescue" + }) + void testSystemNamesAreBusinessFieldsInsideNestedValue(String name) { + assertThat(KafkaTableSchema.isReservedName(name)).isTrue(); + RowType valueType = DataTypes.ROW(DataTypes.FIELD(name, DataTypes.STRING())); + TableDescriptor nested = + enabled( + Schema.newBuilder() + .column("_kafka_key", DataTypes.BYTES()) + .column("_kafka_value", valueType) + .build(), + "parse_json") + .property(ConfigOptions.TABLE_KAFKA_CONVERT_VALUE_LAYOUT.key(), "nested") + .build(); + KafkaTableSchema resolved = KafkaTableSchema.resolve(nested); + assertThat(resolved.keyPosition()).isZero(); + assertThat(resolved.valueProjection().positions()).containsExactly(1); + assertThat(resolved.valueProjection().dataTypeAt(0)).isEqualTo(valueType); + } + + @ParameterizedTest + @ValueSource(strings = {"raw", "avro", ""}) + void testRejectsUnsupportedValueTypes(String type) { + assertThatThrownBy(() -> KafkaTableSchema.resolve(enabled(valueSchema(), type).build())) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("table.kafka.convert.value.type"); + } + + @Test + void testRejectsValueTargetConflicts() { + assertInvalid( + TableDescriptor.builder() + .schema(valueSchema()) + .property(ConfigOptions.TABLE_KAFKA_ENABLED, true) + .build(), + "flat JSON"); + assertInvalid( + enabled(Schema.newBuilder().column("body", DataTypes.BYTES()).build(), "binary") + .build(), + "Missing required value column"); + assertInvalid(enabled(valueSchema(), "string").build(), "must be STRING"); + assertInvalid( + enabled(valueSchema(), "parse_json") + .property(ConfigOptions.TABLE_KAFKA_CONVERT_VALUE_LAYOUT.key(), "nested") + .build(), + "must be ROW"); + assertInvalid(enabled(valueSchema(), "parse_json").build(), "flat JSON"); + assertInvalid( + enabled( + Schema.newBuilder() + .column("_kafka_value", DataTypes.BYTES()) + .column("unused", DataTypes.STRING()) + .build(), + "binary") + .build(), + "no value mapping"); + assertInvalid( + enabled( + Schema.newBuilder() + .column( + "_kafka_value", + DataTypes.ROW( + DataTypes.FIELD("id", DataTypes.STRING()))) + .column("unused", DataTypes.STRING()) + .build(), + "parse_json") + .property(ConfigOptions.TABLE_KAFKA_CONVERT_VALUE_LAYOUT.key(), "nested") + .build(), + "no value mapping"); + } + + @ParameterizedTest + @ValueSource(strings = {"binary", "string"}) + void testNonJsonIgnoresDefaultAndExplicitLayouts(String type) { + Schema schema = + Schema.newBuilder() + .column( + "_kafka_value", + "binary".equals(type) ? DataTypes.BYTES() : DataTypes.STRING()) + .build(); + TableDescriptor table = enabled(schema, type).build(); + KafkaTableSchema expected = KafkaTableSchema.resolve(table); + assertThat(expected.valueConversionLayout()).isNull(); + assertThat(expected.valueProjection().positions()).containsExactly(0); + assertThat(table.getProperties()).doesNotContainKey("table.kafka.convert.value.layout"); + for (KafkaValueConversionLayout layout : KafkaValueConversionLayout.values()) { + assertThat( + KafkaTableSchema.resolve( + TableDescriptor.builder(table) + .property( + ConfigOptions.TABLE_KAFKA_CONVERT_VALUE_LAYOUT, + layout) + .build())) + .isEqualTo(expected); + } + } + + @Test + void testRejectsInvalidJsonLayout() { + TableDescriptor table = + enabled(valueSchema(), "parse_json") + .property(ConfigOptions.TABLE_KAFKA_CONVERT_VALUE_LAYOUT.key(), "unknown") + .build(); + assertThatThrownBy(() -> KafkaTableSchema.resolve(table)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("table.kafka.convert.value.layout"); + } + + @Test + void testSystemColumnTypes() { + for (DataType invalid : Arrays.asList(DataTypes.STRING(), DataTypes.BINARY(8))) { + assertInvalid(withSystem("_kafka_key", invalid), "_kafka_key must be BYTES"); + } + for (DataType invalid : + Arrays.asList( + DataTypes.BYTES(), + DataTypes.ARRAY(DataTypes.STRING()), + DataTypes.MAP(DataTypes.INT(), DataTypes.BYTES()), + DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()))) { + assertInvalid(withSystem("_kafka_headers", invalid), "MAP"); + } + assertInvalid(withSystem("_kafka_rescue", DataTypes.INT()), "nullable STRING"); + assertInvalid( + withSystem("_kafka_rescue", DataTypes.STRING().copy(false)), "nullable STRING"); + } + + @Test + void testSystemFieldsRequireNullableTypes() { + assertInvalid(withSystem("_kafka_key", DataTypes.BYTES().copy(false)), "_kafka_key"); + assertInvalid( + withSystem( + "_kafka_headers", + DataTypes.MAP(DataTypes.STRING(), DataTypes.BYTES()).copy(false)), + "_kafka_headers"); + assertInvalid( + withSystem( + "_kafka_headers", + DataTypes.MAP(DataTypes.STRING(), DataTypes.BYTES().copy(false))), + "_kafka_headers"); + assertInvalid(withSystem("_kafka_metadata", metadataType().copy(false)), "_kafka_metadata"); + for (int i = 0; i < metadataType().getFieldCount(); i++) { + RowType.Builder metadata = RowType.builder(); + for (int j = 0; j < metadataType().getFieldCount(); j++) { + metadata.field( + metadataType().getFields().get(j).getName(), + metadataType().getTypeAt(j).copy(i != j)); + } + assertInvalid( + withSystem("_kafka_metadata", metadata.build()), + "_kafka_metadata." + metadataType().getFields().get(i).getName()); + } + } + + @ParameterizedTest + @ValueSource(strings = {"flat", "nested"}) + void testJsonFieldsRequireNullableTypesRecursively(String layout) { + DataType notNull = DataTypes.INT().copy(false); + for (DataType type : + Arrays.asList( + notNull, + DataTypes.ROW(DataTypes.FIELD("child", notNull)), + DataTypes.ROW(DataTypes.FIELD("child", DataTypes.INT())).copy(false), + DataTypes.ARRAY(notNull), + DataTypes.ARRAY(DataTypes.INT()).copy(false), + DataTypes.MAP(DataTypes.STRING(), notNull), + DataTypes.MAP(DataTypes.STRING(), DataTypes.INT()).copy(false))) { + assertInvalid(jsonTable(layout, type), "nullable fields"); + } + if ("nested".equals(layout)) { + assertInvalid( + enabled( + Schema.newBuilder() + .column( + "_kafka_value", + DataTypes.ROW( + DataTypes.FIELD( + "id", DataTypes.INT())) + .copy(false)) + .build(), + "parse_json") + .property( + ConfigOptions.TABLE_KAFKA_CONVERT_VALUE_LAYOUT, + KafkaValueConversionLayout.NESTED) + .build(), + "_kafka_value"); + } + } + + @Test + void testOptionalBinaryKeyConversion() { + TableDescriptor table = withSystem("_kafka_key", DataTypes.BYTES()); + TableDescriptor explicit = + TableDescriptor.builder(table) + .property( + ConfigOptions.TABLE_KAFKA_CONVERT_KEY_TYPE, + KafkaKeyConversionType.BINARY) + .build(); + assertThat(KafkaTableSchema.resolve(explicit)).isEqualTo(KafkaTableSchema.resolve(table)); + assertThat(table.getProperties()).doesNotContainKey("table.kafka.convert.key.type"); + assertInvalid( + enabled(valueSchema(), "binary") + .property( + ConfigOptions.TABLE_KAFKA_CONVERT_KEY_TYPE, + KafkaKeyConversionType.BINARY) + .build(), + "requires _kafka_key BYTES"); + assertThatThrownBy( + () -> + KafkaTableSchema.resolve( + TableDescriptor.builder(table) + .property("table.kafka.convert.key.type", "string") + .build())) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("table.kafka.convert.key.type"); + } + + @Test + void testMetadataRejectsInvalidStructureAndOffsetField() { + for (DataType invalid : + Arrays.asList( + DataTypes.STRING(), + DataTypes.ROW( + DataTypes.FIELD("timestamp", DataTypes.TIMESTAMP_LTZ(3)), + DataTypes.FIELD("timestamp_type", DataTypes.INT())), + DataTypes.ROW( + DataTypes.FIELD("timestamp", DataTypes.TIMESTAMP_LTZ(3)), + DataTypes.FIELD("timestamp_type", DataTypes.INT()), + DataTypes.FIELD("partition", DataTypes.INT()), + DataTypes.FIELD("offset", DataTypes.BIGINT())), + DataTypes.ROW( + DataTypes.FIELD("timestamp", DataTypes.TIMESTAMP(3)), + DataTypes.FIELD("timestamp_type", DataTypes.INT()), + DataTypes.FIELD("partition", DataTypes.INT())), + DataTypes.ROW( + DataTypes.FIELD("timestamp", DataTypes.TIMESTAMP_LTZ(6)), + DataTypes.FIELD("timestamp_type", DataTypes.INT()), + DataTypes.FIELD("partition", DataTypes.INT())), + DataTypes.ROW( + DataTypes.FIELD("timestamp", DataTypes.TIMESTAMP_LTZ(3)), + DataTypes.FIELD("timestamp_type", DataTypes.STRING()), + DataTypes.FIELD("partition", DataTypes.INT())), + DataTypes.ROW( + DataTypes.FIELD("timestamp", DataTypes.TIMESTAMP_LTZ(3)), + DataTypes.FIELD("timestamp_type", DataTypes.INT()), + DataTypes.FIELD("partition", DataTypes.BIGINT())), + DataTypes.ROW( + DataTypes.FIELD("timestamp", DataTypes.TIMESTAMP_LTZ(3)), + DataTypes.FIELD("partition", DataTypes.INT()), + DataTypes.FIELD("timestamp_type", DataTypes.INT())), + DataTypes.ROW( + DataTypes.FIELD("timestamp", DataTypes.TIMESTAMP_LTZ(3)), + DataTypes.FIELD("timestamp_type", DataTypes.INT()), + DataTypes.FIELD("bucket", DataTypes.INT())))) { + assertInvalid( + withSystem("_kafka_metadata", invalid), + "ROW"); + } + } + + @Test + void testUnrelatedPropertiesDoNotChangeMapping() { + TableDescriptor table = enabled(valueSchema(), "binary").build(); + KafkaTableSchema expected = KafkaTableSchema.resolve(table); + // Storage option registration is checked by the server, not by this shared resolver. + assertThat( + KafkaTableSchema.resolve( + TableDescriptor.builder(table) + .property("table.kafka.future-option", "anything") + .build())) + .isEqualTo(expected); + for (String key : + Arrays.asList( + "kafka.value.format", + "fluss.value.format", + "table.kafka.convert.value.type", + "owner")) { + assertThat( + KafkaTableSchema.resolve( + TableDescriptor.builder(table) + .customProperty(key, "anything") + .build())) + .isEqualTo(expected); + } + } + + @Test + void testUnsupportedTableKinds() { + Schema keyed = + Schema.newBuilder() + .column("_kafka_value", DataTypes.BYTES().copy(false)) + .primaryKey("_kafka_value") + .build(); + assertInvalid(enabled(keyed, "binary").build(), "without a primary key"); + assertInvalid( + enabled(valueSchema(), "binary").partitionedBy("_kafka_value").build(), + "non-partitioned"); + assertInvalid( + enabled(valueSchema(), "binary").logFormat(LogFormat.INDEXED).build(), "Arrow"); + } + + private static RowType metadataType() { + return DataTypes.ROW( + DataTypes.FIELD("timestamp", DataTypes.TIMESTAMP_LTZ(3)), + DataTypes.FIELD("timestamp_type", DataTypes.INT()), + DataTypes.FIELD("partition", DataTypes.INT())); + } + + private static TableDescriptor withSystem(String name, DataType type) { + return enabled( + Schema.newBuilder() + .column("_kafka_value", DataTypes.BYTES()) + .column(name, type) + .build(), + "binary") + .build(); + } + + private static Schema valueSchema() { + return Schema.newBuilder().column("_kafka_value", DataTypes.BYTES()).build(); + } + + private static TableDescriptor.Builder enabled(Schema schema, String type) { + return TableDescriptor.builder() + .schema(schema) + .logFormat(LogFormat.ARROW) + .property(ConfigOptions.TABLE_KAFKA_ENABLED, true) + .property(ConfigOptions.TABLE_KAFKA_CONVERT_VALUE_TYPE.key(), type); + } + + private static void assertInvalid(TableDescriptor table, String message) { + assertThatThrownBy(() -> KafkaTableSchema.resolve(table)) + .isInstanceOf(InvalidTableException.class) + .hasMessageContaining(message); + } +} diff --git a/fluss-kafka/pom.xml b/fluss-kafka/pom.xml index f48a94ab9e9..b77695848ca 100644 --- a/fluss-kafka/pom.xml +++ b/fluss-kafka/pom.xml @@ -64,6 +64,13 @@ + + org.apache.curator + curator-test + ${curator.version} + test + + org.apache.fluss fluss-test-utils 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..aa31d618986 --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/mapping/KafkaTopicMapper.java @@ -0,0 +1,125 @@ +/* + * 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.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.metadata.KafkaTableSchema; +import org.apache.fluss.metadata.TablePath; + +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.errors.InvalidTopicException; +import org.apache.kafka.common.internals.Topic; + +import static org.apache.fluss.utils.Preconditions.checkArgument; + +/** Maps canonical Kafka topic names and default-database aliases to Fluss tables. */ +@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 defaultDatabase; + + /** Creates a mapper using the default database configuration. */ + KafkaTopicMapper() { + this(new Configuration()); + } + + /** Creates a mapper using the server's default database; it never creates databases. */ + public KafkaTopicMapper(Configuration config) { + this.defaultDatabase = config.get(ConfigOptions.KAFKA_DEFAULT_DATABASE); + TablePath.validateDatabaseName(defaultDatabase); + } + + /** Maps a canonical topic or an unqualified alias to a table path, without fallback lookup. */ + public TablePath toTablePath(String topicName) { + if (!isValidTopic(topicName)) { + throw new InvalidTopicException( + "Kafka topic must be a valid table or database.table name: " + topicName); + } + int separator = topicName.indexOf('.'); + TablePath path = + separator < 0 + ? TablePath.of(defaultDatabase, topicName) + : TablePath.of( + topicName.substring(0, separator), + topicName.substring(separator + 1)); + if (!isMappedTableValid(path)) { + throw new InvalidTopicException("Canonical Kafka topic name is invalid: " + path); + } + return path; + } + + /** Returns the fully qualified Kafka name of a representable Fluss table. */ + public String toTopicName(TablePath tablePath) { + if (!isMappedTableValid(tablePath)) { + throw new InvalidTopicException( + "Fluss table cannot be represented as a Kafka database.table name: " + + tablePath); + } + return tablePath.toString(); + } + + /** + * Returns whether a table path maps to a valid, fully qualified Kafka topic name. This does not + * check whether the table exists or enables Kafka compatibility. + */ + public boolean isMappedTableValid(TablePath tablePath) { + return KafkaTableSchema.isValidTablePath(tablePath); + } + + /** Returns whether a name uniquely represents a user table in a Fluss database. */ + public static boolean isValidTopic(String topicName) { + if (topicName == null || !Topic.isValid(topicName)) { + return false; + } + int separator = topicName.indexOf('.'); + if (separator < 0) { + return TablePath.detectInvalidName(topicName) == null + && TablePath.validatePrefix(topicName) == null; + } + if (separator == 0 || separator != topicName.lastIndexOf('.')) { + return false; + } + return KafkaTableSchema.isValidTablePath( + TablePath.of( + topicName.substring(0, separator), topicName.substring(separator + 1))); + } + + /** 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/main/java/org/apache/fluss/kafka/schema/KafkaTopicSchemaException.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/schema/KafkaTopicSchemaException.java new file mode 100644 index 00000000000..0e6a294c4d5 --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/schema/KafkaTopicSchemaException.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.schema; + +import org.apache.fluss.annotation.Internal; + +/** Indicates that a Fluss table does not define a valid Kafka record mapping contract. */ +@Internal +public final class KafkaTopicSchemaException extends IllegalArgumentException { + + /** Creates a Kafka topic schema exception. */ + public KafkaTopicSchemaException(String message) { + super(message); + } +} diff --git a/fluss-kafka/src/main/java/org/apache/fluss/kafka/schema/KafkaTopicSchemaResolver.java b/fluss-kafka/src/main/java/org/apache/fluss/kafka/schema/KafkaTopicSchemaResolver.java new file mode 100644 index 00000000000..c02879e922a --- /dev/null +++ b/fluss-kafka/src/main/java/org/apache/fluss/kafka/schema/KafkaTopicSchemaResolver.java @@ -0,0 +1,46 @@ +/* + * 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.schema; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.exception.InvalidTableException; +import org.apache.fluss.metadata.KafkaTableSchema; +import org.apache.fluss.metadata.TableDescriptor; + +/** Resolves the same persisted Kafka contract that native DDL validates. */ +@Internal +public final class KafkaTopicSchemaResolver { + + /** Returns the table's explicit opt-in flag without inferring it from schema compatibility. */ + public boolean isEnabled(TableDescriptor table) { + try { + return KafkaTableSchema.isEnabled(table); + } catch (IllegalArgumentException e) { + throw new KafkaTopicSchemaException(e.getMessage()); + } + } + + /** Resolves one enabled table's contract; invalid enabled tables remain visible as errors. */ + public KafkaTableSchema resolve(TableDescriptor table) { + try { + return KafkaTableSchema.resolve(table); + } catch (InvalidTableException | IllegalArgumentException e) { + throw new KafkaTopicSchemaException(e.getMessage()); + } + } +} 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 deleted file mode 100644 index 7f502494f2e..00000000000 --- a/fluss-kafka/src/test/java/org/apache/fluss/kafka/KafkaConfigsTest.java +++ /dev/null @@ -1,56 +0,0 @@ -/* - * 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.config.Configuration; - -import org.junit.jupiter.api.Test; - -import java.util.Arrays; -import java.util.Collections; -import java.util.HashMap; -import java.util.Map; - -import static org.assertj.core.api.Assertions.assertThat; - -/** Tests for Kafka configuration. */ -public class KafkaConfigsTest { - @Test - public void testFromMap() throws Exception { - Map map = new HashMap<>(); - map.put(ConfigOptions.KAFKA_ENABLED.key(), "true"); - map.put(ConfigOptions.KAFKA_LISTENER_NAMES.key(), "kafka,kafka_sasl"); - map.put(ConfigOptions.KAFKA_DATABASE.key(), "fluss"); - 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"); - } - - @Test - public void testFromDefault() throws Exception { - Configuration configuration = Configuration.fromMap(new HashMap<>()); - assertThat(configuration.getBoolean(ConfigOptions.KAFKA_ENABLED)).isFalse(); - assertThat(configuration.get(ConfigOptions.KAFKA_LISTENER_NAMES)) - .isEqualTo(Collections.singletonList("KAFKA")); - assertThat(configuration.getString(ConfigOptions.KAFKA_DATABASE)).isEqualTo("kafka"); - } -} 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..c2aa5206640 --- /dev/null +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/mapping/KafkaTopicMapperTest.java @@ -0,0 +1,177 @@ +/* + * 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.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.metadata.TablePath; + +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.errors.InvalidTopicException; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.NullAndEmptySource; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.Arrays; +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link KafkaTopicMapper}. */ +public class KafkaTopicMapperTest { + + private final KafkaTopicMapper mapper = new KafkaTopicMapper(); + + @Test + public void testQualifiedNamesAcrossDatabases() { + for (String database : Arrays.asList("sales", "archive", "sales_1-2026")) { + TablePath tablePath = TablePath.of(database, "orders_1-2026"); + String topicName = database + ".orders_1-2026"; + assertThat(mapper.toTablePath(topicName)).isEqualTo(tablePath); + assertThat(mapper.toTopicName(tablePath)).isEqualTo(topicName); + assertThat(mapper.isMappedTableValid(tablePath)).isTrue(); + assertThat(KafkaTopicMapper.isValidTopic(topicName)).isTrue(); + } + } + + @Test + void testDefaultDatabaseAliases() { + assertThat(mapper.toTablePath("orders")).isEqualTo(TablePath.of("fluss", "orders")); + assertThat(mapper.toTopicName(mapper.toTablePath("orders"))).isEqualTo("fluss.orders"); + assertThat(mapper.toTablePath("kafka.orders")).isEqualTo(TablePath.of("kafka", "orders")); + Configuration config = new Configuration(); + config.set(ConfigOptions.KAFKA_DEFAULT_DATABASE, "sales"); + KafkaTopicMapper custom = new KafkaTopicMapper(config); + assertThat(custom.toTablePath("orders")).isEqualTo(TablePath.of("sales", "orders")); + assertThat(custom.toTablePath("warehouse.orders")) + .isEqualTo(TablePath.of("warehouse", "orders")); + assertThat(custom.toTopicName(custom.toTablePath("orders"))).isEqualTo("sales.orders"); + config.setString("kafka.database", "legacy"); + assertThat(new KafkaTopicMapper(config).toTablePath("orders")) + .isEqualTo(TablePath.of("sales", "orders")); + } + + @Test + void testInvalidDefaultDatabaseAndCanonicalAliasLength() { + Configuration config = new Configuration(); + config.set(ConfigOptions.KAFKA_DEFAULT_DATABASE, "bad.name"); + assertThatThrownBy(() -> new KafkaTopicMapper(config)).hasMessageContaining("invalid"); + config.set( + ConfigOptions.KAFKA_DEFAULT_DATABASE, + String.join("", Collections.nCopies(128, "d"))); + KafkaTopicMapper longDatabase = new KafkaTopicMapper(config); + assertThatThrownBy( + () -> + longDatabase.toTablePath( + String.join("", Collections.nCopies(128, "t")))) + .isInstanceOf(InvalidTopicException.class); + } + + @ParameterizedTest + @NullAndEmptySource + @ValueSource( + strings = { + ".orders", + "sales.", + "sales.orders.extra", + "sales..orders", + "sales.bad name", + " sales.orders", + "sales.orders ", + "sales/orders", + "__system.orders", + "sales.__internal", + "数据库.orders", + "sales.订单" + }) + public void testRejectsInvalidTopicNames(String topicName) { + assertThat(KafkaTopicMapper.isValidTopic(topicName)).isFalse(); + assertThatThrownBy(() -> mapper.toTablePath(topicName)) + .isInstanceOf(InvalidTopicException.class); + } + + @Test + public void testNameLengthLimits() { + for (TablePath valid : + Arrays.asList( + TablePath.of(repeat('a', 200), "orders"), + TablePath.of("sales", repeat('a', 200)), + TablePath.of(repeat('a', 124), repeat('b', 124)))) { + assertThat(mapper.toTablePath(mapper.toTopicName(valid))).isEqualTo(valid); + } + for (TablePath invalid : + Arrays.asList( + TablePath.of(repeat('a', 201), "orders"), + TablePath.of("sales", repeat('a', 201)), + TablePath.of(repeat('a', 124), repeat('b', 125)))) { + assertThat(mapper.isMappedTableValid(invalid)).isFalse(); + assertThatThrownBy(() -> mapper.toTablePath(invalid.toString())) + .isInstanceOf(InvalidTopicException.class); + assertThatThrownBy(() -> mapper.toTopicName(invalid)) + .isInstanceOf(InvalidTopicException.class); + } + } + + @Test + public void testRejectsUnrepresentableTablePaths() { + for (TablePath invalid : + Arrays.asList( + null, + TablePath.of(null, "orders"), + TablePath.of("sales", null), + TablePath.of("", "orders"), + TablePath.of("sales", ""), + TablePath.of("sales.region", "orders"), + TablePath.of("sales", "orders.v1"), + TablePath.of("__system", "orders"), + TablePath.of("sales", "__internal"), + TablePath.of("sales", "bad name"))) { + assertThat(mapper.isMappedTableValid(invalid)).isFalse(); + assertThatThrownBy(() -> mapper.toTopicName(invalid)) + .isInstanceOf(InvalidTopicException.class); + } + } + + @Test + public void testTopicIdMapping() { + 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(); + } + + @Test + public void testRejectsInvalidTopicIds() { + assertThatThrownBy(() -> mapper.toTopicId(-1L)) + .isInstanceOf(IllegalArgumentException.class); + assertThat(mapper.isMappedTopicId(Uuid.ZERO_UUID)).isFalse(); + assertThatThrownBy(() -> mapper.toTableId(Uuid.ZERO_UUID)) + .isInstanceOf(IllegalArgumentException.class); + } + + private static String repeat(char character, int count) { + return String.join("", Collections.nCopies(count, String.valueOf(character))); + } +} diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/schema/KafkaTableMappingITCase.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/schema/KafkaTableMappingITCase.java new file mode 100644 index 00000000000..43eb7c9a40e --- /dev/null +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/schema/KafkaTableMappingITCase.java @@ -0,0 +1,636 @@ +/* + * 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.schema; + +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.KafkaValueConversionLayout; +import org.apache.fluss.config.KafkaValueConversionType; +import org.apache.fluss.exception.InvalidConfigException; +import org.apache.fluss.exception.InvalidTableException; +import org.apache.fluss.kafka.mapping.KafkaTopicMapper; +import org.apache.fluss.metadata.KafkaTableSchema; +import org.apache.fluss.metadata.LogFormat; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.rpc.gateway.CoordinatorGateway; +import org.apache.fluss.rpc.messages.AlterTableRequest; +import org.apache.fluss.rpc.messages.GetTableInfoResponse; +import org.apache.fluss.rpc.messages.PbAddColumn; +import org.apache.fluss.server.testutils.FlussClusterExtension; +import org.apache.fluss.types.DataType; +import org.apache.fluss.types.DataTypes; +import org.apache.fluss.utils.json.DataTypeJsonSerde; +import org.apache.fluss.utils.json.JsonSerdeUtils; + +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import static org.apache.fluss.server.coordinator.CoordinatorServer.DEFAULT_DATABASE; +import static org.apache.fluss.server.testutils.RpcMessageTestUtils.createTable; +import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newAlterTableRequest; +import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newCreateDatabaseRequest; +import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newCreateTableRequest; +import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newDatabaseExistsRequest; +import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newGetTableInfoRequest; +import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newTableExistsRequest; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Verifies Kafka DDL contracts through real coordinator RPC and persisted metadata. */ +class KafkaTableMappingITCase { + + private static final Configuration CONFIG = new Configuration(); + + static { + CONFIG.set(ConfigOptions.KAFKA_ENABLED, true); + CONFIG.set(ConfigOptions.KAFKA_DEFAULT_DATABASE, "kafka_default"); + CONFIG.setString("kafka.database", "legacy_database"); + } + + @RegisterExtension + public static final FlussClusterExtension CLUSTER = + FlussClusterExtension.builder().setNumOfTabletServers(1).setClusterConf(CONFIG).build(); + + private CoordinatorGateway gateway; + + @BeforeAll + static void verifyConfiguredDatabaseCreatedOnStartup() throws Exception { + assertThat( + CLUSTER.newCoordinatorClient() + .databaseExists(newDatabaseExistsRequest("kafka_default")) + .get() + .isExists()) + .isTrue(); + } + + @BeforeEach + void setup() { + gateway = CLUSTER.newCoordinatorClient(); + } + + @Test + void testConfiguredKafkaDatabaseAndQualifiedAccess() throws Exception { + for (String database : Arrays.asList("kafka", "legacy_database")) { + assertThat(gateway.databaseExists(newDatabaseExistsRequest(database)).get().isExists()) + .isFalse(); + } + assertThat( + gateway.databaseExists(newDatabaseExistsRequest(DEFAULT_DATABASE)) + .get() + .isExists()) + .isTrue(); + KafkaTopicMapper mapper = new KafkaTopicMapper(CONFIG); + TablePath sales = mapper.toTablePath("sales.orders"); + TablePath warehouse = mapper.toTablePath("warehouse.orders"); + long salesId = createTable(CLUSTER, sales, enabled().build()); + long warehouseId = createTable(CLUSTER, warehouse, enabled().build()); + assertThat(salesId).isNotEqualTo(warehouseId); + assertThat(mapper.toTablePath("orders")).isEqualTo(TablePath.of("kafka_default", "orders")); + // The cluster fixture removes custom databases between tests; recreate this table's + // database through the usual DDL helper after checking startup creation in @BeforeAll. + createTable(CLUSTER, mapper.toTablePath("orders"), enabled().build()); + assertThat( + gateway.databaseExists(newDatabaseExistsRequest("kafka_default")) + .get() + .isExists()) + .isTrue(); + assertThat(KafkaTableSchema.isEnabled(stored(mapper.toTablePath("orders")))).isTrue(); + Configuration config = new Configuration(); + config.set(ConfigOptions.KAFKA_DEFAULT_DATABASE, "sales"); + KafkaTopicMapper aliasMapper = new KafkaTopicMapper(config); + assertThat(info(aliasMapper.toTablePath("orders")).getTableId()) + .isEqualTo(info(sales).getTableId()); + assertThat(aliasMapper.toTopicName(sales)).isEqualTo("sales.orders"); + } + + @ParameterizedTest + @ValueSource(strings = {"flat", "nested"}) + void testJsonCreateAndEnableRejectUnsupportedBusinessSchema(String layout) throws Exception { + gateway.createDatabase(newCreateDatabaseRequest("ddl_json", true)).get(); + TablePath path = TablePath.of("ddl_json", layout + "_invalid"); + DataType invalid = + DataTypes.ARRAY( + DataTypes.ROW( + DataTypes.FIELD( + "attributes", + DataTypes.MAP(DataTypes.INT(), DataTypes.STRING())))); + TableDescriptor descriptor = jsonTable(layout, invalid).build(); + assertThatThrownBy( + () -> + gateway.createTable(newCreateTableRequest(path, descriptor, false)) + .get()) + .hasCauseInstanceOf(InvalidTableException.class); + assertThat(gateway.tableExists(newTableExistsRequest(path)).get().isExists()).isFalse(); + assertThat(CLUSTER.getZooKeeperClient().getCurrentSchemaId(path)).isZero(); + + createTable( + CLUSTER, + path, + TableDescriptor.builder(descriptor) + .property(ConfigOptions.TABLE_KAFKA_ENABLED, false) + .build()); + GetTableInfoResponse before = info(path); + assertThatThrownBy(() -> set(path, Collections.singletonMap("table.kafka.enabled", "true"))) + .hasCauseInstanceOf(InvalidTableException.class); + assertUnchanged(path, before); + } + + @ParameterizedTest + @ValueSource(strings = {"flat", "nested"}) + void testJsonComplexSchemaRoundtripAndEnable(String layout) throws Exception { + TablePath path = TablePath.of("ddl_json", layout + "_complex"); + DataType type = + DataTypes.ARRAY( + DataTypes.MAP( + DataTypes.STRING(), + DataTypes.ROW( + DataTypes.FIELD("amount", DataTypes.DECIMAL(18, 2)), + DataTypes.FIELD("time", DataTypes.TIMESTAMP_LTZ(3)), + DataTypes.FIELD("_kafka_key", DataTypes.BYTES())))); + TableDescriptor descriptor = + jsonTable(layout, type).property(ConfigOptions.TABLE_KAFKA_ENABLED, false).build(); + createTable(CLUSTER, path, descriptor); + GetTableInfoResponse before = info(path); + set(path, Collections.singletonMap("table.kafka.enabled", "true")); + TableDescriptor actual = stored(path); + assertThat(actual.getSchema()).isEqualTo(descriptor.getSchema()); + assertThat(info(path).getSchemaId()).isEqualTo(before.getSchemaId()); + Map expected = new HashMap<>(kafkaProperties(descriptor)); + expected.put("table.kafka.enabled", "true"); + assertThat(kafkaProperties(actual)).isEqualTo(expected); + addColumn(path, "_kafka_rescue", DataTypes.STRING()); + assertThat(info(path).getSchemaId()).isGreaterThan(before.getSchemaId()); + assertThat(stored(path).getSchema().getRowType().getTypeAt(0)) + .isEqualTo(descriptor.getSchema().getRowType().getTypeAt(0)); + assertThat(KafkaTableSchema.resolve(stored(path)).rescuePosition()).isEqualTo(1); + assertThat(kafkaProperties(stored(path))).isEqualTo(expected); + } + + @Test + void testJsonSchemaAlterRejectsUnsupportedNestedMapAtomically() throws Exception { + TablePath path = TablePath.of("ddl_json", "alter_business_type"); + createTable(CLUSTER, path, jsonTable("flat", DataTypes.STRING()).build()); + DataType invalid = + DataTypes.ARRAY( + DataTypes.ROW( + DataTypes.FIELD( + "attributes", + DataTypes.MAP(DataTypes.INT(), DataTypes.STRING())))); + GetTableInfoResponse before = info(path); + assertThatThrownBy(() -> addColumn(path, "bad", invalid)) + .hasCauseInstanceOf(InvalidTableException.class) + .hasStackTraceContaining("only supports STRING map keys"); + assertUnchanged(path, before); + addColumn( + path, "good", DataTypes.MAP(DataTypes.STRING(), DataTypes.ARRAY(DataTypes.INT()))); + assertThat(info(path).getSchemaId()).isGreaterThan(before.getSchemaId()); + set(path, Collections.singletonMap("table.kafka.enabled", "false")); + addColumn(path, "ordinary", invalid); + GetTableInfoResponse disabled = info(path); + assertThat(stored(path).getSchema().getRowType().getFieldNames()).contains("ordinary"); + assertThatThrownBy(() -> set(path, Collections.singletonMap("table.kafka.enabled", "true"))) + .hasCauseInstanceOf(InvalidTableException.class) + .hasStackTraceContaining("only supports STRING map keys"); + assertUnchanged(path, disabled); + } + + private void assertUnchanged(TablePath path, GetTableInfoResponse before) throws Exception { + GetTableInfoResponse after = info(path); + assertThat(after.getTableJson()).isEqualTo(before.getTableJson()); + assertThat(after.getSchemaId()).isEqualTo(before.getSchemaId()); + } + + @ParameterizedTest + @ValueSource(strings = {"value", "system", "nested"}) + void testNotNullFieldsAreRejectedBeforePersistence(String field) throws Exception { + boolean system = "system".equals(field); + String column = system ? "_kafka_metadata" : "payload"; + DataType type = + system + ? DataTypes.ROW( + DataTypes.FIELD("timestamp", DataTypes.TIMESTAMP_LTZ(3)), + DataTypes.FIELD("timestamp_type", DataTypes.INT()), + DataTypes.FIELD("partition", DataTypes.INT().copy(false))) + : "nested".equals(field) + ? DataTypes.ARRAY( + DataTypes.MAP( + DataTypes.STRING(), + DataTypes.ROW( + DataTypes.FIELD( + "id", + DataTypes.INT().copy(false))))) + : DataTypes.INT().copy(false); + TableDescriptor descriptor = + TableDescriptor.builder() + .schema(Schema.newBuilder().column(column, type).build()) + .distributedBy(1) + .property(ConfigOptions.TABLE_KAFKA_ENABLED, true) + .build(); + TablePath path = TablePath.of("ddl", "not_null_" + field); + gateway.createDatabase(newCreateDatabaseRequest("ddl", true)).get(); + assertThatThrownBy( + () -> + gateway.createTable(newCreateTableRequest(path, descriptor, false)) + .get()) + .hasCauseInstanceOf(InvalidTableException.class) + .hasStackTraceContaining("nullable fields"); + assertThat(gateway.tableExists(newTableExistsRequest(path)).get().isExists()).isFalse(); + assertThat(CLUSTER.getZooKeeperClient().getCurrentSchemaId(path)).isZero(); + createTable( + CLUSTER, + path, + TableDescriptor.builder(descriptor) + .property(ConfigOptions.TABLE_KAFKA_ENABLED, false) + .build()); + GetTableInfoResponse before = info(path); + assertThatThrownBy(() -> set(path, Collections.singletonMap("table.kafka.enabled", "true"))) + .hasCauseInstanceOf(InvalidTableException.class) + .hasStackTraceContaining("nullable fields"); + assertUnchanged(path, before); + + // Native schema evolution already rejects new NOT NULL columns. Kafka additionally + // rejects NOT NULL children of otherwise nullable columns. + if (type.isNullable()) { + TablePath alter = TablePath.of("ddl", "alter_not_null_" + field); + createTable(CLUSTER, alter, jsonTable("flat", DataTypes.STRING()).build()); + GetTableInfoResponse beforeAlter = info(alter); + assertThatThrownBy(() -> addColumn(alter, system ? column : "invalid", type)) + .hasCauseInstanceOf(InvalidTableException.class) + .hasStackTraceContaining("nullable fields"); + assertUnchanged(alter, beforeAlter); + } + } + + private static TableDescriptor.Builder jsonTable(String layout, DataType businessType) { + Schema schema = + "flat".equals(layout) + ? Schema.newBuilder().column("payload", businessType).build() + : Schema.newBuilder() + .column( + "_kafka_value", + DataTypes.ROW(DataTypes.FIELD("_kafka_key", businessType))) + .build(); + TableDescriptor.Builder builder = + TableDescriptor.builder() + .schema(schema) + .distributedBy(1) + .property(ConfigOptions.TABLE_KAFKA_ENABLED, true); + // Read-time JSON defaults must remain omitted through CREATE and ALTER. + if ("nested".equals(layout)) { + builder.property(ConfigOptions.TABLE_KAFKA_CONVERT_VALUE_LAYOUT.key(), layout); + } + return builder; + } + + @Test + void testOnlyRequestedKafkaPropertiesArePersisted() throws Exception { + TablePath path = TablePath.of("ddl", "roundtrip"); + Schema schema = + Schema.newBuilder() + .column("_kafka_key", DataTypes.BYTES()) + .column( + "payload", + DataTypes.ROW(DataTypes.FIELD("nested", DataTypes.STRING()))) + .column( + "_kafka_headers", + DataTypes.MAP(DataTypes.STRING(), DataTypes.BYTES())) + .column("_kafka_rescue", DataTypes.STRING()) + .build(); + TableDescriptor descriptor = + TableDescriptor.builder() + .schema(schema) + .distributedBy(1) + .property(ConfigOptions.TABLE_KAFKA_ENABLED, true) + .customProperty("table.kafka.convert.value.type", "custom-value") + .customProperty("kafka.value.format", "custom-format") + .customProperty("fluss.owner", "team") + .build(); + createTable(CLUSTER, path, descriptor); + TableDescriptor stored = stored(path); + assertThat(stored.getSchema()).isEqualTo(schema); + assertThat(stored.getCustomProperties()).isEqualTo(descriptor.getCustomProperties()); + assertThat(kafkaProperties(stored)) + .isEqualTo(Collections.singletonMap("table.kafka.enabled", "true")); + assertThat(new KafkaTopicSchemaResolver().resolve(stored).valueProjection().positions()) + .containsExactly(1); + set(path, Collections.singletonMap("table.kafka.convert.key.type", "binary")); + assertThat(KafkaTableSchema.resolve(stored(path))) + .isEqualTo(KafkaTableSchema.resolve(stored)); + reset(path, "table.kafka.convert.key.type"); + assertThat(stored(path)).isEqualTo(stored); + TableDescriptor unknownOption = + TableDescriptor.builder(descriptor) + .property("table.kafka.unknown", "anything") + .build(); + assertThatThrownBy( + () -> + gateway.createTable( + newCreateTableRequest( + TablePath.of("ddl", "unknown_option"), + unknownOption, + false)) + .get()) + .hasCauseInstanceOf(InvalidConfigException.class) + .hasStackTraceContaining("table.kafka.unknown"); + } + + @Test + void testEnableDisableAndRejectInvalidProperties() throws Exception { + TablePath path = TablePath.of("ddl", "enable_disable"); + TableDescriptor disabled = + enabled().property(ConfigOptions.TABLE_KAFKA_ENABLED, false).build(); + createTable(CLUSTER, path, disabled); + TableDescriptor before = stored(path); + assertThat(KafkaTableSchema.isEnabled(before)).isFalse(); + set(path, Collections.singletonMap("table.kafka.enabled", "true")); + TableDescriptor enabled = stored(path); + assertThat(KafkaTableSchema.isEnabled(enabled)).isTrue(); + Map expected = new HashMap<>(before.getProperties()); + expected.put("table.kafka.enabled", "true"); + assertThat(enabled.getProperties()).isEqualTo(expected); + assertThat(enabled.getSchema()).isEqualTo(before.getSchema()); + assertThatThrownBy( + () -> + set( + path, + Collections.singletonMap( + "table.kafka.convert.key.type", "binary"))) + .hasCauseInstanceOf(InvalidTableException.class) + .hasStackTraceContaining("requires _kafka_key BYTES"); + assertThat(stored(path)).isEqualTo(enabled); + assertThatThrownBy( + () -> + set( + path, + Collections.singletonMap( + "table.kafka.convert.value.type", "string"))) + .hasCauseInstanceOf(InvalidTableException.class); + assertThat(stored(path)).isEqualTo(enabled); + assertThatThrownBy(() -> reset(path, "table.kafka.convert.value.type")) + .hasCauseInstanceOf(InvalidTableException.class); + assertThat(stored(path)).isEqualTo(enabled); + reset(path, "table.kafka.enabled"); + assertThat(KafkaTableSchema.isEnabled(stored(path))).isFalse(); + assertThat(stored(path).getProperties()).doesNotContainKey("table.kafka.enabled"); + // Disabled tables can stage incompatible Kafka conversion properties without Kafka + // validation. + set(path, Collections.singletonMap("table.kafka.convert.value.type", "string")); + TableDescriptor staged = stored(path); + assertThatThrownBy(() -> set(path, Collections.singletonMap("table.kafka.enabled", "true"))) + .hasCauseInstanceOf(InvalidTableException.class); + assertThat(stored(path)).isEqualTo(staged); + } + + @ParameterizedTest + @ValueSource(strings = {"binary", "string"}) + void testNonJsonLayoutDoesNotChangeMapping(String type) throws Exception { + TablePath path = TablePath.of("ddl", type + "_layout"); + TableDescriptor descriptor = + enabled() + .schema( + Schema.newBuilder() + .column( + "_kafka_value", + "binary".equals(type) + ? DataTypes.BYTES() + : DataTypes.STRING()) + .build()) + .property(ConfigOptions.TABLE_KAFKA_CONVERT_VALUE_TYPE.key(), type) + .build(); + createTable(CLUSTER, path, descriptor); + TableDescriptor before = stored(path); + KafkaTableSchema expected = KafkaTableSchema.resolve(before); + assertThat(expected.valueConversionLayout()).isNull(); + assertThat(stored(path).getProperties()) + .doesNotContainKey("table.kafka.convert.value.layout"); + for (KafkaValueConversionLayout layout : KafkaValueConversionLayout.values()) { + set(path, Collections.singletonMap("table.kafka.convert.value.layout", layout.name())); + assertThat(KafkaTableSchema.resolve(stored(path))).isEqualTo(expected); + assertThat(stored(path).getProperties()) + .containsEntry("table.kafka.convert.value.layout", layout.name()); + } + reset(path, "table.kafka.convert.value.layout"); + assertThat(stored(path)).isEqualTo(before); + } + + @Test + void testSchemaAlterValidatesBeforePersistenceAndCanBeDisabled() throws Exception { + TablePath path = TablePath.of("ddl", "schema_change"); + createTable(CLUSTER, path, enabled().build()); + GetTableInfoResponse before = info(path); + assertThatThrownBy(() -> addColumn(path, "_kafka_headers", DataTypes.STRING())) + .hasCauseInstanceOf(InvalidTableException.class); + assertThat(info(path).getTableJson()).isEqualTo(before.getTableJson()); + assertThat(info(path).getSchemaId()).isEqualTo(before.getSchemaId()); + addColumn(path, "_kafka_headers", DataTypes.MAP(DataTypes.STRING(), DataTypes.BYTES())); + assertThat(KafkaTableSchema.resolve(stored(path)).headersPosition()).isEqualTo(1); + assertThat(info(path).getSchemaId()).isGreaterThan(before.getSchemaId()); + addColumn(path, "_kafka_metadata", metadataType()); + assertThat(KafkaTableSchema.resolve(stored(path)).metadataPosition()).isEqualTo(2); + TableDescriptor beforeFailedRescue = stored(path); + assertThatThrownBy(() -> addColumn(path, "_kafka_rescue", DataTypes.INT())) + .hasCauseInstanceOf(InvalidTableException.class); + assertThat(stored(path)).isEqualTo(beforeFailedRescue); + set(path, Collections.singletonMap("table.kafka.enabled", "false")); + addColumn(path, "unmapped_business_field", DataTypes.INT()); + assertThat(stored(path).getSchema().getRowType().getFieldNames()) + .contains("unmapped_business_field"); + assertThatThrownBy(() -> set(path, Collections.singletonMap("table.kafka.enabled", "true"))) + .hasCauseInstanceOf(InvalidTableException.class); + assertThat(KafkaTableSchema.isEnabled(stored(path))).isFalse(); + } + + @Test + void testMetadataCanBeCreatedAndEnabledWithoutOffset() throws Exception { + Schema schema = + Schema.newBuilder() + .column("_kafka_metadata", metadataType()) + .column("_kafka_value", DataTypes.BYTES()) + .build(); + TableDescriptor descriptor = enabled().schema(schema).build(); + TablePath created = TablePath.of("ddl", "metadata_create"); + createTable(CLUSTER, created, descriptor); + assertThat(stored(created).getSchema()).isEqualTo(schema); + + TablePath altered = TablePath.of("ddl", "metadata_enable"); + createTable( + CLUSTER, + altered, + TableDescriptor.builder(descriptor) + .property(ConfigOptions.TABLE_KAFKA_ENABLED, false) + .build()); + int schemaId = info(altered).getSchemaId(); + set(altered, Collections.singletonMap("table.kafka.enabled", "true")); + assertThat(info(altered).getSchemaId()).isEqualTo(schemaId); + assertThat(stored(altered).getSchema()).isEqualTo(schema); + assertThat(KafkaTableSchema.resolve(stored(altered)).metadataPosition()).isZero(); + } + + @Test + void testMetadataOffsetFieldRejectedOnCreateAndEnable() throws Exception { + gateway.createDatabase(newCreateDatabaseRequest("ddl", true)).get(); + TablePath path = TablePath.of("ddl", "metadata_with_offset"); + Schema schema = + Schema.newBuilder() + .column("_kafka_value", DataTypes.BYTES()) + .column( + "_kafka_metadata", + DataTypes.ROW( + DataTypes.FIELD("timestamp", DataTypes.TIMESTAMP_LTZ(3)), + DataTypes.FIELD("timestamp_type", DataTypes.INT()), + DataTypes.FIELD("partition", DataTypes.INT()), + DataTypes.FIELD("offset", DataTypes.BIGINT()))) + .build(); + assertThatThrownBy( + () -> + gateway.createTable( + newCreateTableRequest( + path, + enabled().schema(schema).build(), + false)) + .get()) + .hasCauseInstanceOf(InvalidTableException.class) + .hasStackTraceContaining( + "ROW"); + assertThat(gateway.tableExists(newTableExistsRequest(path)).get().isExists()).isFalse(); + createTable( + CLUSTER, + path, + enabled() + .schema(schema) + .property(ConfigOptions.TABLE_KAFKA_ENABLED, false) + .build()); + TableDescriptor before = stored(path); + assertThatThrownBy(() -> set(path, Collections.singletonMap("table.kafka.enabled", "true"))) + .hasCauseInstanceOf(InvalidTableException.class) + .hasStackTraceContaining( + "ROW"); + assertThat(stored(path)).isEqualTo(before); + } + + @Test + void testCanonicalNameCheckedOnCreateAndEnable() throws Exception { + String database = String.join("", Collections.nCopies(128, "d")); + TablePath path = TablePath.of(database, String.join("", Collections.nCopies(128, "t"))); + gateway.createDatabase(newCreateDatabaseRequest(database, true)).get(); + assertThatThrownBy( + () -> + gateway.createTable( + newCreateTableRequest( + path, enabled().build(), false)) + .get()) + .hasCauseInstanceOf(InvalidTableException.class) + .hasStackTraceContaining("Kafka topic"); + createTable( + CLUSTER, + path, + enabled().property(ConfigOptions.TABLE_KAFKA_ENABLED, false).build()); + TableDescriptor before = stored(path); + assertThatThrownBy(() -> set(path, Collections.singletonMap("table.kafka.enabled", "true"))) + .hasCauseInstanceOf(InvalidTableException.class); + assertThat(stored(path)).isEqualTo(before); + } + + private static DataType metadataType() { + return DataTypes.ROW( + DataTypes.FIELD("timestamp", DataTypes.TIMESTAMP_LTZ(3)), + DataTypes.FIELD("timestamp_type", DataTypes.INT()), + DataTypes.FIELD("partition", DataTypes.INT())); + } + + private GetTableInfoResponse info(TablePath path) throws Exception { + return gateway.getTableInfo(newGetTableInfoRequest(path)).get(); + } + + private TableDescriptor stored(TablePath path) throws Exception { + return TableDescriptor.fromJsonBytes(info(path).getTableJson()); + } + + private void set(TablePath path, Map properties) throws Exception { + gateway.alterTable( + newAlterTableRequest( + path, + properties, + Collections.emptyList(), + Collections.emptyList(), + false)) + .get(); + } + + private void reset(TablePath path, String property) throws Exception { + gateway.alterTable( + newAlterTableRequest( + path, + Collections.emptyMap(), + Collections.singletonList(property), + Collections.emptyList(), + false)) + .get(); + } + + private void addColumn(TablePath path, String name, DataType type) throws Exception { + PbAddColumn column = + new PbAddColumn() + .setColumnName(name) + .setDataTypeJson( + JsonSerdeUtils.writeValueAsBytes(type, DataTypeJsonSerde.INSTANCE)) + .setColumnPositionType(0); + AlterTableRequest request = + newAlterTableRequest( + path, + Collections.emptyMap(), + Collections.emptyList(), + Collections.singletonList(column), + false); + gateway.alterTable(request).get(); + } + + private static Map kafkaProperties(TableDescriptor table) { + Map properties = new HashMap<>(); + table.getProperties() + .forEach( + (key, value) -> { + if (key.startsWith("table.kafka.")) { + properties.put(key, value); + } + }); + return properties; + } + + private static TableDescriptor.Builder enabled() { + return TableDescriptor.builder() + .schema(Schema.newBuilder().column("_kafka_value", DataTypes.BYTES()).build()) + .distributedBy(1) + .logFormat(LogFormat.ARROW) + .property(ConfigOptions.TABLE_KAFKA_ENABLED, true) + .property( + ConfigOptions.TABLE_KAFKA_CONVERT_VALUE_TYPE, + KafkaValueConversionType.BINARY); + } +} diff --git a/fluss-kafka/src/test/java/org/apache/fluss/kafka/schema/KafkaTopicSchemaResolverTest.java b/fluss-kafka/src/test/java/org/apache/fluss/kafka/schema/KafkaTopicSchemaResolverTest.java new file mode 100644 index 00000000000..7b0d18edbb3 --- /dev/null +++ b/fluss-kafka/src/test/java/org/apache/fluss/kafka/schema/KafkaTopicSchemaResolverTest.java @@ -0,0 +1,68 @@ +/* + * 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.schema; + +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.types.DataTypes; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests the Kafka adapter's use of the shared DDL contract. */ +class KafkaTopicSchemaResolverTest { + private final KafkaTopicSchemaResolver resolver = new KafkaTopicSchemaResolver(); + + @Test + void testCompatibleSchemaDoesNotEnableTable() { + assertThat(resolver.isEnabled(table().build())).isFalse(); + assertThatThrownBy(() -> resolver.resolve(table().build())) + .isInstanceOf(KafkaTopicSchemaException.class) + .hasMessageContaining("disabled"); + } + + @Test + void testInvalidEnabledContractIsNotTreatedAsDisabled() { + TableDescriptor table = + table().property(ConfigOptions.TABLE_KAFKA_ENABLED, true) + .property( + ConfigOptions.TABLE_KAFKA_CONVERT_VALUE_TYPE.key(), "schema_latest") + .build(); + assertThat(resolver.isEnabled(table)).isTrue(); + assertThatThrownBy(() -> resolver.resolve(table)) + .isInstanceOf(KafkaTopicSchemaException.class) + .hasMessageContaining("schema_latest"); + } + + @Test + void testMalformedPersistedEnableFlagIsAnExplicitContractError() { + TableDescriptor table = table().property("table.kafka.enabled", "maybe").build(); + assertThatThrownBy(() -> resolver.isEnabled(table)) + .isInstanceOf(KafkaTopicSchemaException.class) + .hasMessageContaining("table.kafka.enabled"); + } + + private static TableDescriptor.Builder table() { + return TableDescriptor.builder() + .schema(Schema.newBuilder().column("_kafka_value", DataTypes.BYTES()).build()) + .property(ConfigOptions.TABLE_KAFKA_CONVERT_VALUE_TYPE.key(), "binary"); + } +} diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java index 2cdd2c5bf97..faa1367a9f9 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java @@ -64,7 +64,9 @@ import java.util.ArrayList; import java.util.Collection; +import java.util.HashSet; import java.util.List; +import java.util.Set; import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; @@ -548,17 +550,22 @@ private interface ThrowingRunnable { private void createDefaultDatabase() { MetadataManager metadataManager = new MetadataManager(zkClient, conf, lakeCatalogDynamicLoader); - List databases = metadataManager.listDatabases(); + createDefaultDatabases(metadataManager, conf); + } + + @VisibleForTesting + static void createDefaultDatabases(MetadataManager metadataManager, Configuration conf) { + Set databases = new HashSet<>(metadataManager.listDatabases()); if (databases.isEmpty()) { metadataManager.createDatabase(DEFAULT_DATABASE, DatabaseDescriptor.EMPTY, true); + databases.add(DEFAULT_DATABASE); LOG.info("Created default database '{}' because no database exists.", DEFAULT_DATABASE); } - // create Kafka default database if Kafka is enabled. if (conf.get(ConfigOptions.KAFKA_ENABLED)) { - String kafkaDB = conf.get(ConfigOptions.KAFKA_DATABASE); - if (!databases.contains(kafkaDB)) { - metadataManager.createDatabase(kafkaDB, DatabaseDescriptor.EMPTY, true); - LOG.info("Created default database '{}' for Kafka protocol.", kafkaDB); + String kafkaDatabase = conf.get(ConfigOptions.KAFKA_DEFAULT_DATABASE); + if (!databases.contains(kafkaDatabase)) { + metadataManager.createDatabase(kafkaDatabase, DatabaseDescriptor.EMPTY, true); + LOG.info("Created default database '{}' for Kafka protocol.", kafkaDatabase); } } } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java index 6bed0ecd3f0..df8713338a8 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java @@ -507,7 +507,7 @@ public CompletableFuture createTable(CreateTableRequest req // validate table descriptor before creating table in lake or fluss metadata, // to avoid orphaned lake tables when validation fails - metadataManager.validateTableDescriptor(tableDescriptor); + metadataManager.validateTableDescriptor(tablePath, tableDescriptor); // the distribution and bucket count must be set now //noinspection OptionalGetWithoutIsPresent diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java index 14b97c927eb..52a89625dac 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java @@ -39,6 +39,7 @@ import org.apache.fluss.metadata.DatabaseDescriptor; import org.apache.fluss.metadata.DatabaseInfo; import org.apache.fluss.metadata.DatabaseSummary; +import org.apache.fluss.metadata.KafkaTableSchema; import org.apache.fluss.metadata.LakeTableUtil; import org.apache.fluss.metadata.MergeEngineType; import org.apache.fluss.metadata.ResolvedPartitionSpec; @@ -123,12 +124,13 @@ public MetadataManager( this.lakeCatalogDynamicLoader = lakeCatalogDynamicLoader; } - /** Validates the table descriptor. */ - public void validateTableDescriptor(TableDescriptor tableDescriptor) { + /** Validates a descriptor and its Kafka topic identity before persistence. */ + public void validateTableDescriptor(TablePath tablePath, TableDescriptor tableDescriptor) { TableDescriptorValidation.validateTableDescriptor( tableDescriptor, maxBucketNum, lakeCatalogDynamicLoader.getLakeCatalogContainer().getDataLakeFormat()); + KafkaTableSchema.validateTablePath(tablePath, tableDescriptor); } public void createDatabase( @@ -852,7 +854,7 @@ private void doAlterTablePropertiesOnce( } // reuse the same validate logic with the createTable() method - validateTableDescriptor(newDescriptor); + validateTableDescriptor(tablePath, newDescriptor); beforeUpdate.accept(tableInfo, newDescriptor); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java index de40fbf01a8..2b5d7d82c83 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java @@ -32,6 +32,7 @@ import org.apache.fluss.metadata.ChangelogImage; import org.apache.fluss.metadata.DataLakeFormat; import org.apache.fluss.metadata.DeleteBehavior; +import org.apache.fluss.metadata.KafkaTableSchema; import org.apache.fluss.metadata.KvFormat; import org.apache.fluss.metadata.LogFormat; import org.apache.fluss.metadata.MergeEngineType; @@ -122,6 +123,8 @@ public static void validateTableDescriptor( } } + KafkaTableSchema.validate(tableDescriptor); + // check distribution checkDistribution(tableDescriptor, maxBucketNum); checkPrimaryKey(tableDescriptor); @@ -155,6 +158,10 @@ public static void validateAlterTableSchema(TableInfo table, Schema newSchema) { } else { validateNoAggregationFunctions(newSchema); } + TableDescriptor updated = + TableDescriptor.builder(table.toTableDescriptor()).schema(newSchema).build(); + KafkaTableSchema.validate(updated); + KafkaTableSchema.validateTablePath(table.getTablePath(), updated); } private static void checkTableLakeFormatMatchesCluster( diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/KafkaDefaultDatabaseTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/KafkaDefaultDatabaseTest.java new file mode 100644 index 00000000000..ed97d0b5251 --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/KafkaDefaultDatabaseTest.java @@ -0,0 +1,122 @@ +/* + * 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.server.coordinator; + +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.metadata.DatabaseDescriptor; +import org.apache.fluss.server.zk.NOPErrorHandler; +import org.apache.fluss.server.zk.ZooKeeperClient; +import org.apache.fluss.server.zk.ZooKeeperExtension; +import org.apache.fluss.server.zk.data.ZkData.DatabasesZNode; +import org.apache.fluss.testutils.common.AllCallbackWrapper; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.ValueSource; + +import static org.apache.fluss.server.coordinator.CoordinatorServer.DEFAULT_DATABASE; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +/** Tests startup database creation for native and Kafka default database configurations. */ +class KafkaDefaultDatabaseTest { + + @RegisterExtension + public static final AllCallbackWrapper ZK = + new AllCallbackWrapper<>(new ZooKeeperExtension()); + + private MetadataManager manager; + + @BeforeEach + void setup() throws Exception { + ZooKeeperClient zk = ZK.getCustomExtension().getZooKeeperClient(NOPErrorHandler.INSTANCE); + if (zk.getStat(DatabasesZNode.path()).isPresent()) { + zk.getCuratorClient() + .delete() + .deletingChildrenIfNeeded() + .forPath(DatabasesZNode.path()); + } + manager = + new MetadataManager( + zk, + new Configuration(), + new LakeCatalogDynamicLoader(new Configuration(), null, true)); + } + + @ParameterizedTest + @CsvSource({"false, false", "false, true", "true, false", "true, true"}) + void testNativeAndKafkaDefaultInitialization(boolean hasDatabase, boolean kafkaEnabled) { + if (hasDatabase) { + manager.createDatabase("existing", DatabaseDescriptor.EMPTY, false); + } + Configuration config = kafkaConfiguration(DEFAULT_DATABASE); + config.set(ConfigOptions.KAFKA_ENABLED, kafkaEnabled); + MetadataManager observed = spy(manager); + CoordinatorServer.createDefaultDatabases(observed, config); + CoordinatorServer.createDefaultDatabases(observed, config); + boolean createsDefault = !hasDatabase || kafkaEnabled; + assertThat(manager.listDatabases()) + .containsExactlyInAnyOrder( + !hasDatabase + ? new String[] {DEFAULT_DATABASE} + : kafkaEnabled + ? new String[] {"existing", DEFAULT_DATABASE} + : new String[] {"existing"}); + verify(observed, times(createsDefault ? 1 : 0)) + .createDatabase(DEFAULT_DATABASE, DatabaseDescriptor.EMPTY, true); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testCreateMissingCustomDefaultAndPreserveExisting(boolean exists) { + String database = "kafka_default"; + manager.createDatabase("existing", DatabaseDescriptor.EMPTY, false); + DatabaseDescriptor descriptor = + DatabaseDescriptor.builder().comment("existing default").build(); + if (exists) { + manager.createDatabase(database, descriptor, false); + } + Configuration config = kafkaConfiguration(database); + CoordinatorServer.createDefaultDatabases(manager, config); + CoordinatorServer.createDefaultDatabases(manager, config); + assertThat(manager.listDatabases()).containsExactlyInAnyOrder("existing", database); + assertThat(manager.getDatabase(database).getDatabaseDescriptor().toJsonBytes()) + .isEqualTo((exists ? descriptor : DatabaseDescriptor.EMPTY).toJsonBytes()); + } + + @Test + void testDisabledKafkaDoesNotCreateConfiguredDatabase() { + Configuration config = kafkaConfiguration("kafka_default"); + config.set(ConfigOptions.KAFKA_ENABLED, false); + CoordinatorServer.createDefaultDatabases(manager, config); + assertThat(manager.listDatabases()).containsExactly(DEFAULT_DATABASE); + } + + private static Configuration kafkaConfiguration(String database) { + Configuration config = new Configuration(); + config.set(ConfigOptions.KAFKA_ENABLED, true); + config.set(ConfigOptions.KAFKA_DEFAULT_DATABASE, database); + return config; + } +} diff --git a/website/docs/kafka-compatibility.md b/website/docs/kafka-compatibility.md new file mode 100644 index 00000000000..bfb7aa47fde --- /dev/null +++ b/website/docs/kafka-compatibility.md @@ -0,0 +1,147 @@ +--- +sidebar_label: Kafka Compatibility +title: Kafka Compatibility +sidebar_position: 5.5 +--- + +# Kafka Compatibility + +Fluss can expose eligible Fluss tables as Kafka topics. Create the table in Fluss and enable its +Kafka mapping with `table.kafka.enabled=true`. + +## Supported tables + +Kafka mapping supports non-partitioned [Log Tables](table-design/table-types/log-table.md) with +the Arrow log format and no primary key. CREATE and ALTER validate the mapping before saving it. +All declared columns and nested value fields must be nullable, including ROW fields, ARRAY +elements, MAP values, and system columns. MAP keys retain Fluss's non-null constraint. The four +system columns are optional and can be declared independently. + +## Cluster configuration + +Enable Kafka on the cluster and configure a Kafka listener on each TabletServer. For a local +TabletServer: + +```yaml title="server.yaml" +kafka.enabled: true +kafka.listener.names: KAFKA +kafka.default.database: fluss +bind.listeners: FLUSS://localhost:9123,KAFKA://localhost:9092 +``` + +For distributed deployments, configure reachable `bind.listeners` and `advertised.listeners`. +The coordinator and all TabletServers must use the same `kafka.default.database`; see the +[server configuration reference](maintenance/configuration.md#common). + +## Topic-to-table mapping + +Topic names use `.`. Names without a database prefix use +`kafka.default.database` (default: `fluss`). With the default setting, mappings are: + +| Kafka topic name | Fluss table path | +|---|---| +| `orders` | `fluss.orders` | +| `fluss.orders` | `fluss.orders` | +| `sales.orders` | `sales.orders` | +| `warehouse.events` | `warehouse.events` | + +The native `fluss` database is created when the cluster has no databases. With Kafka enabled, the +coordinator also creates the configured default database if it is missing, even in a non-empty +cluster. Create other databases and all tables through Fluss; Kafka CreateTopics and DeleteTopics +are not supported. + +A Kafka partition maps to the Fluss bucket with the same number, so the topic partition count must +match the table bucket count. + +## Create or enable a table + +The examples use an existing [Flink Fluss catalog](engine-flink/ddl.md#create-catalog). Create a +table for JSON values, optionally storing Kafka keys and headers: + +```sql title="Flink SQL" +CREATE DATABASE IF NOT EXISTS sales; + +CREATE TABLE sales.orders ( + order_id BIGINT, + amount DECIMAL(18, 2), + _kafka_key BYTES, + _kafka_headers MAP +) WITH ( + 'table.kafka.enabled' = 'true' +); +``` + +This maps the `sales.orders` topic to the table. JSON members map to `order_id` and `amount`; the +Kafka key and headers map to their system columns. With no conversion properties, values use +`parse_json` with the `flat` layout. + +Enable Kafka mapping for an existing compatible table with: + +```sql title="Flink SQL" +ALTER TABLE sales.orders SET ( + 'table.kafka.enabled' = 'true' +); +``` + +Set the property to `false` to disable Kafka mapping. Disabled tables are not subject to Kafka +schema validation. + +## Conversion properties + +Set these table properties with CREATE or ALTER TABLE. + +| Option | Type | Default | Description | +|---|---|---|---| +| `table.kafka.enabled` | Boolean | `false` | Enable Kafka mapping for the table. | +| `table.kafka.convert.value.type` | Enum | `parse_json` | Convert values as `binary`, `string`, or `parse_json`. | +| `table.kafka.convert.value.layout` | Enum | `flat` | JSON layout: `flat` maps top-level columns; `nested` maps fields in `_kafka_value ROW<...>`. Ignored for binary and string. | +| `table.kafka.convert.key.type` | Enum | None | Optional `binary` key conversion; requires `_kafka_key BYTES`. | + +Declare `_kafka_key BYTES` to store Kafka keys as bytes; `table.kafka.convert.key.type` can be +omitted. To discard keys, omit both the column and the property. + +| Value conversion | Required schema | Mapping | +|---|---|---| +| `binary` | `_kafka_value BYTES` | Store the value as bytes. | +| `string` | `_kafka_value STRING` | Decode the value as UTF-8. | +| `parse_json` with `flat` | Top-level business columns | Map JSON object members to columns by name. | +| `parse_json` with `nested` | `_kafka_value ROW<...>` | Map JSON members to fields in the value ROW. | + +Binary, string, and nested JSON use `_kafka_value`; other top-level columns must be optional +system columns. Flat JSON uses top-level business columns and cannot declare `_kafka_value`. +JSON values must be objects. Nested ROW, ARRAY, and MAP schemas are supported. System names are +reserved at the table's top level, but are ordinary names inside business ROWs. + +JSON schema limits: + +- MAP keys must be STRING, including nested MAPs. +- Nesting depth is at most 64. Each top-level business field starts at depth zero; ROW fields, + ARRAY elements, and MAP values each add one. The JSON root and `_kafka_value` container do not + count. +- Fields must be nullable as described above. Value constraints are checked when records are + decoded. + +## System columns + +Declare only the Kafka information the table needs to store. Unmapped keys, headers, and metadata +are not stored in row fields. `_kafka_value` is the conversion target described above. + +| Column | Type | Mapping | +|---|---|---| +| `_kafka_key` | `BYTES` | Kafka record key. | +| `_kafka_headers` | `MAP` | Kafka headers; the last value wins for duplicate names. | +| `_kafka_metadata` | `ROW` | Input timestamp, timestamp type, and requested partition. | +| `_kafka_rescue` | Nullable `STRING` | Compact JSON for unknown or unconvertible fields and flat-layout system-name conflicts. | + +The metadata ROW must use the field names, order, and types shown above; the ROW and its fields +must be nullable. Timestamp type is `-1` when absent, `0` for CreateTime, or `1` for LogAppendTime. +An absent timestamp maps to null. Log offsets remain native log metadata and are not included in +this ROW. + +Use the optional `_kafka_rescue` column to retain JSON fields that cannot be mapped to the table +schema for later inspection. It stores unknown fields, values that cannot be converted to the +declared type, and fields whose names conflict with system columns in the flat layout as compact +JSON. When a value cannot be converted, the corresponding table field is set to null while valid +fields are preserved. Without `_kafka_rescue`, these errors cause conversion to fail. Invalid JSON +and non-object roots always fail conversion. The column is null for binary/string conversion or +when there is nothing to rescue. diff --git a/website/docs/maintenance/configuration.md b/website/docs/maintenance/configuration.md index c88f83ccf08..624e670938a 100644 --- a/website/docs/maintenance/configuration.md +++ b/website/docs/maintenance/configuration.md @@ -259,7 +259,10 @@ Kafka protocol compatibility is still in development. |--------------------------------|----------|---------|--------------------------------------------------------------------------------------------------------------------| | kafka.enabled | Boolean | false | Whether enable Fluss Kafka. Disabled by default. When this option is set to true, the Fluss Kafka will be enabled. | | kafka.listener.names | String | KAFKA | The listener names for Kafka wire protocol communication. Support multiple listener names, separated by comma. | -| kafka.database | String | kafka | The database for Fluss Kafka. The default database is `kafka`. | +| kafka.default.database | String | fluss | Database used for unqualified Kafka topic names. When Kafka is enabled, the coordinator creates this database if missing, including `fluss`. | Kafka listeners share `netty.connection.max-idle-time` with native Fluss RPC listeners (see [Netty](#netty)). The default is `10min`; connections close when neither reads nor writes occur for the configured duration. + +See [Kafka Compatibility](../kafka-compatibility.md) for cluster setup, table properties, +topic-to-table mapping, value conversion, and system columns.