This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 4570ecdf2db7d314021ec2aad45fcca38e3736a0 Author: Sivabalan Narayanan <[email protected]> AuthorDate: Fri Jun 26 16:51:20 2026 -0700 fix(streamer): override all deserialize() overloads in KafkaAvroSchemaDeserializer (#18892) fix(streamer): prevent ArrayIndexOutOfBoundsException on Kafka Avro schema evolution KafkaAvroSchemaDeserializer only overrode deserialize(String, Boolean, byte[], Schema). The other overloads — deserialize(String, byte[]), (String, byte[], Schema), and (String, Headers, byte[]) — bypassed sourceSchema injection and fell back to the writer's schema, causing ArrayIndexOutOfBoundsException when consuming records serialized with an older (fewer-field) schema while the deserializer was configured with an evolved reader schema. Override all four overloads to route through super.deserialize(topic, false, bytes, sourceSchema) so Avro resolution consistently uses the configured reader schema regardless of the Kafka client's entry point. Closes #18891 --------- Co-authored-by: Claude Opus 4.7 (1M context) <[email protected]> (cherry picked from commit 4efc74739158b9210548aaeae276fcd7dc95f41d) --- .../deser/KafkaAvroSchemaDeserializer.java | 16 ++ .../deser/TestKafkaAvroSchemaDeserializer.java | 205 ++++++++++++++++ .../test/resources/schema/cdc_envelope_new.avsc | 261 +++++++++++++++++++++ .../test/resources/schema/cdc_envelope_old.avsc | 237 +++++++++++++++++++ 4 files changed, 719 insertions(+) diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java index a2ffb4878454..b1d0589150c2 100644 --- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java +++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deser/KafkaAvroSchemaDeserializer.java @@ -26,6 +26,7 @@ import io.confluent.kafka.serializers.KafkaAvroDeserializer; import lombok.NoArgsConstructor; import org.apache.avro.Schema; import org.apache.kafka.common.errors.SerializationException; +import org.apache.kafka.common.header.Headers; import java.util.Map; import java.util.Map.Entry; @@ -57,6 +58,21 @@ public class KafkaAvroSchemaDeserializer extends KafkaAvroDeserializer { } } + @Override + public Object deserialize(String topic, byte[] bytes) { + return this.deserialize(topic, false, bytes, sourceSchema); + } + + @Override + public Object deserialize(String topic, byte[] bytes, Schema readerSchema) { + return this.deserialize(topic, false, bytes, sourceSchema); + } + + @Override + public Object deserialize(String topic, Headers headers, byte[] bytes) { + return super.deserialize(topic, false, bytes, sourceSchema); + } + /** * We need to inject sourceSchema instead of reader schema during deserialization or later stages of the pipeline. * diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deser/TestKafkaAvroSchemaDeserializer.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deser/TestKafkaAvroSchemaDeserializer.java index 684672470433..01887573a91f 100644 --- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deser/TestKafkaAvroSchemaDeserializer.java +++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deser/TestKafkaAvroSchemaDeserializer.java @@ -30,8 +30,12 @@ import org.apache.avro.Schema; import org.apache.avro.generic.GenericData; import org.apache.avro.generic.GenericRecord; import org.apache.avro.generic.IndexedRecord; +import org.apache.kafka.common.header.internals.RecordHeaders; import org.junit.jupiter.api.Test; +import java.io.IOException; +import java.io.InputStream; +import java.nio.ByteBuffer; import java.util.HashMap; import java.util.Properties; @@ -125,4 +129,205 @@ public class TestKafkaAvroSchemaDeserializer { assertEquals(HoodieSchema.fromAvroSchema(actualRec.getSchema()), evolSchema); assertNull(genericRecord.get("age")); } + + private Schema loadSchemaFromResource(String resourcePath) throws IOException { + try (InputStream is = getClass().getClassLoader().getResourceAsStream(resourcePath)) { + return new Schema.Parser().parse(is); + } + } + + private static Schema getRecordTypeFromUnion(Schema unionSchema) { + return unionSchema.getTypes().stream() + .filter(s -> s.getType() == Schema.Type.RECORD) + .findFirst() + .orElseThrow(() -> new IllegalArgumentException("No record type in union")); + } + + private GenericRecord createCdcSourceRecord(Schema envelopeSchema) { + Schema sourceSchema = envelopeSchema.getField("source").schema(); + GenericRecord source = new GenericData.Record(sourceSchema); + source.put("version", "2.3.0.Final"); + source.put("connector", "mysql"); + source.put("name", "cdc"); + source.put("ts_ms", 1700000000000L); + source.put("snapshot", "false"); + source.put("db", "testdb"); + source.put("sequence", null); + source.put("table", "item"); + source.put("server_id", 1L); + source.put("gtid", null); + source.put("file", "binlog.000001"); + source.put("pos", 12345L); + source.put("row", 0); + source.put("thread", null); + source.put("query", null); + return source; + } + + private GenericRecord createCdcValueRecord(Schema envelopeSchema) { + Schema valueSchema = getRecordTypeFromUnion(envelopeSchema.getField("before").schema()); + GenericRecord value = new GenericData.Record(valueSchema); + value.put("id", ByteBuffer.wrap(new byte[]{1, 2, 3, 4})); + value.put("account_id", 42); + value.put("title", "Test Item"); + value.put("query", "test query"); + value.put("request_query_id", null); + value.put("page", null); + value.put("request_page_id", null); + value.put("source_rank_id", null); + value.put("property_id", 100); + value.put("created", "2024-01-01T00:00:00Z"); + value.put("created_by", 1); + value.put("updated", "2024-01-02T00:00:00Z"); + value.put("updated_by", 2); + value.put("version_id", null); + value.put("related_urls", null); + value.put("assignee", 5); + value.put("status", "IN_PROGRESS"); + value.put("tags", null); + value.put("deleted", false); + value.put("profile_id", null); + return value; + } + + private GenericRecord createCdcValueRecordBefore(Schema envelopeSchema) { + Schema valueSchema = getRecordTypeFromUnion(envelopeSchema.getField("before").schema()); + GenericRecord value = new GenericData.Record(valueSchema); + value.put("id", ByteBuffer.wrap(new byte[]{1, 2, 3, 4})); + value.put("account_id", 42); + value.put("title", "Old Item Title"); + value.put("query", "old query"); + value.put("request_query_id", null); + value.put("page", "https://example.com/old";); + value.put("request_page_id", null); + value.put("source_rank_id", 10); + value.put("property_id", 100); + value.put("created", "2024-01-01T00:00:00Z"); + value.put("created_by", 1); + value.put("updated", "2024-01-02T00:00:00Z"); + value.put("updated_by", 1); + value.put("version_id", "v1"); + value.put("related_urls", "[\"https://example.com\"]";); + value.put("assignee", 5); + value.put("status", "TO_DO"); + value.put("tags", null); + value.put("deleted", false); + value.put("profile_id", null); + return value; + } + + private GenericRecord createCdcEnvelopeRecord(Schema envelopeSchema, GenericRecord beforeRecord, GenericRecord afterRecord) { + GenericRecord envelope = new GenericData.Record(envelopeSchema); + envelope.put("before", beforeRecord); + envelope.put("after", afterRecord); + envelope.put("source", createCdcSourceRecord(envelopeSchema)); + envelope.put("op", "u"); + envelope.put("ts_ms", 1700000000000L); + envelope.put("transaction", null); + return envelope; + } + + /** + * Tests deserialization of a CDC (Debezium) envelope schema with schema evolution + * across all deserialize method overloads. The old schema lacks 4 fields + * (notes, search_engine_id, locale_id, language_id) in the nested Value record. + * When deserializing old records with the evolved schema, those new fields should default to null. + * + * Exercises: + * - deserialize(String topic, Boolean isKey, byte[] payload, Schema readerSchema) + * - deserialize(String topic, byte[] bytes) + * - deserialize(String topic, byte[] bytes, Schema readerSchema) + * - deserialize(String topic, Headers headers, byte[] bytes) + */ + @Test + public void testKafkaAvroSchemaDeserializerWithCdcEnvelopeEvolution() throws IOException { + Schema cdcOldSchema = loadSchemaFromResource("schema/cdc_envelope_old.avsc"); + Schema cdcNewSchema = loadSchemaFromResource("schema/cdc_envelope_new.avsc"); + + // Create and serialize records with old schema (no notes/search_engine_id/locale_id/language_id) + GenericRecord oldBefore = createCdcValueRecordBefore(cdcOldSchema); + GenericRecord oldAfter = createCdcValueRecord(cdcOldSchema); + GenericRecord oldEnvelope = createCdcEnvelopeRecord(cdcOldSchema, oldBefore, oldAfter); + byte[] bytesOldRecord = avroSerializer.serialize(topic, oldEnvelope); + + // Create and serialize records with evolved schema (new fields populated) + GenericRecord newBefore = createCdcValueRecordBefore(cdcNewSchema); + GenericRecord newAfter = createCdcValueRecord(cdcNewSchema); + newAfter.put("notes", "[{\"note\": \"test\"}]"); + newAfter.put("search_engine_id", "[\"default\"]"); + newAfter.put("locale_id", 1); + newAfter.put("language_id", 2); + GenericRecord newEnvelope = createCdcEnvelopeRecord(cdcNewSchema, newBefore, newAfter); + byte[] bytesNewRecord = avroSerializer.serialize(topic, newEnvelope); + + // Configure deserializer with evolved schema + config.put(AvroKafkaSource.KAFKA_AVRO_VALUE_DESERIALIZER_SCHEMA, cdcNewSchema.toString()); + KafkaAvroSchemaDeserializer deserializer = new KafkaAvroSchemaDeserializer(schemaRegistry, new HashMap(config)); + deserializer.configure(new HashMap(config), false); + + // === 1. deserialize(String, Boolean, byte[], Schema) — the existing override === + IndexedRecord evolvedDeserialized = (IndexedRecord) deserializer.deserialize(topic, false, bytesOldRecord, cdcNewSchema); + GenericRecord evolvedEnvelope = (GenericRecord) evolvedDeserialized; + assertEquals(cdcNewSchema, evolvedEnvelope.getSchema()); + assertEquals("u", evolvedEnvelope.get("op").toString()); + // Validate before record + GenericRecord beforeRecord = (GenericRecord) evolvedEnvelope.get("before"); + assertEquals("Old Item Title", beforeRecord.get("title").toString()); + assertEquals(42, beforeRecord.get("account_id")); + assertEquals("TO_DO", beforeRecord.get("status").toString()); + assertNull(beforeRecord.get(20)); + assertNull(beforeRecord.get(21)); + assertNull(beforeRecord.get(22)); + assertNull(beforeRecord.get(23)); + // Validate after record + GenericRecord afterRecord = (GenericRecord) evolvedEnvelope.get("after"); + assertEquals("Test Item", afterRecord.get("title").toString()); + assertNull(afterRecord.get("notes")); + assertNull(afterRecord.get("search_engine_id")); + assertNull(afterRecord.get("locale_id")); + assertNull(afterRecord.get("language_id")); + + // Evolved record via same method — should round-trip exactly + IndexedRecord newDeserialized = (IndexedRecord) deserializer.deserialize(topic, false, bytesNewRecord, cdcNewSchema); + assertEquals(newEnvelope, newDeserialized); + + // === 2. deserialize(String, byte[]) === + GenericRecord topicBytesResult = (GenericRecord) deserializer.deserialize(topic, bytesOldRecord); + GenericRecord tbBefore = (GenericRecord) topicBytesResult.get("before"); + assertEquals("Old Item Title", tbBefore.get("title").toString()); + assertNull(tbBefore.get(20)); + assertNull(tbBefore.get(21)); + assertNull(tbBefore.get(22)); + assertNull(tbBefore.get(23)); + + // === 3. deserialize(String, byte[], Schema) === + GenericRecord topicBytesSchemaResult = (GenericRecord) deserializer.deserialize(topic, bytesOldRecord, cdcNewSchema); + GenericRecord tbsBefore = (GenericRecord) topicBytesSchemaResult.get("before"); + assertEquals("Old Item Title", tbsBefore.get("title").toString()); + assertNull(tbsBefore.get(20)); + assertNull(tbsBefore.get(21)); + assertNull(tbsBefore.get(22)); + assertNull(tbsBefore.get(23)); + + // === 4. deserialize(String, Headers, byte[]) === + RecordHeaders headers = new RecordHeaders(); + GenericRecord headersResult = (GenericRecord) deserializer.deserialize(topic, headers, bytesOldRecord); + assertEquals(cdcNewSchema, headersResult.getSchema()); + GenericRecord hdrBefore = (GenericRecord) headersResult.get("before"); + assertEquals("Old Item Title", hdrBefore.get("title").toString()); + assertEquals(42, hdrBefore.get("account_id")); + assertNull(hdrBefore.get(20)); + assertNull(hdrBefore.get(21)); + assertNull(hdrBefore.get(22)); + assertNull(hdrBefore.get(23)); + + // New record via headers method + GenericRecord newHdrResult = (GenericRecord) deserializer.deserialize(topic, headers, bytesNewRecord); + GenericRecord newHdrAfter = (GenericRecord) newHdrResult.get("after"); + assertEquals("[{\"note\": \"test\"}]", newHdrAfter.get("notes").toString()); + assertEquals("[\"default\"]", newHdrAfter.get("search_engine_id").toString()); + assertEquals(1, newHdrAfter.get("locale_id")); + assertEquals(2, newHdrAfter.get("language_id")); + } + } diff --git a/hudi-utilities/src/test/resources/schema/cdc_envelope_new.avsc b/hudi-utilities/src/test/resources/schema/cdc_envelope_new.avsc new file mode 100644 index 000000000000..bf3ed9c0432d --- /dev/null +++ b/hudi-utilities/src/test/resources/schema/cdc_envelope_new.avsc @@ -0,0 +1,261 @@ +/* + * 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. + */ +{ + "type" : "record", + "name" : "Envelope", + "namespace" : "cdc.test.inventory.item", + "fields" : [ { + "name" : "before", + "type" : [ "null", { + "type" : "record", + "name" : "Value", + "fields" : [ { + "name" : "id", + "type" : "bytes" + }, { + "name" : "account_id", + "type" : "int" + }, { + "name" : "title", + "type" : "string" + }, { + "name" : "query", + "type" : [ "null", "string" ], + "default" : null + }, { + "name" : "request_query_id", + "type" : [ "null", "bytes" ], + "default" : null + }, { + "name" : "page", + "type" : [ "null", "string" ], + "default" : null + }, { + "name" : "request_page_id", + "type" : [ "null", "bytes" ], + "default" : null + }, { + "name" : "source_rank_id", + "type" : [ "null", "int" ], + "default" : null + }, { + "name" : "property_id", + "type" : [ "null", "int" ], + "default" : null + }, { + "name" : "created", + "type" : { + "type" : "string", + "connect.version" : 1, + "connect.default" : "1970-01-01T00:00:00Z", + "connect.name" : "io.debezium.time.ZonedTimestamp" + }, + "default" : "1970-01-01T00:00:00Z" + }, { + "name" : "created_by", + "type" : "int" + }, { + "name" : "updated", + "type" : { + "type" : "string", + "connect.version" : 1, + "connect.default" : "1970-01-01T00:00:00Z", + "connect.name" : "io.debezium.time.ZonedTimestamp" + }, + "default" : "1970-01-01T00:00:00Z" + }, { + "name" : "updated_by", + "type" : "int" + }, { + "name" : "version_id", + "type" : [ "null", "string" ], + "default" : null + }, { + "name" : "related_urls", + "type" : [ "null", { + "type" : "string", + "connect.version" : 1, + "connect.name" : "io.debezium.data.Json" + } ], + "default" : null + }, { + "name" : "assignee", + "type" : [ "null", "int" ], + "default" : null + }, { + "name" : "status", + "type" : { + "type" : "string", + "connect.version" : 1, + "connect.parameters" : { + "allowed" : "TO_DO,IN_PROGRESS,IN_REVIEW,APPROVED,PUBLISHED" + }, + "connect.default" : "TO_DO", + "connect.name" : "io.debezium.data.Enum" + }, + "default" : "TO_DO" + }, { + "name" : "tags", + "type" : [ "null", { + "type" : "string", + "connect.version" : 1, + "connect.name" : "io.debezium.data.Json" + } ], + "default" : null + }, { + "name" : "deleted", + "type" : { + "type" : "boolean", + "connect.default" : false + }, + "default" : false + }, { + "name" : "profile_id", + "type" : [ "null", "bytes" ], + "default" : null + }, { + "name" : "notes", + "type" : [ "null", { + "type" : "string", + "connect.version" : 1, + "connect.name" : "io.debezium.data.Json" + } ], + "default" : null + }, { + "name" : "search_engine_id", + "type" : [ "null", { + "type" : "string", + "connect.version" : 1, + "connect.name" : "io.debezium.data.Json" + } ], + "default" : null + }, { + "name" : "locale_id", + "type" : [ "null", "int" ], + "default" : null + }, { + "name" : "language_id", + "type" : [ "null", "int" ], + "default" : null + } ], + "connect.name" : "cdc.test.inventory.item.Value" + } ], + "default" : null + }, { + "name" : "after", + "type" : [ "null", "Value" ], + "default" : null + }, { + "name" : "source", + "type" : { + "type" : "record", + "name" : "Source", + "namespace" : "io.debezium.connector.mysql", + "fields" : [ { + "name" : "version", + "type" : "string" + }, { + "name" : "connector", + "type" : "string" + }, { + "name" : "name", + "type" : "string" + }, { + "name" : "ts_ms", + "type" : "long" + }, { + "name" : "snapshot", + "type" : [ { + "type" : "string", + "connect.version" : 1, + "connect.parameters" : { + "allowed" : "true,last,false,incremental" + }, + "connect.default" : "false", + "connect.name" : "io.debezium.data.Enum" + }, "null" ], + "default" : "false" + }, { + "name" : "db", + "type" : "string" + }, { + "name" : "sequence", + "type" : [ "null", "string" ], + "default" : null + }, { + "name" : "table", + "type" : [ "null", "string" ], + "default" : null + }, { + "name" : "server_id", + "type" : "long" + }, { + "name" : "gtid", + "type" : [ "null", "string" ], + "default" : null + }, { + "name" : "file", + "type" : "string" + }, { + "name" : "pos", + "type" : "long" + }, { + "name" : "row", + "type" : "int" + }, { + "name" : "thread", + "type" : [ "null", "long" ], + "default" : null + }, { + "name" : "query", + "type" : [ "null", "string" ], + "default" : null + } ], + "connect.name" : "io.debezium.connector.mysql.Source" + } + }, { + "name" : "op", + "type" : "string" + }, { + "name" : "ts_ms", + "type" : [ "null", "long" ], + "default" : null + }, { + "name" : "transaction", + "type" : [ "null", { + "type" : "record", + "name" : "block", + "namespace" : "event", + "fields" : [ { + "name" : "id", + "type" : "string" + }, { + "name" : "total_order", + "type" : "long" + }, { + "name" : "data_collection_order", + "type" : "long" + } ], + "connect.version" : 1, + "connect.name" : "event.block" + } ], + "default" : null + } ], + "connect.version" : 1, + "connect.name" : "cdc.test.inventory.item.Envelope" +} diff --git a/hudi-utilities/src/test/resources/schema/cdc_envelope_old.avsc b/hudi-utilities/src/test/resources/schema/cdc_envelope_old.avsc new file mode 100644 index 000000000000..3d2a832d5d3c --- /dev/null +++ b/hudi-utilities/src/test/resources/schema/cdc_envelope_old.avsc @@ -0,0 +1,237 @@ +/* + * 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. + */ +{ + "type" : "record", + "name" : "Envelope", + "namespace" : "cdc.test.inventory.item", + "fields" : [ { + "name" : "before", + "type" : [ "null", { + "type" : "record", + "name" : "Value", + "fields" : [ { + "name" : "id", + "type" : "bytes" + }, { + "name" : "account_id", + "type" : "int" + }, { + "name" : "title", + "type" : "string" + }, { + "name" : "query", + "type" : [ "null", "string" ], + "default" : null + }, { + "name" : "request_query_id", + "type" : [ "null", "bytes" ], + "default" : null + }, { + "name" : "page", + "type" : [ "null", "string" ], + "default" : null + }, { + "name" : "request_page_id", + "type" : [ "null", "bytes" ], + "default" : null + }, { + "name" : "source_rank_id", + "type" : [ "null", "int" ], + "default" : null + }, { + "name" : "property_id", + "type" : [ "null", "int" ], + "default" : null + }, { + "name" : "created", + "type" : { + "type" : "string", + "connect.version" : 1, + "connect.default" : "1970-01-01T00:00:00Z", + "connect.name" : "io.debezium.time.ZonedTimestamp" + }, + "default" : "1970-01-01T00:00:00Z" + }, { + "name" : "created_by", + "type" : "int" + }, { + "name" : "updated", + "type" : { + "type" : "string", + "connect.version" : 1, + "connect.default" : "1970-01-01T00:00:00Z", + "connect.name" : "io.debezium.time.ZonedTimestamp" + }, + "default" : "1970-01-01T00:00:00Z" + }, { + "name" : "updated_by", + "type" : "int" + }, { + "name" : "version_id", + "type" : [ "null", "string" ], + "default" : null + }, { + "name" : "related_urls", + "type" : [ "null", { + "type" : "string", + "connect.version" : 1, + "connect.name" : "io.debezium.data.Json" + } ], + "default" : null + }, { + "name" : "assignee", + "type" : [ "null", "int" ], + "default" : null + }, { + "name" : "status", + "type" : { + "type" : "string", + "connect.version" : 1, + "connect.parameters" : { + "allowed" : "TO_DO,IN_PROGRESS,IN_REVIEW,APPROVED,PUBLISHED" + }, + "connect.default" : "TO_DO", + "connect.name" : "io.debezium.data.Enum" + }, + "default" : "TO_DO" + }, { + "name" : "tags", + "type" : [ "null", { + "type" : "string", + "connect.version" : 1, + "connect.name" : "io.debezium.data.Json" + } ], + "default" : null + }, { + "name" : "deleted", + "type" : { + "type" : "boolean", + "connect.default" : false + }, + "default" : false + }, { + "name" : "profile_id", + "type" : [ "null", "bytes" ], + "default" : null + } ], + "connect.name" : "cdc.test.inventory.item.Value" + } ], + "default" : null + }, { + "name" : "after", + "type" : [ "null", "Value" ], + "default" : null + }, { + "name" : "source", + "type" : { + "type" : "record", + "name" : "Source", + "namespace" : "io.debezium.connector.mysql", + "fields" : [ { + "name" : "version", + "type" : "string" + }, { + "name" : "connector", + "type" : "string" + }, { + "name" : "name", + "type" : "string" + }, { + "name" : "ts_ms", + "type" : "long" + }, { + "name" : "snapshot", + "type" : [ { + "type" : "string", + "connect.version" : 1, + "connect.parameters" : { + "allowed" : "true,last,false,incremental" + }, + "connect.default" : "false", + "connect.name" : "io.debezium.data.Enum" + }, "null" ], + "default" : "false" + }, { + "name" : "db", + "type" : "string" + }, { + "name" : "sequence", + "type" : [ "null", "string" ], + "default" : null + }, { + "name" : "table", + "type" : [ "null", "string" ], + "default" : null + }, { + "name" : "server_id", + "type" : "long" + }, { + "name" : "gtid", + "type" : [ "null", "string" ], + "default" : null + }, { + "name" : "file", + "type" : "string" + }, { + "name" : "pos", + "type" : "long" + }, { + "name" : "row", + "type" : "int" + }, { + "name" : "thread", + "type" : [ "null", "long" ], + "default" : null + }, { + "name" : "query", + "type" : [ "null", "string" ], + "default" : null + } ], + "connect.name" : "io.debezium.connector.mysql.Source" + } + }, { + "name" : "op", + "type" : "string" + }, { + "name" : "ts_ms", + "type" : [ "null", "long" ], + "default" : null + }, { + "name" : "transaction", + "type" : [ "null", { + "type" : "record", + "name" : "block", + "namespace" : "event", + "fields" : [ { + "name" : "id", + "type" : "string" + }, { + "name" : "total_order", + "type" : "long" + }, { + "name" : "data_collection_order", + "type" : "long" + } ], + "connect.version" : 1, + "connect.name" : "event.block" + } ], + "default" : null + } ], + "connect.version" : 1, + "connect.name" : "cdc.test.inventory.item.Envelope" +}
