KAFKA-16448: Handle fatal user exception during processing error by sebastienviale · Pull Request #16675 · apache/kafka

sebastienviale · 2024-07-24T07:40:55Z

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 catch the exceptions thrown while handling a processing exception

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

Contributors
@Dabz
@sebastienviale
@loicgreffier

cadonna

Thanks for the PR, @sebastienviale !

Here my comments!

cadonna · 2024-07-24T08:22:56Z

+            } catch (final Exception fatalUserException) {
+                log.error(
+                    "Processing error callback failed after processing error for record {}",
+                    record,


We should not log records for privacy reasons. The exception should be enough.

cadonna · 2024-07-24T08:23:14Z

+                response = processingExceptionHandler
+                        .handle(errorHandlerContext, record, e);


nit:

Suggested change

response = processingExceptionHandler

.handle(errorHandlerContext, record, e);

response = processingExceptionHandler.handle(errorHandlerContext, record, e);

cadonna · 2024-07-24T08:44:11Z

+            try {
+                response = processingExceptionHandler
+                        .handle(errorHandlerContext, record, e);
+            } catch (final Exception fatalUserException) {


Could you please also add a unit test for this?

cadonna

Thanks for the updates!

LGTM!

mjsax · 2024-07-25T23:22:00Z

+            try {
+                response = processingExceptionHandler.handle(errorHandlerContext, record, e);
+            } catch (final Exception fatalUserException) {
+                throw new StreamsException("Fatal user code error in processing error callback", fatalUserException);


Should we add an error log before throwing?

Btw: it seems we have such a guard only for the DeserializationExceptionHandler#handle(...) call, but not for the ProductionExceptionHanlder#call(...)? Might be good to do a follow up PR to add it to the proction-error-handler case, too?

One more thing: if we throw StreamsException here, we might catch it in an upstream process(), right? So should we rather wrap this StreamsException with a FailedProcessingException?

Do we actually have a test for "passing through" of FailedProcessingException (ie, a test which checks if upstream ProcessingExceptionHandler are not called any longer, if a downstream processor had an error and the corresponding handler did return FAIL? If yes, we could re-use this test to also test a crashing handler. If not, let's add new tests for both cases :)

done:
I changed the StreamException into a FailedProcessingException
and added some test to check that no more processor is executed when an exception occurs or when an exception occurs in a handler.

Should we add an error log before throwing?

Btw: it seems we have such a guard only for the DeserializationExceptionHandler#handle(...) call, but not for the ProductionExceptionHanlder#call(...)? Might be good to do a follow up PR to add it to the proction-error-handler case, too?

@mjsax is it mandatory for this KIP ?

One more thing: if we throw StreamsException here, we might catch it in an upstream process(), right? So should we rather wrap this StreamsException with a FailedProcessingException?

@mjsax Correct. FailedProcessingException is thrown as is:

kafka/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java

Line 205 in 683b8a2

throw e;

Do we actually have a test for "passing through" of FailedProcessingException

We do:

kafka/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java

Line 130 in 683b8a2

public void shouldNotHandleInternalExceptionsThrownDuringProcessing(final String ignoredExceptionName,

It asserts FailedProcessingException is not handled and thrown as is. Another test has been added to ensure that the cause of the exception is from the processing exception handler (if it is actually thrown by it).

Btw: it seems we have such a guard only for the DeserializationExceptionHandler#handle(...) call, but not for the ProductionExceptionHanlder#call(...)? Might be good to do a follow up PR to add it to the proction-error-handler case, too?

@mjsax is it mandatory for this KIP ?

Well, we don't want a handler which throws an exception to not crash KS, because the exception is caught in an upstream handler, right? But should be a simple thing to add?

But I see your point that it's not really related to the KIP. I can do a follow up myself.

Btw: it seems we have such a guard only for the DeserializationExceptionHandler#handle(...) call, but not for the ProductionExceptionHanlder#call(...)? Might be good to do a follow up PR to add it to the proction-error-handler case, too?

@mjsax is it mandatory for this KIP ?

Well, we don't want a handler which throws an exception to not crash KS, because the exception is caught in an upstream handler, right? But should be a simple thing to add?

But I see your point that it's not really related to the KIP. I can do a follow up myself.

Ok, I will open a new PR for that !

mjsax · 2024-07-30T03:34:51Z

@sebastienviale -- The PR has merge conflicts and needs to be rebased.

Maybe also address the open comments from the other PRs on this one.

Co-authored-by: Dabz <d.gasparina@gmail.com> Co-authored-by: loicgreffier <loic.greffier@michelin.com> KAFKA-16448 Handle fatal user exception during processing error Co-authored-by: Dabz <d.gasparina@gmail.com> Co-authored-by: loicgreffier <loic.greffier@michelin.com> KAFKA-16448 Handle fatal user exception during processing error Co-authored-by: Dabz <d.gasparina@gmail.com> Co-authored-by: loicgreffier <loic.greffier@michelin.com> KAFKA-16448 Handle fatal user exception during processing error Co-authored-by: Dabz <d.gasparina@gmail.com> Co-authored-by: loicgreffier <loic.greffier@michelin.com> KAFKA-16448 Handle fatal user exception during processing error Co-authored-by: Dabz <d.gasparina@gmail.com> Co-authored-by: loicgreffier <loic.greffier@michelin.com>

Co-authored-by: Dabz <d.gasparina@gmail.com> Co-authored-by: loicgreffier <loic.greffier@michelin.com>

) This PR is part of KIP-1033 which aims to bring a ProcessingExceptionHandler to Kafka Streams in order to deal with exceptions that occur during processing. This PR catch the exceptions thrown while handling a processing exception Co-authored-by: Dabz <d.gasparina@gmail.com> Co-authored-by: loicgreffier <loic.greffier@michelin.com> Reviewers: Bruno Cadonna <bruno@confluent.io>, Matthias J. Sax <matthias@confluent.io>

mjsax · 2024-07-31T06:07:38Z

Thanks for the PR. Merged to trunk and cherry-picked to 3.9 branch.

sebastienviale mentioned this pull request Jul 24, 2024

KAFKA-16448: Add ErrorHandlerContext in deserialization exception handler #16432

Merged

cadonna reviewed Jul 24, 2024

View reviewed changes

cadonna added streams kip Requires or implements a KIP labels Jul 24, 2024

cadonna reviewed Jul 24, 2024

View reviewed changes

cadonna approved these changes Jul 24, 2024

View reviewed changes

mjsax reviewed Jul 25, 2024

View reviewed changes

sebastienviale force-pushed the KAFKA-16448-Handle-Fatal-User-Exception-During-Processing-Error branch from 77d2ed2 to 4d50ab3 Compare July 29, 2024 14:16

sebastienviale force-pushed the KAFKA-16448-Handle-Fatal-User-Exception-During-Processing-Error branch from 9661027 to 9c6f40e Compare July 30, 2024 06:48

sebastienviale force-pushed the KAFKA-16448-Handle-Fatal-User-Exception-During-Processing-Error branch from 9c6f40e to 2be7af8 Compare July 30, 2024 06:50

sebastienviale and others added 4 commits July 30, 2024 15:33

KAFKA-16448 Handle fatal user exception during processing error

b93561b

Co-authored-by: Dabz <d.gasparina@gmail.com> Co-authored-by: loicgreffier <loic.greffier@michelin.com>

KAFKA-16448: Fix processing exception handler throwing exception itself

92235e5

Co-authored-by: Dabz <d.gasparina@gmail.com> Co-authored-by: loicgreffier <loic.greffier@michelin.com>

KAFKA-16448: Fix spaces

458ea81

KAFKA-16448: Fix spaces

9587069

mjsax merged commit 0dc9b9e into apache:trunk Jul 31, 2024

		response = processingExceptionHandler
		.handle(errorHandlerContext, record, e);

	response = processingExceptionHandler
	.handle(errorHandlerContext, record, e);
	response = processingExceptionHandler.handle(errorHandlerContext, record, e);

Conversation

sebastienviale commented Jul 24, 2024

Uh oh!

cadonna 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!

cadonna left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax Jul 30, 2024 • 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!

sebastienviale Jul 30, 2024 • 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!

loicgreffier Jul 30, 2024 • 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!

mjsax Jul 31, 2024 • 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!

mjsax commented Jul 30, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mjsax commented Jul 31, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

mjsax Jul 30, 2024 •

edited

Loading

sebastienviale Jul 30, 2024 •

edited

Loading

loicgreffier Jul 30, 2024 •

edited

Loading

mjsax Jul 31, 2024 •

edited

Loading

mjsax commented Jul 30, 2024 •

edited

Loading