Offset-based deduplication for Kafka source.

[tomstepp]

Offset-based deduplication for Kafka source.

---

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

[github-actions]

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

[tomstepp]

R: @scwhittle

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

[tomstepp]

R: @chamikaramj and @johnjcasey

Can one of you help review source/translation changes here? I am working on a change that applies to Kafka Read on Java legacy. I am having trouble understanding why the change breaks KafkaIOExternalTest.testConstructKafkaRead and KafkaIOExternalTest.testConstructKafkaReadWithoutMetadata http://screen/3taBLZo28VjePkZ.

[tomstepp]

@scwhittle if you have any tips, or know who we can ask:

WARNING: Configuration class 'org.apache.beam.sdk.io.kafka.KafkaIO$Read$External$Configuration' has no schema registered. Attempting to construct with setter approach.

Expansion result: error: "java.lang.RuntimeException: Failed to build transform from spec urn: 

\"beam:transform:org.apache.beam:kafka_read_with_metadata:v1\"\npayload: \"\\n\\212\\003\\n\\020\\n\\006topics\\032\\006\\032\\004\\n\\002\\020\\a\\n!\\n\\017consumer_config\\032\\n*\\b\\n\\002\\020\\a\\022\\002\\020\\a \\001(\\001\\n\\032\\n\\020key_deserializer\\032\\002\\020\\a \\002(\\002\\n\\034\\n\\022value_deserializer\\032\\002\\020\\a \\003(\\003\\n\\031\\n\\017start_read_time\\032\\002\\020\\004 \\004(\\004\\n#\\n\\031commit_offset_in_finalize\\032\\002\\020\\b \\005(\\005\\n\\032\\n\\020timestamp_policy\\032\\002\\020\\a \\006(\\006\\n\\\"\\n\\030consumer_polling_timeout\\032\\002\\020\\004 \\a(\\a\\n\\037\\n\\025redistribute_num_keys\\032\\002\\020\\003 \\b(\\b\\n\\026\\n\\fredistribute\\032\\002\\020\\b \\t(\\t\\n\\032\\n\\020allow_duplicates\\032\\002\\020\\b \\n(\\n\\n\\036\\n\\024offset_deduplication\\032\\002\\020\\b \\v(\\v\\022$059ad81c-549f-4ab3-adb7-37134f22938e\\022\\372\\002\\f\\000\\000\\000\\000\\002\\006

...

worker.org.gradle.process.internal.worker.GradleWorkerMain.main(GradleWorkerMain.java:74)\nCaused by: 

java.lang.IllegalStateException: Missing required properties: offsetDeduplication\n\tat org.apache.beam.sdk.io.kafka.AutoValue_KafkaIO_Read$Builder.build(AutoValue_KafkaIO_Read.java:627) org.apache.beam.sdk.io.kafka.KafkaIO$RowsWithMetadata$Builder.buildExternal(KafkaIO.java:2139)
org.apache.beam.sdk.io.kafka.KafkaIO$RowsWithMetadata$Builder.buildExternal(KafkaIO.java:2113) org.apache.beam.sdk.expansion.service.ExpansionService$TransformProviderForBuilder.getTransform(ExpansionService.java:335)
... 47 more"

Expected: a collection containing a string matching the pattern '.*KafkaIO-Read.*'
     but: was empty

[chamikaramj]

Probably try changing the property definition to

public abstract @Nullable Boolean isOffsetDeduplication();

[chamikaramj]

Seems like the builder is considering this to be a required parameter and failing since it's not provided through the test schema here:

beam/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOExternalTest.java

Line 102 in ad006a8

Schema.of(

[tomstepp]

Probably try changing the property definition

Done, I think this resolves it.

Seems like the builder is considering this to be a required parameter and failing since it's not provided through the test schema

I think this PR had it added already, done.

Thanks for the feedback!

[tomstepp]

Run Java_Kafka_IO_Direct PreCommit

[tomstepp]

Ok, I think some minor updates to KafkaIO translation allows to pass pre commit (such as no conflicting unbounded vs SDF settings). It was a combination of Cham's suggestion to make nullable, and then not setting a default values.

[chamikaramj]

I think you need to set a default value for this change to be compatible. May be it just passed randomly due to the retry ?

[chamikaramj]

Lemme know if you need help with re-triggering a test suite.

[tomstepp]

I think you need to set a default value for this change to be compatible. May be it just passed randomly due to the retry ?

When setting a default value and marking offset dedup as only compatible with legacy, Kafka read jobs with SDF-only arguments will fail saying there is no possible implementation (legacy/unbounded or SDF).

Example from testReadTransformCreationWithSdfImplementationBoundProperty:

There is no Kafka read implementation that supports every configured property! Not supported implementations with the associated properties: {SDF=[OFFSET_DEDUPLICATION], LEGACY=[STOP_READ_TIME]}

@chamikaramj I don't think we need to specify a non-null default. Am I misunderstanding something?

[tomstepp]

Synced offline with Cham. I think we are on the same page. Made a minor change KafkaIOTranslation.

[chamikaramj]

Thanks. LGTM.

Just some nits.