You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "sebastienviale (via GitHub)" <gi...@apache.org> on 2024/07/24 07:40:58 UTC
[PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
sebastienviale opened a new pull request, #16675:
URL: https://github.com/apache/kafka/pull/16675
This PR is part of [KAFKA-16448](https://issues.apache.org/jira/browse/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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #16675:
URL: https://github.com/apache/kafka/pull/16675#discussion_r1697084682
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -214,9 +214,13 @@ public void process(final Record<KIn, VIn> record) {
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());
- final ProcessingExceptionHandler.ProcessingHandlerResponse response = processingExceptionHandler
- .handle(errorHandlerContext, record, e);
+ final ProcessingExceptionHandler.ProcessingHandlerResponse response;
+ try {
+ response = processingExceptionHandler.handle(errorHandlerContext, record, e);
+ } catch (final Exception fatalUserException) {
+ throw new StreamsException("Fatal user code error in processing error callback", fatalUserException);
Review Comment:
> 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 ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #16675:
URL: https://github.com/apache/kafka/pull/16675#discussion_r1698015268
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -214,9 +214,13 @@ public void process(final Record<KIn, VIn> record) {
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());
- final ProcessingExceptionHandler.ProcessingHandlerResponse response = processingExceptionHandler
- .handle(errorHandlerContext, record, e);
+ final ProcessingExceptionHandler.ProcessingHandlerResponse response;
+ try {
+ response = processingExceptionHandler.handle(errorHandlerContext, record, e);
+ } catch (final Exception fatalUserException) {
+ throw new StreamsException("Fatal user code error in processing error callback", fatalUserException);
Review Comment:
> > > 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 !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #16675:
URL: https://github.com/apache/kafka/pull/16675
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
Posted by "loicgreffier (via GitHub)" <gi...@apache.org>.
loicgreffier commented on code in PR #16675:
URL: https://github.com/apache/kafka/pull/16675#discussion_r1697564800
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -214,9 +214,13 @@ public void process(final Record<KIn, VIn> record) {
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());
- final ProcessingExceptionHandler.ProcessingHandlerResponse response = processingExceptionHandler
- .handle(errorHandlerContext, record, e);
+ final ProcessingExceptionHandler.ProcessingHandlerResponse response;
+ try {
+ response = processingExceptionHandler.handle(errorHandlerContext, record, e);
+ } catch (final Exception fatalUserException) {
+ throw new StreamsException("Fatal user code error in processing error callback", fatalUserException);
Review Comment:
> 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`?
Correct. `FailedProcessingException` is thrown as is:
https://github.com/apache/kafka/blob/683b8a2beeb5a75ef0a2ac878cf5a963f9b9d915/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java#L205
> Do we actually have a test for "passing through" of `FailedProcessingException`
We do:
https://github.com/apache/kafka/blob/683b8a2beeb5a75ef0a2ac878cf5a963f9b9d915/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java#L130
It asserts `FailedProcessingException` is not handled and thrown as is. Another test has been added to ensure that the cause of the caught exception is from the processing exception handler.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #16675:
URL: https://github.com/apache/kafka/pull/16675#issuecomment-2259742160
Thanks for the PR. Merged to `trunk` and cherry-picked to `3.9` branch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16675:
URL: https://github.com/apache/kafka/pull/16675#discussion_r1692266139
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -214,9 +214,13 @@ public void process(final Record<KIn, VIn> record) {
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());
- final ProcessingExceptionHandler.ProcessingHandlerResponse response = processingExceptionHandler
- .handle(errorHandlerContext, record, e);
+ final ProcessingExceptionHandler.ProcessingHandlerResponse response;
+ try {
+ response = processingExceptionHandler.handle(errorHandlerContext, record, e);
+ } catch (final Exception fatalUserException) {
+ throw new StreamsException("Fatal user code error in processing error callback", fatalUserException);
Review Comment:
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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #16675:
URL: https://github.com/apache/kafka/pull/16675#issuecomment-2257391601
@sebastienviale -- The PR has merge conflicts and needs to be rebased.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #16675:
URL: https://github.com/apache/kafka/pull/16675#discussion_r1696979756
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -214,9 +214,13 @@ public void process(final Record<KIn, VIn> record) {
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());
- final ProcessingExceptionHandler.ProcessingHandlerResponse response = processingExceptionHandler
- .handle(errorHandlerContext, record, e);
+ final ProcessingExceptionHandler.ProcessingHandlerResponse response;
+ try {
+ response = processingExceptionHandler.handle(errorHandlerContext, record, e);
+ } catch (final Exception fatalUserException) {
+ throw new StreamsException("Fatal user code error in processing error callback", fatalUserException);
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16675:
URL: https://github.com/apache/kafka/pull/16675#discussion_r1697940240
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -214,9 +214,13 @@ public void process(final Record<KIn, VIn> record) {
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());
- final ProcessingExceptionHandler.ProcessingHandlerResponse response = processingExceptionHandler
- .handle(errorHandlerContext, record, e);
+ final ProcessingExceptionHandler.ProcessingHandlerResponse response;
+ try {
+ response = processingExceptionHandler.handle(errorHandlerContext, record, e);
+ } catch (final Exception fatalUserException) {
+ throw new StreamsException("Fatal user code error in processing error callback", fatalUserException);
Review Comment:
>> 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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #16675:
URL: https://github.com/apache/kafka/pull/16675#discussion_r1689473987
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -214,9 +214,18 @@ public void process(final Record<KIn, VIn> record) {
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());
- final ProcessingExceptionHandler.ProcessingHandlerResponse response = processingExceptionHandler
- .handle(errorHandlerContext, record, e);
-
+ final ProcessingExceptionHandler.ProcessingHandlerResponse response;
+
+ try {
+ response = processingExceptionHandler
+ .handle(errorHandlerContext, record, e);
+ } catch (final Exception fatalUserException) {
+ log.error(
+ "Processing error callback failed after processing error for record {}",
+ record,
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
Posted by "loicgreffier (via GitHub)" <gi...@apache.org>.
loicgreffier commented on code in PR #16675:
URL: https://github.com/apache/kafka/pull/16675#discussion_r1697564800
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -214,9 +214,13 @@ public void process(final Record<KIn, VIn> record) {
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());
- final ProcessingExceptionHandler.ProcessingHandlerResponse response = processingExceptionHandler
- .handle(errorHandlerContext, record, e);
+ final ProcessingExceptionHandler.ProcessingHandlerResponse response;
+ try {
+ response = processingExceptionHandler.handle(errorHandlerContext, record, e);
+ } catch (final Exception fatalUserException) {
+ throw new StreamsException("Fatal user code error in processing error callback", fatalUserException);
Review Comment:
> 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:
https://github.com/apache/kafka/blob/683b8a2beeb5a75ef0a2ac878cf5a963f9b9d915/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java#L205
> Do we actually have a test for "passing through" of `FailedProcessingException`
We do:
https://github.com/apache/kafka/blob/683b8a2beeb5a75ef0a2ac878cf5a963f9b9d915/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java#L130
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).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
Posted by "loicgreffier (via GitHub)" <gi...@apache.org>.
loicgreffier commented on code in PR #16675:
URL: https://github.com/apache/kafka/pull/16675#discussion_r1697564800
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -214,9 +214,13 @@ public void process(final Record<KIn, VIn> record) {
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());
- final ProcessingExceptionHandler.ProcessingHandlerResponse response = processingExceptionHandler
- .handle(errorHandlerContext, record, e);
+ final ProcessingExceptionHandler.ProcessingHandlerResponse response;
+ try {
+ response = processingExceptionHandler.handle(errorHandlerContext, record, e);
+ } catch (final Exception fatalUserException) {
+ throw new StreamsException("Fatal user code error in processing error callback", fatalUserException);
Review Comment:
> 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:
https://github.com/apache/kafka/blob/683b8a2beeb5a75ef0a2ac878cf5a963f9b9d915/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java#L205
> Do we actually have a test for "passing through" of `FailedProcessingException`
We do:
https://github.com/apache/kafka/blob/683b8a2beeb5a75ef0a2ac878cf5a963f9b9d915/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java#L130
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)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #16675:
URL: https://github.com/apache/kafka/pull/16675#discussion_r1689474509
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -214,9 +214,18 @@ public void process(final Record<KIn, VIn> record) {
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());
- final ProcessingExceptionHandler.ProcessingHandlerResponse response = processingExceptionHandler
- .handle(errorHandlerContext, record, e);
-
+ final ProcessingExceptionHandler.ProcessingHandlerResponse response;
+
+ try {
+ response = processingExceptionHandler
+ .handle(errorHandlerContext, record, e);
+ } catch (final Exception fatalUserException) {
Review Comment:
done
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -214,9 +214,18 @@ public void process(final Record<KIn, VIn> record) {
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());
- final ProcessingExceptionHandler.ProcessingHandlerResponse response = processingExceptionHandler
- .handle(errorHandlerContext, record, e);
-
+ final ProcessingExceptionHandler.ProcessingHandlerResponse response;
+
+ try {
+ response = processingExceptionHandler
+ .handle(errorHandlerContext, record, e);
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16675:
URL: https://github.com/apache/kafka/pull/16675#discussion_r1696289377
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -214,9 +214,13 @@ public void process(final Record<KIn, VIn> record) {
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());
- final ProcessingExceptionHandler.ProcessingHandlerResponse response = processingExceptionHandler
- .handle(errorHandlerContext, record, e);
+ final ProcessingExceptionHandler.ProcessingHandlerResponse response;
+ try {
+ response = processingExceptionHandler.handle(errorHandlerContext, record, e);
+ } catch (final Exception fatalUserException) {
+ throw new StreamsException("Fatal user code error in processing error callback", fatalUserException);
Review Comment:
One more thing: if we throw `StreamsException` here, we might catch it in an upstream `process()`, right? So should we rather wrap it with a `FailedProcessingException`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16675:
URL: https://github.com/apache/kafka/pull/16675#discussion_r1697940240
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -214,9 +214,13 @@ public void process(final Record<KIn, VIn> record) {
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());
- final ProcessingExceptionHandler.ProcessingHandlerResponse response = processingExceptionHandler
- .handle(errorHandlerContext, record, e);
+ final ProcessingExceptionHandler.ProcessingHandlerResponse response;
+ try {
+ response = processingExceptionHandler.handle(errorHandlerContext, record, e);
+ } catch (final Exception fatalUserException) {
+ throw new StreamsException("Fatal user code error in processing error callback", fatalUserException);
Review Comment:
>> 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #16675:
URL: https://github.com/apache/kafka/pull/16675#discussion_r1689355081
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -214,9 +214,18 @@ public void process(final Record<KIn, VIn> record) {
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());
- final ProcessingExceptionHandler.ProcessingHandlerResponse response = processingExceptionHandler
- .handle(errorHandlerContext, record, e);
-
+ final ProcessingExceptionHandler.ProcessingHandlerResponse response;
+
+ try {
+ response = processingExceptionHandler
+ .handle(errorHandlerContext, record, e);
+ } catch (final Exception fatalUserException) {
+ log.error(
+ "Processing error callback failed after processing error for record {}",
+ record,
Review Comment:
We should not log records for privacy reasons. The exception should be enough.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -214,9 +214,18 @@ public void process(final Record<KIn, VIn> record) {
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());
- final ProcessingExceptionHandler.ProcessingHandlerResponse response = processingExceptionHandler
- .handle(errorHandlerContext, record, e);
-
+ final ProcessingExceptionHandler.ProcessingHandlerResponse response;
+
+ try {
+ response = processingExceptionHandler
+ .handle(errorHandlerContext, record, e);
Review Comment:
nit:
```suggestion
response = processingExceptionHandler.handle(errorHandlerContext, record, e);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #16675:
URL: https://github.com/apache/kafka/pull/16675#discussion_r1689386319
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -214,9 +214,18 @@ public void process(final Record<KIn, VIn> record) {
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());
- final ProcessingExceptionHandler.ProcessingHandlerResponse response = processingExceptionHandler
- .handle(errorHandlerContext, record, e);
-
+ final ProcessingExceptionHandler.ProcessingHandlerResponse response;
+
+ try {
+ response = processingExceptionHandler
+ .handle(errorHandlerContext, record, e);
+ } catch (final Exception fatalUserException) {
Review Comment:
Could you please also add a unit test for this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16675:
URL: https://github.com/apache/kafka/pull/16675#discussion_r1696289377
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -214,9 +214,13 @@ public void process(final Record<KIn, VIn> record) {
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());
- final ProcessingExceptionHandler.ProcessingHandlerResponse response = processingExceptionHandler
- .handle(errorHandlerContext, record, e);
+ final ProcessingExceptionHandler.ProcessingHandlerResponse response;
+ try {
+ response = processingExceptionHandler.handle(errorHandlerContext, record, e);
+ } catch (final Exception fatalUserException) {
+ throw new StreamsException("Fatal user code error in processing error callback", fatalUserException);
Review Comment:
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 :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
Posted by "loicgreffier (via GitHub)" <gi...@apache.org>.
loicgreffier commented on code in PR #16675:
URL: https://github.com/apache/kafka/pull/16675#discussion_r1697564800
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -214,9 +214,13 @@ public void process(final Record<KIn, VIn> record) {
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());
- final ProcessingExceptionHandler.ProcessingHandlerResponse response = processingExceptionHandler
- .handle(errorHandlerContext, record, e);
+ final ProcessingExceptionHandler.ProcessingHandlerResponse response;
+ try {
+ response = processingExceptionHandler.handle(errorHandlerContext, record, e);
+ } catch (final Exception fatalUserException) {
+ throw new StreamsException("Fatal user code error in processing error callback", fatalUserException);
Review Comment:
> 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:
https://github.com/apache/kafka/blob/683b8a2beeb5a75ef0a2ac878cf5a963f9b9d915/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java#L205
> Do we actually have a test for "passing through" of `FailedProcessingException`
We do:
https://github.com/apache/kafka/blob/683b8a2beeb5a75ef0a2ac878cf5a963f9b9d915/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java#L130
It asserts `FailedProcessingException` is not handled and thrown as is. Another test has been added to ensure that the cause of the caught exception is from the processing exception handler.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16675:
URL: https://github.com/apache/kafka/pull/16675#discussion_r1696289377
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -214,9 +214,13 @@ public void process(final Record<KIn, VIn> record) {
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());
- final ProcessingExceptionHandler.ProcessingHandlerResponse response = processingExceptionHandler
- .handle(errorHandlerContext, record, e);
+ final ProcessingExceptionHandler.ProcessingHandlerResponse response;
+ try {
+ response = processingExceptionHandler.handle(errorHandlerContext, record, e);
+ } catch (final Exception fatalUserException) {
+ throw new StreamsException("Fatal user code error in processing error callback", fatalUserException);
Review Comment:
One more thing: if we throw `StreamsException` here, we might catch it in an upstream `process()`, right? So should we rather wrap it 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 :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16448: Handle fatal user exception during processing error [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #16675:
URL: https://github.com/apache/kafka/pull/16675#discussion_r1696979756
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -214,9 +214,13 @@ public void process(final Record<KIn, VIn> record) {
internalProcessorContext.currentNode().name(),
internalProcessorContext.taskId());
- final ProcessingExceptionHandler.ProcessingHandlerResponse response = processingExceptionHandler
- .handle(errorHandlerContext, record, e);
+ final ProcessingExceptionHandler.ProcessingHandlerResponse response;
+ try {
+ response = processingExceptionHandler.handle(errorHandlerContext, record, e);
+ } catch (final Exception fatalUserException) {
+ throw new StreamsException("Fatal user code error in processing error callback", fatalUserException);
Review Comment:
done:
I changed the StreamException into a FailedProcessingException
added some test to check that no more processor is executed when an exception occurs or when an exception occurs in a handler.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org