This is an automated email from the ASF dual-hosted git repository.
nsivabalan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 4efc74739158 fix(streamer): override all deserialize() overloads in
KafkaAvroSchemaDeserializer (#18892)
4efc74739158 is described below
commit 4efc74739158b9210548aaeae276fcd7dc95f41d
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]>
---
.../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"
+}
