Add Error Handling to Kafka IO #29546

johnjcasey · 2023-11-28T16:39:52Z

Add .withBadRecordHandler() functionality to both KafkaIO.read() and .write()

This also adds usage to the Kafka streaming example

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

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.

# Conflicts: # CHANGES.md

Create gradle task and github actions config for GCS using this.

…tests Feature/automate performance tests

…tests fix call to gradle

…tests run on hosted runner for testing

…tests add additional checkout

…tests add destination for triggered tests

…tests move env variables to correct location

…tests try uploading against separate dataset

…tests try without a user

…tests update branch checkout, try to view the failure log

…tests run on failure

…tests update to use correct BigQuery instance

…tests convert to matrix

update error handler to be serializable to support using it as a member of an auto-value based PTransform

…er-queue-core # Conflicts: # CHANGES.md

…-dlq

…d-dlq # Conflicts: # sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/errorhandling/ErrorHandler.java

github-actions · 2023-12-05T23:34:44Z

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

shunping · 2023-12-06T01:46:12Z

sdks/java/io/kafka/kafka-01103/build.gradle

@@ -18,6 +18,7 @@
 project.ext {
    delimited="0.11.0.3"
    undelimited="01103"
+    sdfCompatable=false


is "Compatable" a typo here?

shunping · 2023-12-06T01:53:49Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java

@@ -739,6 +754,9 @@ abstract Builder<K, V> setValueDeserializerProvider(

      abstract Builder<K, V> setCheckStopReadingFn(@Nullable CheckStopReadingFn checkStopReadingFn);

+      abstract Builder<K, V> setBadRecordErrorHandler(
+          @Nullable ErrorHandler<BadRecord, ?> badRecordErrorHandler);


When will we ever put null in this parameter?

We will never set it to be null. This is @nullable to indicate that the property may not be set, and that is valid

shunping · 2023-12-06T02:00:19Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java

+          } catch (SerializationException e) {
+            // This exception should only occur during the key and value deserialization when
+            // creating the Kafka Record
+            badRecordRouter.route(


QQ: do we have timestamp in DLQ?

shunping

Just a few comments and questions. Overall LGTM.

johnjcasey · 2023-12-06T17:03:49Z

GCP IO Direct test is a flake, whitespace failure is due to files this PR doesn't touch

* Update 2.50 release notes to include new Kafka topicPattern feature * Create groovy class for io performance tests Create gradle task and github actions config for GCS using this. * delete unnecessary class * fix env call * fix call to gradle * run on hosted runner for testing * add additional checkout * add destination for triggered tests * move env variables to correct location * try uploading against separate dataset * try without a user * update branch checkout, try to view the failure log * run on failure * update to use correct BigQuery instance * convert to matrix * add result reporting * add failure clause * remove failure clause, update to run on self-hosted * address comments, clean up build * clarify branching * Add error handling base implementation & test DLQ enabled class * Add test cases * apply spotless * Fix Checkstyles * Fix Checkstyles * make DLH serializable * rename dead letter to bad record * make DLH serializable * Change bad record router name, and use multioutputreceiver instead of process context * Refactor BadRecord to be nested * clean up checkstyle * Update error handler test * Add metrics for counting error records, and for measuring feature usage * apply spotless * fix checkstyle * make metric reporting static * spotless * Rework annotations to be an explicit label on a PTransform, instead of using java annotations * fix checkstyle * Address comments * Address comments * Fix test cases, spotless * remove flatting without error collections * fix nullness * spotless + encoding issues * spotless * throw error when error handler isn't used * add concrete bad record error handler class * spotless, fix test category * fix checkstyle * clean up comments * fix test case * initial wiring of error handler into KafkaIO Read * remove "failing transform" field on bad record, add note to CHANGES.md * fix failing test cases * fix failing test cases * apply spotless * Add tests * Add tests * fix test case * add documentation * wire error handler into kafka write * fix failing test case * Add tests for writing to kafka with exception handling * fix sdf testing * fix sdf testing * spotless * deflake tests * add error handling to kafka streaming example update error handler to be serializable to support using it as a member of an auto-value based PTransform * apply final comments * apply final comments * apply final comments * add line to CHANGES.md * fix spotless * fix checkstyle * make sink transform static for serialization * spotless * fix typo * fix typo * fix spotbugs

johnjcasey and others added 30 commits July 5, 2023 16:05

Update 2.50 release notes to include new Kafka topicPattern feature

c22b8c1

Merge remote-tracking branch 'origin/master'

f7cf5de

# Conflicts: # CHANGES.md

Merge remote-tracking branch 'origin/master'

ab68ecb

Merge remote-tracking branch 'origin/master'

40ad1d5

Create groovy class for io performance tests

6c9c28d

Create gradle task and github actions config for GCS using this.

delete unnecessary class

520c9d1

fix env call

062de23

Merge pull request #181 from johnjcasey/feature/automate-performance-…

01fc25b

…tests Feature/automate performance tests

fix call to gradle

9c9f86b

Merge pull request #182 from johnjcasey/feature/automate-performance-…

92306fa

…tests fix call to gradle

run on hosted runner for testing

925ce55

Merge pull request #183 from johnjcasey/feature/automate-performance-…

2dcfb70

…tests run on hosted runner for testing

add additional checkout

117ef8b

Merge pull request #184 from johnjcasey/feature/automate-performance-…

1f73cda

…tests add additional checkout

add destination for triggered tests

cb6e01b

Merge pull request #185 from johnjcasey/feature/automate-performance-…

a9e86aa

…tests add destination for triggered tests

move env variables to correct location

8ea6c51

Merge pull request #186 from johnjcasey/feature/automate-performance-…

d8822d7

…tests move env variables to correct location

try uploading against separate dataset

320a4cc

Merge pull request #187 from johnjcasey/feature/automate-performance-…

e89b59e

…tests try uploading against separate dataset

try without a user

1cd4e55

Merge pull request #188 from johnjcasey/feature/automate-performance-…

4473f17

…tests try without a user

update branch checkout, try to view the failure log

4fc5b8e

Merge pull request #189 from johnjcasey/feature/automate-performance-…

706650d

…tests update branch checkout, try to view the failure log

run on failure

59069f2

Merge pull request #190 from johnjcasey/feature/automate-performance-…

7f79b62

…tests run on failure

update to use correct BigQuery instance

6f51976

Merge pull request #191 from johnjcasey/feature/automate-performance-…

e95d920

…tests update to use correct BigQuery instance

convert to matrix

df716cb

Merge pull request #192 from johnjcasey/feature/automate-performance-…

4d8eded

…tests convert to matrix

johnjcasey changed the title ~~Feature/kafka read dlq~~ Add Error Handling to Kafka IO Nov 30, 2023

add error handling to kafka streaming example

edd725d

update error handler to be serializable to support using it as a member of an auto-value based PTransform

github-actions bot added the examples label Dec 1, 2023

johnjcasey added 11 commits December 4, 2023 09:54

Merge remote-tracking branch 'upstream/master' into feature/dead-lett…

5067bd1

…er-queue-core # Conflicts: # CHANGES.md

apply final comments

864c429

apply final comments

8307b63

apply final comments

fad9d56

Merge branch 'feature/dead-letter-queue-core' into feature/kafka-read…

8b02c86

…-dlq

add line to CHANGES.md

ad37dda

fix spotless

70437c1

fix checkstyle

6a2801f

make sink transform static for serialization

0c1ab35

Merge remote-tracking branch 'upstream/master' into feature/kafka-rea…

a0781c4

…d-dlq # Conflicts: # sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/errorhandling/ErrorHandler.java

spotless

c589fd2

johnjcasey marked this pull request as ready for review December 5, 2023 20:20

johnjcasey requested a review from Abacn December 5, 2023 20:20

github-actions bot removed runners core labels Dec 5, 2023

shunping reviewed Dec 6, 2023

View reviewed changes

johnjcasey added 2 commits December 6, 2023 09:51

fix typo

4b1a671

fix typo

edb419b

fix spotbugs

58f6ca2

johnjcasey merged commit 5d11c20 into apache:master Dec 7, 2023
32 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add Error Handling to Kafka IO #29546

Add Error Handling to Kafka IO #29546

johnjcasey commented Nov 28, 2023 •

edited

Loading

github-actions bot commented Dec 5, 2023

shunping Dec 6, 2023

shunping Dec 6, 2023

johnjcasey Dec 6, 2023

shunping Dec 6, 2023

johnjcasey Dec 6, 2023

shunping left a comment

johnjcasey commented Dec 6, 2023

Add Error Handling to Kafka IO #29546

Add Error Handling to Kafka IO #29546

Conversation

johnjcasey commented Nov 28, 2023 • edited Loading

This also adds usage to the Kafka streaming example

GitHub Actions Tests Status (on master branch)

github-actions bot commented Dec 5, 2023

shunping Dec 6, 2023

Choose a reason for hiding this comment

shunping Dec 6, 2023

Choose a reason for hiding this comment

johnjcasey Dec 6, 2023

Choose a reason for hiding this comment

shunping Dec 6, 2023

Choose a reason for hiding this comment

johnjcasey Dec 6, 2023

Choose a reason for hiding this comment

shunping left a comment

Choose a reason for hiding this comment

johnjcasey commented Dec 6, 2023

johnjcasey commented Nov 28, 2023 •

edited

Loading