[HUDI-1648] Added custom kafka meta fields and custom kafka avro decoder. #2598

vburenin · 2021-02-23T23:42:03Z

What is the purpose of the pull request

This PR adds ability to:

Inject kafka meta information into the received data from avro kafka source, so users can track which kafka topic, partition, offset and the key delivered a specific record (super useful for debugging consistency issues). In SQL query user would need to specify <KAFKA_FIELDS> and enable appropriate options to enable the injection.
specify custom kafka avro data decoder that takes into account source latest schema and different schema versions.
cache retrieved source and target latest schemas
return null for the target schema if it is not in use by specifying "null" as a target schema url. (I know, there is NullTargetSchemaProvider, however it looks redundant).

Brief change log

Added kafka meta info injection
Kafka data deserializer can be swapped with a custom one.
source and target latest schema are optionally can being cached
if target schema url is set to "null", target schema will return null as a value forcing Hudi to infer the schema from the source one.

Verify this pull request

This change added tests and can be verified as follows:

Added unit tests for the components that were not originally covered.

Committer checklist

Has a corresponding JIRA in PR title & commit
Commit message is descriptive of the change
CI is green
Necessary doc changes done or have another open PR
For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.

codecov-io · 2021-02-24T00:08:42Z

Codecov Report

Merging #2598 (9bc13c2) into master (0dde7f9) will decrease coverage by 42.37%.
The diff coverage is 0.00%.

@@             Coverage Diff              @@
##             master   #2598       +/-   ##
============================================
- Coverage     51.55%   9.18%   -42.38%     
+ Complexity     3282      48     -3234     
============================================
  Files           445      55      -390     
  Lines         20317    2037    -18280     
  Branches       2102     251     -1851     
============================================
- Hits          10475     187    -10288     
+ Misses         8976    1837     -7139     
+ Partials        866      13      -853

Flag	Coverage Δ	Complexity Δ
hudicli	`?`	`?`
hudiclient	`?`	`?`
hudicommon	`?`	`?`
hudiflink	`?`	`?`
hudihadoopmr	`?`	`?`
hudisparkdatasource	`?`	`?`
hudisync	`?`	`?`
huditimelineservice	`?`	`?`
hudiutilities	`9.18% <0.00%> (-60.37%)`	`0.00 <0.00> (ø)`

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ	Complexity Δ
...i/utilities/deser/KafkaAvroSchemaDeserializer.java	`0.00% <0.00%> (ø)`	`0.00 <0.00> (?)`
.../hudi/utilities/schema/SchemaRegistryProvider.java	`0.00% <0.00%> (ø)`	`0.00 <0.00> (ø)`
...apache/hudi/utilities/sources/AvroKafkaSource.java	`0.00% <0.00%> (ø)`	`0.00 <0.00> (ø)`
...lities/sources/helpers/AvroKafkaSourceHelpers.java	`0.00% <0.00%> (ø)`	`0.00 <0.00> (?)`
.../utilities/transform/SqlQueryBasedTransformer.java	`0.00% <0.00%> (-75.00%)`	`0.00 <0.00> (-3.00)`
...va/org/apache/hudi/utilities/IdentitySplitter.java	`0.00% <0.00%> (-100.00%)`	`0.00% <0.00%> (-2.00%)`
...va/org/apache/hudi/utilities/schema/SchemaSet.java	`0.00% <0.00%> (-100.00%)`	`0.00% <0.00%> (-3.00%)`
...a/org/apache/hudi/utilities/sources/RowSource.java	`0.00% <0.00%> (-100.00%)`	`0.00% <0.00%> (-4.00%)`
.../org/apache/hudi/utilities/sources/AvroSource.java	`0.00% <0.00%> (-100.00%)`	`0.00% <0.00%> (-1.00%)`
.../org/apache/hudi/utilities/sources/JsonSource.java	`0.00% <0.00%> (-100.00%)`	`0.00% <0.00%> (-1.00%)`
... and 429 more

vinothchandar · 2021-02-24T19:11:01Z

@vburenin thanks for this. can we create a JIRA for this?

vinothchandar · 2021-02-24T19:12:11Z

the goals here all look good to me. Please give me and @nsivabalan sometime to review

vburenin · 2021-02-24T19:13:10Z

@vburenin thanks for this. can we create a JIRA for this?

It is in process, I may need to create multiple tickets. What I am really interested in right now is a code review so I do not spent too much time going forward and writing unit tests and then endup rolling it back.

nsivabalan · 2021-02-26T16:07:03Z

...-utilities/src/main/java/org/apache/hudi/utilities/decoders/KafkaAvroSchemaDeserializer.java

