KAFKA-16448: Add ProcessingExceptionHandler in Streams configuration #16092

loicgreffier · 2024-05-27T15:11:32Z

This PR is part of KAFKA-16448 which aims to bring a ProcessingExceptionHandler to Kafka Streams in order to deal with exceptions that occur during processing.

This PR brings ProcessingExceptionHandler in Streams configuration.

Jira: https://issues.apache.org/jira/browse/KAFKA-16448.

Contributors

@Dabz
@sebastienviale
@loicgreffier

Depends On

#16090

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

…ions Co-authored-by: Dabz <[email protected]> Co-authored-by: sebastienviale <[email protected]>

Co-authored-by: Dabz <[email protected]> Co-authored-by: sebastienviale <[email protected]>

…ions Co-authored-by: Dabz <[email protected]> Co-authored-by: sebastienviale <[email protected]>

Co-authored-by: Dabz <[email protected]> Co-authored-by: sebastienviale <[email protected]>

cadonna

@loicgreffier Thanks for the PR!

Here my comment.

Since today is feature freeze, it would be great if you could make the changes as soon as possible.

cadonna · 2024-05-29T07:45:44Z

streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java

+            .define(PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG,
+                    Type.CLASS,
+                    LogAndFailProcessingExceptionHandler.class.getName(),
+                    Importance.MEDIUM,
+                    PROCESSING_EXCEPTION_HANDLER_CLASS_DOC)


Could you please add unit tests in StreamsConfigTest?
I think you need the following unit tests:

One where config is not set to verify the default value

One where the config is set and you get the correct instance from processingExceptionHandler().

hi,
It is done !

…ndler Co-authored-by: Dabz <[email protected]> Co-authored-by: loicgreffier <[email protected]>

…fig' of https://github.com/loicgreffier/kafka into KAFKA-16448-Add-Processing-Exception-Handler-StreamsConfig

…fig' of https://github.com/loicgreffier/kafka into KAFKA-16448-Add-Processing-Exception-Handler-StreamsConfig Co-authored-by: Dabz <[email protected]> Co-authored-by: loicgreffier <[email protected]>

…fig' of https://github.com/loicgreffier/kafka into KAFKA-16448-Add-Processing-Exception-Handler-StreamsConfig

…eptionHandler Co-authored-by: Dabz <[email protected]> Co-authored-by: loicgreffier <[email protected]>

cadonna

Thanks for the updates, @loicgreffier and @sebastienviale!

After addressing my comment, I will approve and merge this PR.

cadonna · 2024-05-29T10:40:40Z

streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java

+    @Test
+    public void testInvalidProcessingExceptionHandler() {
+        props.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG, "org.apache.kafka.streams.errors.InvalidProcessingExceptionHandler");
+        assertThrows(ConfigException.class, () -> new StreamsConfig(props));


Could you please also verify the message?

Suggested change

assertThrows(ConfigException.class, () -> new StreamsConfig(props));

final Exception exception = assertThrows(ConfigException.class, () -> new StreamsConfig(props));

assertThat(

exception.getMessage(),

containsString("<the message to verify>")

);

Verification has been added

…ptionHandler Co-authored-by: Dabz <[email protected]> Co-authored-by: loicgreffier <[email protected]>

cadonna

Thanks @loicgreffier @sebastienviale !

LGTM!

Co-authored-by: Dabz <[email protected]> Co-authored-by: sebastienviale <[email protected]>

loicgreffier · 2024-05-29T17:33:22Z

@cadonna I've rebased this PR so file changes are now clear. Do we need to squash commits manually or is it automatically done when merging ?

…fig' of https://github.com/loicgreffier/kafka into KAFKA-16448-Add-Processing-Exception-Handler-StreamsConfig

Co-authored-by: Dabz <[email protected]> Co-authored-by: loicgreffier <[email protected]>

cadonna · 2024-05-30T07:12:01Z

@cadonna I've rebased this PR so file changes are now clear. Do we need to squash commits manually or is it automatically done when merging ?

Thanks! Squashing happens when merging. Don't worry!

