[FLINK-17149][debezium][canal] Introduce Debezium and Canal changelog format #12152

wuchong · 2020-05-14T12:11:19Z

What is the purpose of the change

Introduce DebeziumFormatFactory and CanalFormatFactory to read changelogs.

CREATE TABLE my_table (
  ...
) WITH (
  'connector'='...',  -- e.g. 'kafka'
  'format'='debezium-json',
  'debezium-json.schema-include'='true' -- default false, Debeizum can be configured to include or exclude the message schema
  'debezium-json.ignore-parse-errors'='true' -- default false
);

CREATE TABLE my_table (
  ...
) WITH (
 'connector'='...', -- e.g. 'kafka'
 'format'='canal-json',
 'canal-json.ignore-parse-errors'='true' -- default false
);

Brief change log

Introduced DebeziumJsonDeserializationSchema and DebeziumFormatFactory to deserialize Debezium messages.
Introduced CanalJsonDeserializationSchema and CanalFormatFactory to deserialize Canal messages.
Throw exception in DebeziumFormatFactory#createSinkFormat and CanalFormatFactory#createSinkFormat to tell users they can't be as sink for now.

Verifying this change

Added DebeziumJsonDeserializationSchemaTest and CanalsonDeserializationSchemaTest to deserialize debezium messages and canal messages.
Added DebeziumFormatFactoryTest and CanalFormatFactoryTest to unit test factories.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? not documented

wuchong · 2020-05-14T12:11:40Z

cc @danny0405

flinkbot · 2020-05-14T12:17:34Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 81d1f17 (Thu May 14 12:17:34 UTC 2020)

Warnings:

2 pom.xml files were touched: Check for build and licensing issues.
No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

Details

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2020-05-14T13:00:08Z

CI report:

aae6bd7 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

…ading debezium changelogs

…ng canal changelogs

danny0405 · 2020-05-15T07:38:16Z

.../src/main/java/org/apache/flink/formats/json/debezium/DebeziumJsonDeserializationSchema.java

+		} else {
+			if (!ignoreParseErrors) {
+				throw new IOException(format("Failed to deserialize Debezium JSON '%s'.", new String(message)));
+			}


Give more detailed message about why the deserialization fails.

danny0405 · 2020-05-15T07:51:14Z

...s/flink-json/src/main/java/org/apache/flink/formats/json/debezium/DebeziumFormatFactory.java

+ */
+public class DebeziumFormatFactory implements DeserializationFormatFactory, SerializationFormatFactory {
+
+	public static final String IDENTIFIER = "debezium-json";


DebeziumJsonFormatFactory or maybe we can make the underlying format configurable.

danny0405 · 2020-05-15T07:58:55Z

...k-json/src/main/java/org/apache/flink/formats/json/canal/CanalJsonDeserializationSchema.java

+		RowData row = jsonDeserializer.deserialize(message);
+		String type = row.getString(2).toString(); // "type" field
+		if (OP_INSERT.equals(type)) {
+			// "data" field is an array of row, contains inserted rows


Replace the if else clause with switch case seems better.

It's not easy to rewrite in switch case, because we can't allocate local variable with the same name.

danny0405 · 2020-05-15T08:00:47Z

...k-json/src/main/java/org/apache/flink/formats/json/canal/CanalJsonDeserializationSchema.java

+					if (before.isNullAt(f)) {
+						// not null fields in "old" (before) means the fields are changed
+						// null/empty fields in "old" (before) means the fields not not changed
+						// so we just copy the not changed fields into before


not not changed => are not changed

danny0405 · 2020-05-15T08:02:47Z

...k-json/src/main/java/org/apache/flink/formats/json/canal/CanalJsonDeserializationSchema.java

+		} else {
+			if (!ignoreParseErrors) {
+				throw new IOException(format("Failed to deserialize Canal JSON '%s'.", new String(message)));
+			}


Give more detailed exception for an intentionally thrown exception.

danny0405 · 2020-05-15T08:10:59Z

...formats/flink-json/src/main/java/org/apache/flink/formats/json/canal/CanalFormatFactory.java

+ */
+public class CanalFormatFactory implements DeserializationFormatFactory, SerializationFormatFactory {
+
+	public static final String IDENTIFIER = "canal-json";


CanalJsonFormatFactory ? or makes the underlying format configurable because Canal default format is Protobuf

danny0405 · 2020-05-15T08:11:49Z

...on/src/test/java/org/apache/flink/formats/json/canal/CanalJsonDeserializationSchemaTest.java

+		}
+
+		// Debezium captures change data (`debezium-data-schema-include.txt`) on the `product` table:
+		//


wuchong · 2020-05-15T08:51:58Z

Thanks for the reviewing @danny0405 . PR updated.

danny0405

+1, LGTM.

wuchong · 2020-05-15T14:34:57Z

Thanks for the reviewing @danny0405 .

Merging...

wuchong · 2020-05-15T15:11:56Z

The failed kafka case is related to FLINK-16383.

rmetzger added review=description? component=Formats component=TableSQL/Ecosystem labels May 14, 2020

wuchong added 2 commits May 15, 2020 09:19

[FLINK-17149][json][debezium] Introduce Debezium format to support re…

103839d

…ading debezium changelogs

[FLINK-17150][json][debezium] Introduce Canal format to support readi…

4ddbfe9

…ng canal changelogs

wuchong force-pushed the changelog-formats branch from 81d1f17 to 4ddbfe9 Compare May 15, 2020 01:19

danny0405 reviewed May 15, 2020

View reviewed changes

address Danny's review comments.

aae6bd7

danny0405 approved these changes May 15, 2020

View reviewed changes

wuchong closed this in d2b8ebb May 15, 2020

wuchong deleted the changelog-formats branch May 15, 2020 15:09

[FLINK-17149][debezium][canal] Introduce Debezium and Canal changelog format #12152

[FLINK-17149][debezium][canal] Introduce Debezium and Canal changelog format #12152

Uh oh!

Conversation

wuchong commented May 14, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

wuchong commented May 14, 2020

Uh oh!

flinkbot commented May 14, 2020

Automated Checks

Review Progress

Uh oh!

flinkbot commented May 14, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

danny0405 May 15, 2020

Choose a reason for hiding this comment

Uh oh!

danny0405 May 15, 2020

Choose a reason for hiding this comment

Uh oh!

danny0405 May 15, 2020

Choose a reason for hiding this comment

Uh oh!

wuchong May 15, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

danny0405 May 15, 2020

Choose a reason for hiding this comment

Uh oh!

danny0405 May 15, 2020

Choose a reason for hiding this comment

Uh oh!

danny0405 May 15, 2020

Choose a reason for hiding this comment

Uh oh!

danny0405 May 15, 2020

Choose a reason for hiding this comment

Uh oh!

wuchong commented May 15, 2020

Uh oh!

danny0405 left a comment

Choose a reason for hiding this comment

Uh oh!

wuchong commented May 15, 2020

Uh oh!

wuchong commented May 15, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

wuchong commented May 14, 2020 •

edited

Loading

flinkbot commented May 14, 2020 •

edited

Loading

wuchong May 15, 2020 •

edited

Loading