+        result = bytes;
+      } else {
+        int start = buffer.position() + buffer.arrayOffset();
+        DatumReader reader = new GenericDatumReader(schema, sourceSchema);


this looks good.

may I know whats the plan in keeping this file in sync with AbstractKafkaAvroDeserializer with version upgrades?

does this work.

@Override protected Object deserialize( boolean includeSchemaAndVersion, String topic, Boolean isKey, byte[] payload, Schema readerSchema) throws SerializationException { super.deserialize(includeSchemaAndVersion,topic,isKey,payload, sourceSchema); // pass sourceSchema as last arg instead of readerSchema }

It actually should work, there was a history behind this change that predates me that ended up with only sourceSchema change as I currently see. So yah, it will drastically simplify things. Thanks for catching it!

one more clarification @vburenin. Is this approach (passing sourceSchema instead of reader schema) is going to be useful for everyone(kakfa source users), even if they are not interested in meta fields to assist in schema evolution? If yes, wondering if we should make this customer deserializer as default kafka deserializer for hudi going forward?
Can you throw some light here.

It is definitely going to be useful for everybody. Kafka fields were added after this change was introduced. We had to keep up with the schema changes in the registry and this was the way to do that. The primary assumption, that is most likely true for every user, is that a schema is evolving... so, this is the way to keep up with the schema evolution.

...tilities/src/main/java/org/apache/hudi/utilities/sources/helpers/AvroKafkaSourceHelpers.java

hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/SchemaRegistryProvider.java

nsivabalan · 2021-02-26T16:17:31Z

...tilities/src/main/java/org/apache/hudi/utilities/sources/helpers/AvroKafkaSourceHelpers.java

+
+  public static final String INJECT_KAFKA_FIELDS = "hoodie.deltastreamer.source.inject_kafka_fields";
+
+  public static final String KAFKA_PARTITION = "_hudi_kafka_partition";


if we plan to provide this as generic solution for everyone, lets have consensus on meta fields naming. @vinothchandar .

+1, we also can make it configurable.

hudi-utilities/src/main/java/org/apache/hudi/utilities/schema/SchemaRegistryProvider.java

… with proper parameters. Renamed KAFKA_FIELD to KAFKA_META_FIELD

…ters is null, forced to add another constructor.

vburenin · 2021-03-02T19:30:46Z

Is being split into multiple PRs. Closing.

ouzhang · 2021-04-21T07:34:13Z

Hello. I notice that "[HUDI-1650] Custom avro kafka deserializer." has been merged. Is there another PR related to "custom kafka meta fields" ? Expecting to have this feature.

vburenin · 2021-04-21T13:15:32Z

There is no related PR yet. I've been planning to contribute this feature to upstream in Q2, so far it seems possible.

Added custom kafka fields and custom kafka avro decoder.

1068595

vinothchandar assigned vinothchandar and nsivabalan Feb 24, 2021

nsivabalan reviewed Feb 26, 2021

View reviewed changes

volodymyr.burenin added 3 commits February 26, 2021 12:26

Reduced KafkaAvroSchemaDeserializer to just call a super class method…

2d6112b

… with proper parameters. Renamed KAFKA_FIELD to KAFKA_META_FIELD

Added unit tests, minor code improvements.

2c86577

Addresed style issues.

1860266

vburenin changed the title ~~[WIP] Added custom kafka meta fields and custom kafka avro decoder.~~ [HUDI-1648] Added custom kafka meta fields and custom kafka avro decoder. Mar 1, 2021

nsivabalan added the priority:critical Production degraded; pipelines stalled label Mar 2, 2021

nsivabalan mentioned this pull request Mar 2, 2021

[Hudi 73] Adding support for vanilla AvroKafkaSource #2380

Closed

5 tasks

volodymyr.burenin added 2 commits March 2, 2021 09:26

Merge remote-tracking branch 'upstream/master' into custom-deserializer

c706d1c

Apparently reflection utils can't create a class if one of the parame…

9bc13c2

…ters is null, forced to add another constructor.

vburenin closed this Mar 2, 2021

hudi-bot mentioned this pull request Dec 9, 2025

Inject kafka metadata fields #14764

Open


		public static final String INJECT_KAFKA_FIELDS = "hoodie.deltastreamer.source.inject_kafka_fields";

		public static final String KAFKA_PARTITION = "_hudi_kafka_partition";

[HUDI-1648] Added custom kafka meta fields and custom kafka avro decoder. #2598

[HUDI-1648] Added custom kafka meta fields and custom kafka avro decoder. #2598

Uh oh!

Conversation

vburenin commented Feb 23, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the pull request

Brief change log

Verify this pull request

Committer checklist

Uh oh!

codecov-io commented Feb 24, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

vinothchandar commented Feb 24, 2021

Uh oh!

vinothchandar commented Feb 24, 2021

Uh oh!

vburenin commented Feb 24, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

nsivabalan Feb 26, 2021 • 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!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

vburenin commented Mar 2, 2021

Uh oh!

ouzhang commented Apr 21, 2021

Uh oh!

vburenin commented Apr 21, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

vburenin commented Feb 23, 2021 •

edited

Loading

codecov-io commented Feb 24, 2021 •

edited

Loading

nsivabalan Feb 26, 2021 •

edited

Loading