feat(process): introduce record processor factory and enrich conversion/transform pipeline (#2796)

* feat(process): introduce flexible record processor factory and enrich conversion/transform pipeline

- Add RecordProcessorFactory to support dynamic creation of processors based on schema and transform configs
- Refactor RegistryConverterFactory for improved schema format handling and converter instantiation
- Implement SchemaFormat, TableTopicConvertType, and TableTopicTransformType enums for config-driven processing
- Enhance Converter interface and conversion records to include key, value, and timestamp fields
- Refactor AvroRegistryConverter and ProtobufRegistryConverter to return unified RecordData objects
- Add ProtoToAvroConverter for robust Protobuf-to-Avro conversion
- Update transform chain: add KafkaMetadataTransform for metadata enrichment, refactor DebeziumUnwrapTransform
- Update DefaultRecordProcessor and TransformContext to support partition-aware processing
- Improve error handling and code clarity across conversion and transform modules

Author: Gezi-lzq

checkstyle/suppressions.xml

@@ -378,5 +378,6 @@
     <suppress id="dontUseSystemExit"
               files="(BenchTool|S3Utils|AutoMQCLI).java"/>
     <suppress checks="ClassDataAbstractionCoupling" files="(StreamControlManagerTest|ControllerStreamManager).java"/>
+    <suppress files="core[\/]src[\/]test[\/]java[\/]kafka[\/]automq[\/]table[\/]process[\/]proto[\/].*\.java$" checks=".*"/>
 
 </suppressions>

clients/src/main/java/org/apache/kafka/common/config/TopicConfig.java

@@ -265,8 +265,39 @@ public class TopicConfig {
     public static final String TABLE_TOPIC_COMMIT_INTERVAL_DOC = "The table topic commit interval(ms)";
     public static final String TABLE_TOPIC_NAMESPACE_CONFIG = "automq.table.topic.namespace";
     public static final String TABLE_TOPIC_NAMESPACE_DOC = "The table topic table namespace";
+
     public static final String TABLE_TOPIC_SCHEMA_TYPE_CONFIG = "automq.table.topic.schema.type";
-    public static final String TABLE_TOPIC_SCHEMA_TYPE_DOC = "The table topic schema type, support schemaless, schema";
+    public static final String TABLE_TOPIC_SCHEMA_TYPE_DOC = "[DEPRECATED] The table topic schema type configuration. " +
+        "This configuration is deprecated and will be removed in a future release. " +
+        "Please use the new separate converter and transform configurations instead. " +
+        "Supported values: 'schemaless' (maps to convert.value.type=raw, transform.value.type=none), " +
+        "'schema' (maps to convert.value.type=by_schema_id, transform.value.type=flatten).";
+
+    public static final String AUTOMQ_TABLE_TOPIC_CONVERT_VALUE_TYPE_CONFIG = "automq.table.topic.convert.value.type";
+    public static final String AUTOMQ_TABLE_TOPIC_CONVERT_VALUE_TYPE_DOC = "How to parse Kafka record values. " +
+        "Supported: 'raw', 'string', 'by_schema_id', 'by_latest_schema'. " +
+        "Schema Registry URL required for 'by_schema_id' and 'by_latest_schema'.";
+    public static final String AUTOMQ_TABLE_TOPIC_CONVERT_KEY_TYPE_CONFIG = "automq.table.topic.convert.key.type";
+    public static final String AUTOMQ_TABLE_TOPIC_CONVERT_KEY_TYPE_DOC = "How to parse Kafka record keys. " +
+        "Supported: 'raw', 'string', 'by_schema_id', 'by_latest_schema'. " +
+        "Schema Registry URL required for 'by_schema_id' and 'by_latest_schema'.";
+
+    public static final String AUTOMQ_TABLE_TOPIC_CONVERT_VALUE_BY_LATEST_SCHEMA_PREFIX = "automq.table.topic.convert.value.by_latest_schema.";
+    public static final String AUTOMQ_TABLE_TOPIC_CONVERT_KEY_BY_LATEST_SCHEMA_PREFIX = "automq.table.topic.convert.key.by_latest_schema.";
+
+    public static final String AUTOMQ_TABLE_TOPIC_CONVERT_KEY_SUBJECT_CONFIG = "automq.table.topic.convert.key.subject";
+    public static final String AUTOMQ_TABLE_TOPIC_CONVERT_KEY_SUBJECT_DOC = "The Schema Registry subject name for key schemas. " +
+        "Defaults to '{topic-name}-key' if not specified.";
+    public static final String AUTOMQ_TABLE_TOPIC_CONVERT_KEY_MESSAGE_FULL_NAME_CONFIG = "automq.table.topic.convert.key.message.full.name";
+    public static final String AUTOMQ_TABLE_TOPIC_CONVERT_KEY_MESSAGE_FULL_NAME_DOC = "The fully qualified message name for Protobuf key schemas. " +
+        "Used when schema contains multiple message types. Defaults to first message type.";
+
+    public static final String AUTOMQ_TABLE_TOPIC_TRANSFORM_VALUE_TYPE_CONFIG = "automq.table.topic.transform.value.type";
+    public static final String AUTOMQ_TABLE_TOPIC_TRANSFORM_VALUE_TYPE_DOC = "Transformation to apply to the record value after conversion. " +
+        "Supported: 'none', 'flatten' (extract fields from structured records), " +
+        "'flatten_debezium' (process Debezium CDC events). " +
+        "Note: 'flatten_debezium' requires schema-based conversion.";
+
     public static final String TABLE_TOPIC_ID_COLUMNS_CONFIG = "automq.table.topic.id.columns";
     public static final String TABLE_TOPIC_ID_COLUMNS_DOC = "The primary key, comma-separated list of columns that identify a row in tables."
         + "ex. [region, name]";

core/src/main/java/kafka/automq/table/deserializer/SchemaResolutionResolver.java

@@ -21,8 +21,6 @@
 
 import kafka.automq.table.deserializer.proto.schema.MessageIndexes;
 
-import org.apache.kafka.common.record.Record;
-
 import java.nio.ByteBuffer;
 
 
@@ -45,7 +43,7 @@ public interface SchemaResolutionResolver {
     SchemaResolution resolve(String topic, ByteBuffer payload);
 
 
-    int getSchemaId(String topic, Record record);
+    int getSchemaId(String topic, ByteBuffer payload);
 
     /**
      * Container class for resolved schema information.

core/src/main/java/kafka/automq/table/deserializer/proto/HeaderBasedSchemaResolutionResolver.java

@@ -24,7 +24,6 @@
 import kafka.automq.table.process.exception.InvalidDataException;
 
 import org.apache.kafka.common.errors.SerializationException;
-import org.apache.kafka.common.record.Record;
 
 import java.nio.ByteBuffer;
 
@@ -59,9 +58,9 @@ public SchemaResolution resolve(String topic, ByteBuffer payload) {
     }
 
     @Override
-    public int getSchemaId(String topic, Record record) {
+    public int getSchemaId(String topic, ByteBuffer payload) {
         // io.confluent.kafka.serializers.DeserializationContext#constructor
-        return readSchemaId(record.value().duplicate());
+        return readSchemaId(payload.duplicate());
     }
 
     private int readSchemaId(ByteBuffer buffer) {

core/src/main/java/kafka/automq/table/deserializer/proto/LatestSchemaResolutionResolver.java

@@ -23,7 +23,6 @@
 import kafka.automq.table.deserializer.proto.schema.MessageIndexes;
 
 import org.apache.kafka.common.errors.SerializationException;
-import org.apache.kafka.common.record.Record;
 
 import com.automq.stream.utils.Time;
 
@@ -83,7 +82,7 @@ public SchemaResolution resolve(String topic, ByteBuffer payload) {
     }
 
     @Override
-    public int getSchemaId(String topic, Record record) {
+    public int getSchemaId(String topic, ByteBuffer payload) {
         LatestSchemaResolutionResolver.CachedSchemaInfo cachedInfo = getCachedSchemaInfo(subject);
         return cachedInfo.schemaId;
     }

core/src/main/java/kafka/automq/table/process/ConversionResult.java

@@ -19,63 +19,38 @@
 
 package kafka.automq.table.process;
 
-import org.apache.kafka.common.record.Record;
-
+import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericRecord;
 
 import java.util.Objects;
 
-/**
- * Result of format conversion operations.
- * Contains either a converted GenericRecord or error information.
- */
+
 public final class ConversionResult {
 
-    private final Record kafkaRecord;
-    private final GenericRecord valueRecord;
+    private final Object value;
+    private final Schema schema;
     private final String schemaIdentity;
 
-    /**
-     * Creates a successful conversion result.
-     *
-     * @param kafkaRecord the original Kafka record
-     * @param value the successfully converted Avro GenericRecord, must not be null
-     * @param schemaIdentity unique identifier for the record's schema
-     */
-    public ConversionResult(Record kafkaRecord, GenericRecord value, String schemaIdentity) {
-        this.kafkaRecord = Objects.requireNonNull(kafkaRecord);
-        this.valueRecord = Objects.requireNonNull(value);
-        this.schemaIdentity = schemaIdentity;
+    public ConversionResult(Object value, Schema schema, String schemaIdentity) {
+        this.value = value;
+        this.schema = Objects.requireNonNull(schema, "schema cannot be null");
+        this.schemaIdentity = Objects.requireNonNull(schemaIdentity, "schemaIdentity cannot be null");
     }
 
+    public ConversionResult(GenericRecord record, String schemaIdentity) {
+        this.value = Objects.requireNonNull(record, "record cannot be null");
+        this.schema = record.getSchema();
+        this.schemaIdentity = Objects.requireNonNull(schemaIdentity, "schemaIdentity cannot be null");
+    }
 
-    public Record getKafkaRecord() {
-        return kafkaRecord;
+    public Object getValue() {
+        return value;
     }
 
-    /**
-     * Returns the converted Avro GenericRecord.
-     *
-     * <p>This record represents the input data in standardized Avro format,
-     * ready for processing by the Transform chain. The record's schema contains
-     * all necessary type information for subsequent operations.</p>
-     *
-     * @return the converted GenericRecord, or null if conversion failed
-     */
-    public GenericRecord getValueRecord() {
-        return valueRecord;
+    public Schema getSchema() {
+        return schema;
     }
 
-    /**
-     * Returns the schema identity for the converted record.
-     *
-     * <p>The schema identity is a unique identifier for the record's schema,
-     * which can be used for schema evolution tracking, caching, and optimization
-     * purposes. The format and meaning of this identifier depends on the specific
-     * converter implementation.</p>
-     *
-     * @return the schema identity string, or null if conversion failed
-     */
     public String getSchemaIdentity() {
         return schemaIdentity;
     }

core/src/main/java/kafka/automq/table/process/Converter.java

@@ -20,70 +20,12 @@
 package kafka.automq.table.process;
 
 import kafka.automq.table.process.exception.ConverterException;
-import kafka.automq.table.process.exception.InvalidDataException;
 
-import org.apache.kafka.common.record.Record;
+import java.nio.ByteBuffer;
 
-import org.apache.avro.Schema;
-import org.apache.avro.SchemaBuilder;
-import org.apache.avro.generic.GenericRecord;
-import org.apache.avro.generic.GenericRecordBuilder;
 
-/**
- * Interface for a component that converts raw Kafka records into a standardized
- * format, typically Avro GenericRecord.
- *
- * <p>Implementations handle different data formats (e.g., Avro, JSON, Protobuf)
- * and may interact with services like a Schema Registry.</p>
- */
 public interface Converter {
 
-    String VALUE_FIELD_NAME = "value";
-    String RECORD_NAME = "ValueRecord";
-
-    /**
-     * Converts a raw Kafka record into a structured representation.
-     *
-     * <p>This method performs the core format conversion operation. It reads the
-     * byte[] payload from the Kafka record and converts it to an Avro GenericRecord
-     * that serves as the standardized internal representation for subsequent processing.</p>
-     *
-     * @param topic the name of the Kafka topic from which the record originated,
-     *              used for topic-specific schema resolution and routing
-     * @param record the Kafka record containing the byte[] payload to convert,
-     *               must not be null
-     * @return a {@code ConversionResult} containing the successfully converted data
-     * @throws IllegalArgumentException if topic or record is null
-     * @throws InvalidDataException if the input record data is invalid (e.g., null value)
-     * @throws ConverterException if a non-data-related error occurs during conversion (e.g., schema resolution failure)
-     */
-    ConversionResult convert(String topic, Record record) throws ConverterException;
+    ConversionResult convert(String topic, ByteBuffer buffer) throws ConverterException;
 
-    /**
-     * A convenience wrapper for {@link #buildValueRecord(Object, Schema)}.
-     * Extracts the schema from the given record.
-     *
-     * @param valueRecord the record to wrap.
-     * @return a new GenericRecord wrapping the value record.
-     */
-    static GenericRecord buildValueRecord(GenericRecord valueRecord) {
-        return buildValueRecord(valueRecord, valueRecord.getSchema());
-    }
-    /**
-     * A utility to build a standard record that wraps a value.
-     * The created record will have a single field named "value".
-     *
-     * @param value the value to wrap
-     * @param valueSchema the schema for the value
-     * @return a new GenericRecord wrapping the value
-     */
-    static GenericRecord buildValueRecord(Object value, Schema valueSchema) {
-        Schema schema = SchemaBuilder.record(RECORD_NAME)
-            .fields()
-            .name(VALUE_FIELD_NAME).type(valueSchema).noDefault()
-            .endRecord();
-        GenericRecordBuilder builder = new GenericRecordBuilder(schema);
-        builder.set(VALUE_FIELD_NAME, value);
-        return builder.build();
-    }
 }