[FLINK-19275][connector-kafka] Support reading and writing Kafka meta… …data #13732

twalthr · 2020-10-21T16:32:45Z

What is the purpose of the change

This updates the KafkaDynamicSource and KafkaDynamicSink to read and
write metadata according to FLIP-107. Reading and writing metadata of formats
is not supported yet.

This PR is based on #13618.

Brief change log

Implement SupportsReadingMetadata and SupportsWritingMetadata in source and sink
Refactor Kafka source and sink accordingly
Various minor improvements (see separate commits)

Verifying this change

Added a test in KafkaTableITCase

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): yes
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? JavaDocs

flinkbot · 2020-10-21T16:35:57Z

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 3f19217 (Wed Oct 21 16:35:57 UTC 2020)

Warnings:

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.

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-10-21T17:10:54Z

CI report:

d5bc61d Azure: SUCCESS

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

wuchong

Thanks @twalthr . I left some comments.

wuchong · 2020-10-22T07:07:15Z

...flink-table-common/src/main/java/org/apache/flink/table/connector/sink/DynamicTableSink.java

 		 * @see TableSchema#toPhysicalRowDataType()
 		 */
-		TypeInformation<?> createTypeInformation(DataType consumedDataType);
+		<T> TypeInformation<T> createTypeInformation(DataType consumedDataType);


This is a nice improvement. But is this a compatible change?

No it is not a compatible change. But given that those interfaces are still relatively new and not many people have changed to the new sources/sinks. We should do this change now or never and avoid @SuppressWarning in almost all implementations.

wuchong · 2020-10-22T07:15:37Z

flink-core/src/main/java/org/apache/flink/types/Row.java

