From 64523b0bae1468e2ebb62803791b7d79adb558c3 Mon Sep 17 00:00:00 2001 From: Christian Tzolov Date: Fri, 4 Dec 2020 19:42:10 +0100 Subject: [PATCH] Fix CDC issues - Fix cdc_key header string wrapping issue that causes problems with the spring.cloud.stream.kafka.default.producer.messageKeyExpression. - Rename Flattering property to the correct Flattening. - Remove wrong dependencies. --- common/cdc-debezium-boot-starter/README.adoc | 14 +++++++------- .../fn/common/cdc/CdcAutoConfiguration.java | 4 ++-- .../cdc/CdcBootStarterIntegrationTest.java | 6 +++--- common/cdc-debezium-common/pom.xml | 2 +- .../fn/common/cdc/CdcCommonConfiguration.java | 16 ++++++++-------- .../cloud/fn/common/cdc/CdcCommonProperties.java | 12 ++++++------ .../cloud/fn/common/cdc/EmbeddedEngine.java | 16 ++-------------- .../src/main/resources/application.properties | 2 +- supplier/cdc-debezium-supplier/pom.xml | 2 +- .../supplier/cdc/CdcSupplierConfiguration.java | 11 ++++++----- .../src/main/resources/application.properties | 2 +- 11 files changed, 38 insertions(+), 49 deletions(-) diff --git a/common/cdc-debezium-boot-starter/README.adoc b/common/cdc-debezium-boot-starter/README.adoc index aa43bc12..a274e652 100644 --- a/common/cdc-debezium-boot-starter/README.adoc +++ b/common/cdc-debezium-boot-starter/README.adoc @@ -94,14 +94,14 @@ cdc.config.database.port=3306 # <3> cdc.schema=false # <4> -cdc.flattering.enabled=true # <5> +cdc.flattening.enabled=true # <5> ---- <1> Metadata used to identify and dispatch the events received by this cdc consumer instance. <2> Configures the CDC Source to use https://debezium.io/docs/connectors/mysql/[MySqlConnector]. (equivalent to setting `cdc.config.connector.class=io.debezium.connector.mysql.MySqlConnector`). <3> Connector specific configurations. MySQL server logical name, connect location and access credentials. <4> Do not serialize the record's schema in the output messages. -<5> Enables the https://debezium.io/docs/configuration/event-flattening/[CDC Event Flattering] feature. +<5> Enables the https://debezium.io/docs/configuration/event-flattening/[CDC Event Flattening] feature. The full list of properties: @@ -110,11 +110,11 @@ The full list of properties: //tag::configuration-properties[] $$cdc.config$$:: $$Spring pass-trough wrapper for debezium configuration properties. All properties with a 'cdc.config.' prefix are native Debezium properties. The prefix is removed, converting them into Debezium io.debezium.config.Configuration.$$ *($$Map$$, default: `$$$$`)* $$cdc.connector$$:: $$Shortcut for the cdc.config.connector.class property. Either of those can be used as long as they do not contradict with each other.$$ *($$ConnectorType$$, default: `$$$$`, possible values: `mysql`,`postgres`,`mongodb`,`oracle`,`sqlserver`)* -$$cdc.flattering.add-fields$$:: $$Comma separated list of metadata fields to add to the flattened message. The fields will be prefixed with "__" or "__[<]struct]__", depending on the specification of the struct.$$ *($$String$$, default: `$$$$`)* -$$cdc.flattering.add-headers$$:: $$Comma separated list specify a list of metadata fields to add to the header of the flattened message. The fields will be prefixed with "__" or "__[struct]__".$$ *($$String$$, default: `$$$$`)* -$$cdc.flattering.delete-handling-mode$$:: $$Options for handling deleted records: (1) none - pass the records through, (2) drop - remove the records and (3) rewrite - add a '__deleted' field to the records.$$ *($$DeleteHandlingMode$$, default: `$$$$`, possible values: `drop`,`rewrite`,`none`)* -$$cdc.flattering.drop-tombstones$$:: $$By default Debezium generates tombstone records to enable Kafka compaction on deleted records. The dropTombstones can suppress the tombstone records.$$ *($$Boolean$$, default: `$$true$$`)* -$$cdc.flattering.enabled$$:: $$Enable flattering the source record events (https://debezium.io/docs/configuration/event-flattening).$$ *($$Boolean$$, default: `$$true$$`)* +$$cdc.flattening.add-fields$$:: $$Comma separated list of metadata fields to add to the flattened message. The fields will be prefixed with "__" or "__[<]struct]__", depending on the specification of the struct.$$ *($$String$$, default: `$$$$`)* +$$cdc.flattening.add-headers$$:: $$Comma separated list specify a list of metadata fields to add to the header of the flattened message. The fields will be prefixed with "__" or "__[struct]__".$$ *($$String$$, default: `$$$$`)* +$$cdc.flattening.delete-handling-mode$$:: $$Options for handling deleted records: (1) none - pass the records through, (2) drop - remove the records and (3) rewrite - add a '__deleted' field to the records.$$ *($$DeleteHandlingMode$$, default: `$$$$`, possible values: `drop`,`rewrite`,`none`)* +$$cdc.flattening.drop-tombstones$$:: $$By default Debezium generates tombstone records to enable Kafka compaction on deleted records. The dropTombstones can suppress the tombstone records.$$ *($$Boolean$$, default: `$$true$$`)* +$$cdc.flattening.enabled$$:: $$Enable flattening the source record events (https://debezium.io/docs/configuration/event-flattening).$$ *($$Boolean$$, default: `$$true$$`)* $$cdc.name$$:: $$Unique name for this sourceConnector instance.$$ *($$String$$, default: `$$$$`)* $$cdc.offset.commit-timeout$$:: $$Maximum number of milliseconds to wait for records to flush and partition offset data to be committed to offset storage before cancelling the process and restoring the offset data to be committed in a future attempt.$$ *($$Duration$$, default: `$$5000ms$$`)* $$cdc.offset.flush-interval$$:: $$Interval at which to try committing offsets. The default is 1 minute.$$ *($$Duration$$, default: `$$60000ms$$`)* diff --git a/common/cdc-debezium-boot-starter/src/main/java/org/springframework/cloud/fn/common/cdc/CdcAutoConfiguration.java b/common/cdc-debezium-boot-starter/src/main/java/org/springframework/cloud/fn/common/cdc/CdcAutoConfiguration.java index e073f0cc..14e51635 100644 --- a/common/cdc-debezium-boot-starter/src/main/java/org/springframework/cloud/fn/common/cdc/CdcAutoConfiguration.java +++ b/common/cdc-debezium-boot-starter/src/main/java/org/springframework/cloud/fn/common/cdc/CdcAutoConfiguration.java @@ -48,10 +48,10 @@ public class CdcAutoConfiguration { @Bean public EmbeddedEngineExecutorService embeddedEngine(EmbeddedEngine.Builder embeddedEngineBuilder, - Consumer sourceRecordConsumer, Function recordFlattering) { + Consumer sourceRecordConsumer, Function recordFlattening) { EmbeddedEngine embeddedEngine = embeddedEngineBuilder - .notifying(sourceRecord -> sourceRecordConsumer.accept(recordFlattering.apply(sourceRecord))) + .notifying(sourceRecord -> sourceRecordConsumer.accept(recordFlattening.apply(sourceRecord))) .build(); return new EmbeddedEngineExecutorService(embeddedEngine) { diff --git a/common/cdc-debezium-boot-starter/src/test/java/org/springframework/cloud/fn/common/cdc/CdcBootStarterIntegrationTest.java b/common/cdc-debezium-boot-starter/src/test/java/org/springframework/cloud/fn/common/cdc/CdcBootStarterIntegrationTest.java index ff183b74..b37cc327 100644 --- a/common/cdc-debezium-boot-starter/src/test/java/org/springframework/cloud/fn/common/cdc/CdcBootStarterIntegrationTest.java +++ b/common/cdc-debezium-boot-starter/src/test/java/org/springframework/cloud/fn/common/cdc/CdcBootStarterIntegrationTest.java @@ -70,7 +70,7 @@ public class CdcBootStarterIntegrationTest { "spring.datasource.type=com.zaxxer.hikari.HikariDataSource", "cdc.name=my-sql-connector", "cdc.schema=false", - "cdc.flattering.enabled=true", + "cdc.flattening.enabled=true", "cdc.stream.header.offset=true", "cdc.connector=mysql", "cdc.config.database.user=debezium", @@ -85,8 +85,8 @@ public class CdcBootStarterIntegrationTest { public void consumerTest() { contextRunner .withPropertyValues( - "cdc.flattering.deleteHandlingMode=drop", - "cdc.flattering.dropTombstones=true") + "cdc.flattening.deleteHandlingMode=drop", + "cdc.flattening.dropTombstones=true") .run(context -> { TestCdcApplication.TestSourceRecordConsumer testConsumer = context .getBean(TestCdcApplication.TestSourceRecordConsumer.class); diff --git a/common/cdc-debezium-common/pom.xml b/common/cdc-debezium-common/pom.xml index efe17d63..88861be9 100644 --- a/common/cdc-debezium-common/pom.xml +++ b/common/cdc-debezium-common/pom.xml @@ -15,7 +15,7 @@ Change Data Capture (CDC) Debezium Common - 1.2.1.Final + 1.3.1.Final 42.2.5 5.2.1.RELEASE diff --git a/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/CdcCommonConfiguration.java b/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/CdcCommonConfiguration.java index 434b8e7a..0f4efc4c 100644 --- a/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/CdcCommonConfiguration.java +++ b/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/CdcCommonConfiguration.java @@ -52,9 +52,9 @@ public class CdcCommonConfiguration { } @Bean - public Function recordFlattering(CdcCommonProperties properties, + public Function recordFlattening(CdcCommonProperties properties, ExtractNewRecordState extractNewRecordState) { - return sourceRecord -> properties.getFlattering().isEnabled() ? + return sourceRecord -> properties.getFlattening().isEnabled() ? (SourceRecord) extractNewRecordState.apply(sourceRecord) : sourceRecord; } @@ -62,13 +62,13 @@ public class CdcCommonConfiguration { public ExtractNewRecordState extractNewRecordState(CdcCommonProperties properties) { ExtractNewRecordState extractNewRecordState = new ExtractNewRecordState(); Map config = extractNewRecordState.config().defaultValues(); - config.put("drop.tombstones", properties.getFlattering().isDropTombstones()); - config.put("delete.handling.mode", properties.getFlattering().getDeleteHandlingMode().name()); - if (!StringUtils.isEmpty(properties.getFlattering().getAddHeaders())) { - config.put("add.headers", properties.getFlattering().getAddHeaders()); + config.put("drop.tombstones", properties.getFlattening().isDropTombstones()); + config.put("delete.handling.mode", properties.getFlattening().getDeleteHandlingMode().name()); + if (!StringUtils.isEmpty(properties.getFlattening().getAddHeaders())) { + config.put("add.headers", properties.getFlattening().getAddHeaders()); } - if (!StringUtils.isEmpty(properties.getFlattering().getAddFields())) { - config.put("add.fields", properties.getFlattering().getAddFields()); + if (!StringUtils.isEmpty(properties.getFlattening().getAddFields())) { + config.put("add.fields", properties.getFlattening().getAddFields()); } extractNewRecordState.configure(config); diff --git a/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/CdcCommonProperties.java b/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/CdcCommonProperties.java index 7756c75b..93e146d1 100644 --- a/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/CdcCommonProperties.java +++ b/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/CdcCommonProperties.java @@ -56,9 +56,9 @@ public class CdcCommonProperties { private boolean schema = false; /** - * Event Flattering (https://debezium.io/docs/configuration/event-flattening). + * Event Flattening (https://debezium.io/docs/configuration/event-flattening). */ - private final Flattering flattering = new Flattering(); + private final Flattening flattening = new Flattening(); /** * Spring pass-trough wrapper for debezium configuration properties. @@ -79,8 +79,8 @@ public class CdcCommonProperties { return offset; } - public Flattering getFlattering() { - return flattering; + public Flattening getFlattening() { + return flattening; } public Map getConfig() { @@ -238,10 +238,10 @@ public class CdcCommonProperties { * https://debezium.io/documentation/reference/0.10/configuration/event-flattening.html . * https://debezium.io/documentation/reference/0.10/configuration/event-flattening.html#configuration_options */ - public static class Flattering { + public static class Flattening { /** - * Enable flattering the source record events (https://debezium.io/docs/configuration/event-flattening). + * Enable flattening the source record events (https://debezium.io/docs/configuration/event-flattening). */ private boolean enabled = true; diff --git a/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/EmbeddedEngine.java b/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/EmbeddedEngine.java index 03fec9f9..624653be 100644 --- a/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/EmbeddedEngine.java +++ b/common/cdc-debezium-common/src/main/java/org/springframework/cloud/fn/common/cdc/EmbeddedEngine.java @@ -218,7 +218,6 @@ public final class EmbeddedEngine implements DebeziumEngine { public static final class BuilderImpl implements Builder { private OffsetBackingStore offsetBackingStore; - private SourceConnector sourceConnector; private Configuration config; private DebeziumEngine.ChangeConsumer handler; private ClassLoader classLoader; @@ -227,12 +226,6 @@ public final class EmbeddedEngine implements DebeziumEngine { private DebeziumEngine.ConnectorCallback connectorCallback; private OffsetCommitPolicy offsetCommitPolicy = null; - @Override - public Builder sourceConnector(SourceConnector sourceConnector) { - this.sourceConnector = sourceConnector; - return this; - } - @Override public Builder offsetBackingStore(OffsetBackingStore offsetBackingStore) { this.offsetBackingStore = offsetBackingStore; @@ -315,8 +308,7 @@ public final class EmbeddedEngine implements DebeziumEngine { Objects.requireNonNull(config, "A connector configuration must be specified."); Objects.requireNonNull(handler, "A connector consumer or changeHandler must be specified."); return new EmbeddedEngine(config, classLoader, clock, - handler, completionCallback, connectorCallback, offsetCommitPolicy, - sourceConnector, offsetBackingStore); + handler, completionCallback, connectorCallback, offsetCommitPolicy, offsetBackingStore); } // backward compatibility methods @@ -541,8 +533,6 @@ public final class EmbeddedEngine implements DebeziumEngine { @Override Builder using(OffsetCommitPolicy policy); - Builder sourceConnector(SourceConnector sourceConnector); - Builder offsetBackingStore(OffsetBackingStore offsetBackingStore); @Override @@ -576,7 +566,6 @@ public final class EmbeddedEngine implements DebeziumEngine { private long recordsSinceLastCommit = 0; private long timeOfLastCommitMillis = 0; private OffsetCommitPolicy offsetCommitPolicy; - private SourceConnector connector; private OffsetBackingStore offsetStore; private SourceTask task; @@ -584,8 +573,7 @@ public final class EmbeddedEngine implements DebeziumEngine { private EmbeddedEngine(Configuration config, ClassLoader classLoader, Clock clock, DebeziumEngine.ChangeConsumer handler, DebeziumEngine.CompletionCallback completionCallback, DebeziumEngine.ConnectorCallback connectorCallback, - OffsetCommitPolicy offsetCommitPolicy, SourceConnector sourceConnector, OffsetBackingStore offsetStore) { - this.connector = sourceConnector; + OffsetCommitPolicy offsetCommitPolicy, OffsetBackingStore offsetStore) { this.offsetStore = offsetStore; this.config = config; this.handler = handler; diff --git a/common/cdc-debezium-common/src/main/resources/application.properties b/common/cdc-debezium-common/src/main/resources/application.properties index 8fb76f9d..6bc13f69 100644 --- a/common/cdc-debezium-common/src/main/resources/application.properties +++ b/common/cdc-debezium-common/src/main/resources/application.properties @@ -1,2 +1,2 @@ spring.autoconfigure.exclude=org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration -spring.cloud.stream.kafka.default.producer.messageKeyExpression=headers['cdc_key'] +spring.cloud.stream.kafka.default.producer.messageKeyExpression=headers['cdc_key'].bytes diff --git a/supplier/cdc-debezium-supplier/pom.xml b/supplier/cdc-debezium-supplier/pom.xml index c1cfdb75..95a97976 100644 --- a/supplier/cdc-debezium-supplier/pom.xml +++ b/supplier/cdc-debezium-supplier/pom.xml @@ -14,7 +14,7 @@ CDC Debezium Suppliers - 1.2.1.Final + 1.3.1.Final 8.0.13 diff --git a/supplier/cdc-debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/cdc/CdcSupplierConfiguration.java b/supplier/cdc-debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/cdc/CdcSupplierConfiguration.java index 9055e3fa..ba546f23 100644 --- a/supplier/cdc-debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/cdc/CdcSupplierConfiguration.java +++ b/supplier/cdc-debezium-supplier/src/main/java/org/springframework/cloud/fn/supplier/cdc/CdcSupplierConfiguration.java @@ -111,13 +111,13 @@ public class CdcSupplierConfiguration implements BeanClassLoaderAware { public EmbeddedEngineExecutorService embeddedEngineExecutorService( EmbeddedEngine.Builder embeddedEngineBuilder, Function valueSerializer, Function keySerializer, - Function recordFlattering, + Function recordFlattening, ObjectMapper mapper, CdcSupplierProperties cdcStreamingEngineProperties) { FluxSink> sink = emitterProcessor.sink(); Consumer messageConsumer = sourceRecord -> { - // When cdc.flattering.deleteHandlingMode=none and cdc.flattering.dropTombstones=false + // When cdc.flattening.deleteHandlingMode=none and cdc.flattening.dropTombstones=false // then on deletion event an additional sourceRecord is sent with value Null. // Here we filter out such condition. if (sourceRecord == null) { @@ -130,7 +130,7 @@ public class CdcSupplierConfiguration implements BeanClassLoaderAware { // When the tombstone event is enabled, Debezium serializes the payload to null (e.g. // empty payload) // while the metadata information is carried through the headers (cdc_key). - // Note: Event for none flattered responses, when the cdc.config.tombstones.on.delete=true + // Note: Event for none flattened responses, when the cdc.config.tombstones.on.delete=true // (default), // tombstones are generate by Debezium and handled by the code below. if (cdcJsonPayload == null) { @@ -151,7 +151,8 @@ public class CdcSupplierConfiguration implements BeanClassLoaderAware { MessageBuilder messageBuilder = MessageBuilder .withPayload(cdcJsonPayload) - .setHeader("cdc_key", new String(key)) +// .setHeader("cdc_key", new String(key)) + .setHeader("cdc_key", key) .setHeader("cdc_topic", sourceRecord.topic()) .setHeader(MessageHeaders.CONTENT_TYPE, (cdcJsonPayload.equals(this.kafkaNull)) ? MimeTypeUtils.TEXT_PLAIN_VALUE @@ -182,7 +183,7 @@ public class CdcSupplierConfiguration implements BeanClassLoaderAware { }; EmbeddedEngine engine = embeddedEngineBuilder - .notifying(record -> messageConsumer.accept(recordFlattering.apply(record))) + .notifying(record -> messageConsumer.accept(recordFlattening.apply(record))) .build(); return new EmbeddedEngineExecutorService(engine); diff --git a/supplier/cdc-debezium-supplier/src/main/resources/application.properties b/supplier/cdc-debezium-supplier/src/main/resources/application.properties index 8fb76f9d..6bc13f69 100644 --- a/supplier/cdc-debezium-supplier/src/main/resources/application.properties +++ b/supplier/cdc-debezium-supplier/src/main/resources/application.properties @@ -1,2 +1,2 @@ spring.autoconfigure.exclude=org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration -spring.cloud.stream.kafka.default.producer.messageKeyExpression=headers['cdc_key'] +spring.cloud.stream.kafka.default.producer.messageKeyExpression=headers['cdc_key'].bytes