…uration (apache#16092)" This reverts commit 3f70c46.

…uration (#16092)" (#16141) This reverts commit 3f70c46. Reviewer: Lucas Brutschy <[email protected]>

…pache#16092) This PR is part of KAFKA-16448 which aims to bring a ProcessingExceptionHandler to Kafka Streams in order to deal with exceptions that occur during processing. This PR brings ProcessingExceptionHandler in Streams configuration. Co-authored-by: Dabz <[email protected]> Co-authored-by: sebastienviale <[email protected]> Reviewer: Bruno Cadonna <[email protected]>

…uration (apache#16092)" (apache#16141) This reverts commit 3f70c46. Reviewer: Lucas Brutschy <[email protected]>

…pache#16092) This PR is part of KAFKA-16448 which aims to bring a ProcessingExceptionHandler to Kafka Streams in order to deal with exceptions that occur during processing. This PR brings ProcessingExceptionHandler in Streams configuration. Co-authored-by: Dabz <[email protected]> Co-authored-by: sebastienviale <[email protected]> Reviewer: Bruno Cadonna <[email protected]>

…uration (apache#16092)" (apache#16141) This reverts commit 3f70c46. Reviewer: Lucas Brutschy <[email protected]>

…pache#16092) This PR is part of KAFKA-16448 which aims to bring a ProcessingExceptionHandler to Kafka Streams in order to deal with exceptions that occur during processing. This PR brings ProcessingExceptionHandler in Streams configuration. Co-authored-by: Dabz <[email protected]> Co-authored-by: sebastienviale <[email protected]> Reviewer: Bruno Cadonna <[email protected]>

…uration (apache#16092)" (apache#16141) This reverts commit 3f70c46. Reviewer: Lucas Brutschy <[email protected]>

…pache#16092) This PR is part of KAFKA-16448 which aims to bring a ProcessingExceptionHandler to Kafka Streams in order to deal with exceptions that occur during processing. This PR brings ProcessingExceptionHandler in Streams configuration. Co-authored-by: Dabz <[email protected]> Co-authored-by: sebastienviale <[email protected]> Reviewer: Bruno Cadonna <[email protected]>

…uration (apache#16092)" (apache#16141) This reverts commit 3f70c46. Reviewer: Lucas Brutschy <[email protected]>

loicgreffier and others added 2 commits May 27, 2024 17:06

KAFKA-16448: Add ProcessingExceptionHandler interface and implementat…

9cfac39

…ions Co-authored-by: Dabz <[email protected]> Co-authored-by: sebastienviale <[email protected]>

KAFKA-16448: Add ProcessingExceptionHandler in Streams configuration

b957861

Co-authored-by: Dabz <[email protected]> Co-authored-by: sebastienviale <[email protected]>

loicgreffier mentioned this pull request May 27, 2024

KAFKA-16448: Catch and handle processing exceptions #16093

Merged

3 tasks

KAFKA-16448: Add ProcessingExceptionHandler interface and implementat…

9eef86d

…ions Co-authored-by: Dabz <[email protected]> Co-authored-by: sebastienviale <[email protected]>

loicgreffier mentioned this pull request May 28, 2024

KAFKA-16448 [WIP]: Add Kafka Streams exception handler for exceptions occuring during processing #15973

Closed

3 tasks

loicgreffier and others added 3 commits May 28, 2024 19:25

KAFKA-16448: Update from parent branch

40fdac5

Co-authored-by: Dabz <[email protected]> Co-authored-by: sebastienviale <[email protected]>

KAFKA-16448: Fix indentation

3cb797b

Co-authored-by: Dabz <[email protected]> Co-authored-by: sebastienviale <[email protected]>

KAFKA-16448: Remove unecessary classes

c1c0d5c

Co-authored-by: Dabz <[email protected]> Co-authored-by: sebastienviale <[email protected]>

gharris1727 added streams kip Requires or implements a KIP labels May 28, 2024

cadonna reviewed May 29, 2024

View reviewed changes

sebastienviale and others added 6 commits May 29, 2024 10:15

KAFKA-16448: add Streams configuration test for ProcessingExceptionHa…

bea1fb8

…ndler Co-authored-by: Dabz <[email protected]> Co-authored-by: loicgreffier <[email protected]>

KAFKA-16448: add Streams configuration test for ProcessingExceptionHa…

b7058d2

…ndler Co-authored-by: Dabz <[email protected]> Co-authored-by: loicgreffier <[email protected]>

Merge branch 'KAFKA-16448-Add-Processing-Exception-Handler-StreamsCon…

0a54bae

…fig' of https://github.com/loicgreffier/kafka into KAFKA-16448-Add-Processing-Exception-Handler-StreamsConfig

Merge branch 'KAFKA-16448-Add-Processing-Exception-Handler-StreamsCon…

014b799

…fig' of https://github.com/loicgreffier/kafka into KAFKA-16448-Add-Processing-Exception-Handler-StreamsConfig Co-authored-by: Dabz <[email protected]> Co-authored-by: loicgreffier <[email protected]>

Merge branch 'KAFKA-16448-Add-Processing-Exception-Handler-StreamsCon…

10e69a9

…fig' of https://github.com/loicgreffier/kafka into KAFKA-16448-Add-Processing-Exception-Handler-StreamsConfig

KAFKA-16448: add Streams configuration test for invalid ProcessingExc…

3c7de0c

…eptionHandler Co-authored-by: Dabz <[email protected]> Co-authored-by: loicgreffier <[email protected]>

cadonna reviewed May 29, 2024

View reviewed changes

KAFKA-16448: add error message test in case of invalid ProcessingExce…

5e7a656

…ptionHandler Co-authored-by: Dabz <[email protected]> Co-authored-by: loicgreffier <[email protected]>

cadonna approved these changes May 29, 2024

View reviewed changes

KAFKA-16448: Add ProcessingExceptionHandler in Streams configuration

60bba3b

Co-authored-by: Dabz <[email protected]> Co-authored-by: sebastienviale <[email protected]>

loicgreffier force-pushed the KAFKA-16448-Add-Processing-Exception-Handler-StreamsConfig branch from 85060e6 to 60bba3b Compare May 29, 2024 17:24

KAFKA-16448: Add ProcessingExceptionHandler in Streams configuration

25115ce

Co-authored-by: Dabz <[email protected]> Co-authored-by: sebastienviale <[email protected]>

sebastienviale and others added 2 commits May 29, 2024 21:49

Merge branch 'KAFKA-16448-Add-Processing-Exception-Handler-StreamsCon…

4b56def

…fig' of https://github.com/loicgreffier/kafka into KAFKA-16448-Add-Processing-Exception-Handler-StreamsConfig

KAFKA-16448: remove unused imports

7d67257

Co-authored-by: Dabz <[email protected]> Co-authored-by: loicgreffier <[email protected]>

cadonna merged commit 3f70c46 into apache:trunk May 30, 2024
1 check failed

cadonna added a commit to cadonna/kafka that referenced this pull request May 30, 2024

Revert "KAFKA-16448: Add ProcessingExceptionHandler in Streams config…

f1e6af9

…uration (apache#16092)" This reverts commit 3f70c46.

cadonna added a commit that referenced this pull request May 30, 2024

Revert "KAFKA-16448: Add ProcessingExceptionHandler in Streams config…

fea3eeb

…uration (#16092)" (#16141) This reverts commit 3f70c46. Reviewer: Lucas Brutschy <[email protected]>

chiacyu pushed a commit to chiacyu/kafka that referenced this pull request Jun 1, 2024

Revert "KAFKA-16448: Add ProcessingExceptionHandler in Streams config…

88bb080

…uration (apache#16092)" (apache#16141) This reverts commit 3f70c46. Reviewer: Lucas Brutschy <[email protected]>

wernerdv pushed a commit to wernerdv/kafka that referenced this pull request Jun 3, 2024

Revert "KAFKA-16448: Add ProcessingExceptionHandler in Streams config…

00ea530

…uration (apache#16092)" (apache#16141) This reverts commit 3f70c46. Reviewer: Lucas Brutschy <[email protected]>

loicgreffier mentioned this pull request Jun 4, 2024

KAFKA-16448: Add ProcessingExceptionHandler in Streams configuration #16188

Merged

3 tasks

TaiJuWu pushed a commit to TaiJuWu/kafka that referenced this pull request Jun 8, 2024

Revert "KAFKA-16448: Add ProcessingExceptionHandler in Streams config…

440db0a

…uration (apache#16092)" (apache#16141) This reverts commit 3f70c46. Reviewer: Lucas Brutschy <[email protected]>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-16448: Add ProcessingExceptionHandler in Streams configuration #16092

KAFKA-16448: Add ProcessingExceptionHandler in Streams configuration #16092

loicgreffier commented May 27, 2024 •

edited

Loading

cadonna left a comment

cadonna May 29, 2024

sebastienviale May 29, 2024

cadonna left a comment

cadonna May 29, 2024

sebastienviale May 29, 2024

cadonna left a comment

loicgreffier commented May 29, 2024

cadonna commented May 30, 2024

-        assertThrows(ConfigException.class, () -> new StreamsConfig(props));
+        final Exception exception = assertThrows(ConfigException.class, () -> new StreamsConfig(props));
+        assertThat(
+            exception.getMessage(),
+            containsString("<the message to verify>")
+        );

KAFKA-16448: Add ProcessingExceptionHandler in Streams configuration #16092

KAFKA-16448: Add ProcessingExceptionHandler in Streams configuration #16092

Conversation

loicgreffier commented May 27, 2024 • edited Loading

Contributors

Depends On

Committer Checklist (excluded from commit message)

cadonna left a comment

Choose a reason for hiding this comment

cadonna May 29, 2024

Choose a reason for hiding this comment

sebastienviale May 29, 2024

Choose a reason for hiding this comment

cadonna left a comment

Choose a reason for hiding this comment

cadonna May 29, 2024

Choose a reason for hiding this comment

sebastienviale May 29, 2024

Choose a reason for hiding this comment

cadonna left a comment

Choose a reason for hiding this comment

loicgreffier commented May 29, 2024

cadonna commented May 30, 2024

loicgreffier commented May 27, 2024 •

edited

Loading