+	 * support comparing arrays stored in the values of a map. We might update the {@link #equals(Object)}
+	 * with this implementation in future versions.
+	 */
+	public static boolean deepEquals(Row row, Object other) {


The other should also be Row class? Because the Javadoc says "Compares two Rows".

wuchong · 2020-10-22T07:22:14Z

flink-core/src/main/java/org/apache/flink/types/Row.java

+	private static <E> boolean deepEqualsList(List<E> l1, List<?> l2) {
+		final Iterator<E> i1 = l1.iterator();
+		final Iterator<?> i2 = l2.iterator();
+		while (i1.hasNext() && i2.hasNext()) {


Why not compare size first?

I copied this implementation from java.util.AbstractList#equals but I don't have strong opinion on this. LinkedList are usually uncommon I guess.

wuchong · 2020-10-22T07:37:17Z

flink-core/src/main/java/org/apache/flink/types/Row.java

+	 *
+	 * <p>The current implementation of {@link Row#equals(Object)} is not able to compare all deeply
+	 * nested row structures that might be created in the table ecosystem. For example, it does not
+	 * support comparing arrays stored in the values of a map. We might update the {@link #equals(Object)}


It seems that we have already support this?

I might not understand your comment here.

The comment says "For example, it does not support comparing arrays stored in the values of a map", however, the tests prove that we have supported this.

I was referring to Row#equals(Object). And this doesn't support arrays in values of maps.

wuchong · 2020-10-22T07:57:15Z

flink-core/src/test/java/org/apache/flink/types/RowTest.java

+					true,
+					new Integer[]{1, null, 3, 99}, // diff here
+					Arrays.asList(1, null, 3),
+					originalMap,


This is a same map reference, create another map?

wuchong · 2020-10-22T08:45:41Z

...afka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSource.java

+		),
+
+		LEADER_EPOCH(
+			"leader-epoch",


I just find if we define metadata key with dash - separator, it has to be declared using FROM clause or escaped, e.g. leader_epoch INT FROM 'leader-epoch'. What do you think about changing the key to leader_epoch which is more SQL identifier compliant.

True, then I need to change also the recommendations in SupportsMetadata interfaces:

* <p>Metadata key names follow the same pattern as mentioned in {@link Factory}. In case of duplicate * names in format and source keys, format keys shall have higher precedence.

I would suggest to keep it as it is. Users can use backticks and leader epoch is not very frequently used. Furthermore, once we introduce metadata for formats such as debezium-json.ingestion-timestamp it would be confusing if the format identifier changes from debezium-json to debezium_json for metadata.

Is that possible to remove the prefix of debezium-json? The Javadoc also says "In case of duplicate names in format and source keys, format keys shall have higher precedence."

So far, the metadata keys of format and source are very different.

Yes, we can shorten them. The idea was to design the metadata similar to regular options. So if key and value are defined they would get a key. and value. prefix. That should be enough. We don't need the debezium-json.

But even for key. and value., users would need to use backticks or the FROM clause. I would stick to the naming convention to not cause confusion.

wuchong · 2020-10-22T08:46:41Z

...afka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSource.java

+
+	// --------------------------------------------------------------------------------------------
+
+	private static class MetadataKafkaDeserializationSchema implements KafkaDeserializationSchema<RowData> {


Declare serialVersionUID for this class.

wuchong · 2020-10-22T08:47:40Z

...afka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSource.java

+
+	// --------------------------------------------------------------------------------------------
+
+	private static final class MetadataAppendingCollector implements Collector<RowData>, Serializable {


Declare serialVersionUID for this class.

wuchong · 2020-10-22T08:49:25Z

...afka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSource.java

+		public void collect(RowData physicalRow) {
+			final int metadataArity = metadataConverters.length;
+			// shortcut if no metadata is required
+			if (metadataArity == 0) {


We can have a hasMetadata final member variable to allow JIT compiler optimization.

wuchong · 2020-10-22T09:00:22Z

...-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java

+				Row.of("data 3", 3, "CreateTime", LocalDateTime.parse("2020-03-10T13:12:11.123"), 2L, 0, headers3, 0, topic, true)
+		);
+
+		assertTrue(Row.deepEquals(expected, result));


What about add exception message for the assertion to print the expected and actual rows. That would be helpful for debugging Azure builds.

…nformation in sources and sinks

…asses

…data This updates the `KafkaDynamicSource` and `KafkaDynamicSink` to read and write metadata according to FLIP-107. Reading and writing metadata of formats is not supported yet.

twalthr · 2020-10-22T18:41:29Z

@wuchong I updated the PR, I hope I addressed most of your comments.

wuchong

LGTM.

+1 to merge

…data This updates the `KafkaDynamicSource` and `KafkaDynamicSink` to read and write metadata according to FLIP-107. Reading and writing metadata of formats is not supported yet. This closes apache#13732.

rmetzger added review=description? component=Connectors/Kafka component=TableSQL/Ecosystem labels Oct 21, 2020

wuchong reviewed Oct 22, 2020

View reviewed changes

twalthr added 6 commits October 22, 2020 19:48

[hotfix][table-common] Avoid unnecessary casting when creating type i…

d9ef907

…nformation in sources and sinks

[hotfix][connector-kafka] Merge KafkaITCase and KafkaTableTestBase

d6cacee

[hotfix][table-common] Add a new Row.deepEquals for all conversion cl…

d9628e5

…asses

Feedback addressed

7f594f5

[FLINK-19275][connector-kafka] Support reading and writing Kafka meta…

739b807

…data This updates the `KafkaDynamicSource` and `KafkaDynamicSink` to read and write metadata according to FLIP-107. Reading and writing metadata of formats is not supported yet.

Feedback addressed

d5bc61d

twalthr force-pushed the FLINK-19275_2 branch from 3f19217 to d5bc61d Compare October 22, 2020 18:34

wuchong approved these changes Oct 23, 2020

View reviewed changes

twalthr closed this in d21f5d9 Oct 23, 2020

fsk119 mentioned this pull request Sep 29, 2021

Expose record metadata to mongo-cdc apache/flink-cdc#476

Closed


		// --------------------------------------------------------------------------------------------

		private static class MetadataKafkaDeserializationSchema implements KafkaDeserializationSchema<RowData> {


		// --------------------------------------------------------------------------------------------

		private static final class MetadataAppendingCollector implements Collector<RowData>, Serializable {

Uh oh!

[FLINK-19275][connector-kafka] Support reading and writing Kafka meta… …data #13732

[FLINK-19275][connector-kafka] Support reading and writing Kafka meta… …data #13732

Uh oh!

Conversation

twalthr commented Oct 21, 2020

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!

flinkbot commented Oct 21, 2020

Automated Checks

Review Progress

Uh oh!

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

Uh oh!

CI report:

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

twalthr Oct 22, 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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

twalthr Oct 22, 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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

twalthr commented Oct 22, 2020

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

flinkbot commented Oct 21, 2020 •

edited

Loading

twalthr Oct 22, 2020 •

edited

Loading

twalthr Oct 22, 2020 •

edited

Loading