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/23 07:36:20 UTC
[PR] KFK-16448: Fix processing exception handler [kafka]
sebastienviale opened a new pull request, #16663:
URL: https://github.com/apache/kafka/pull/16663
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 resolves some minor fixes
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: Fix processing exception handler [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #16663:
URL: https://github.com/apache/kafka/pull/16663#discussion_r1689491260
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -800,35 +800,39 @@ record = null;
record = null;
throw new TaskCorruptedException(Collections.singleton(id));
}
+ } catch (final FailedProcessingException failedProcessingException) {
+ // Do not keep the failed processing exception in the stack trace
+ handleException(new Exception(failedProcessingException.getCause()));
} catch (final StreamsException exception) {
record = null;
throw exception;
} catch (final RuntimeException e) {
Review Comment:
yes indeed, I think I missed a commit.
--
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: Fix processing exception handler [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #16663:
URL: https://github.com/apache/kafka/pull/16663#discussion_r1687653898
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -800,35 +800,39 @@ record = null;
record = null;
throw new TaskCorruptedException(Collections.singleton(id));
}
+ } catch (final FailedProcessingException failedProcessingException){
+ // Do not keep the failed processing exception in the stack trace
+ handleException(failedProcessingException.getCause());
} catch (final StreamsException exception) {
record = null;
throw exception;
} catch (final RuntimeException e) {
- // Do not keep the failed processing exception in the stack trace
- final Throwable processingException = e instanceof FailedProcessingException ? e.getCause() : e;
-
- final StreamsException error = new StreamsException(
- String.format(
- "Exception caught in process. taskId=%s, processor=%s, topic=%s, partition=%d, offset=%d, stacktrace=%s",
- id(),
- processorContext.currentNode().name(),
- record.topic(),
- record.partition(),
- record.offset(),
- getStacktraceString(processingException)
- ),
- processingException
- );
- record = null;
-
- throw error;
+ handleException(e);
} finally {
processorContext.setCurrentNode(null);
}
return true;
}
+ private void handleException(final Throwable e) {
Review Comment:
nit:
```suggestion
private void handleException(final Exception 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: Fix processing exception handler [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #16663:
URL: https://github.com/apache/kafka/pull/16663#issuecomment-2251555673
Thanks for the follow-up PR. Merged to `trunk`.
--
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: Fix processing exception handler [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #16663:
URL: https://github.com/apache/kafka/pull/16663#discussion_r1689383338
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -800,35 +800,39 @@ record = null;
record = null;
throw new TaskCorruptedException(Collections.singleton(id));
}
+ } catch (final FailedProcessingException failedProcessingException) {
+ // Do not keep the failed processing exception in the stack trace
+ handleException(new Exception(failedProcessingException.getCause()));
} catch (final StreamsException exception) {
record = null;
throw exception;
} catch (final RuntimeException e) {
Review Comment:
Didn't we agree on using `Exception` here instead of `RuntimeException` to be consistent with other handlers?
--
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: Fix processing exception handler [kafka]
Posted by "loicgreffier (via GitHub)" <gi...@apache.org>.
loicgreffier commented on code in PR #16663:
URL: https://github.com/apache/kafka/pull/16663#discussion_r1689492785
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -800,35 +800,39 @@ record = null;
record = null;
throw new TaskCorruptedException(Collections.singleton(id));
}
+ } catch (final FailedProcessingException failedProcessingException) {
+ // Do not keep the failed processing exception in the stack trace
+ handleException(new Exception(failedProcessingException.getCause()));
} catch (final StreamsException exception) {
record = null;
throw exception;
} catch (final RuntimeException e) {
Review Comment:
Fixed
--
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: Fix processing exception handler [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna closed pull request #16663: KAFKA-16448: Fix processing exception handler
URL: https://github.com/apache/kafka/pull/16663
--
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: Fix processing exception handler [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #16663:
URL: https://github.com/apache/kafka/pull/16663#discussion_r1689341148
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -800,35 +800,39 @@ record = null;
record = null;
throw new TaskCorruptedException(Collections.singleton(id));
}
+ } catch (final FailedProcessingException failedProcessingException){
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: Fix processing exception handler [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #16663:
URL: https://github.com/apache/kafka/pull/16663#discussion_r1689272059
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -800,35 +800,39 @@ record = null;
record = null;
throw new TaskCorruptedException(Collections.singleton(id));
}
+ } catch (final FailedProcessingException failedProcessingException){
Review Comment:
Checkstyle is complaining here. You need to change it to the following:
```suggestion
} catch (final FailedProcessingException 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: Fix processing exception handler [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #16663:
URL: https://github.com/apache/kafka/pull/16663#discussion_r1689372316
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -800,35 +800,39 @@ record = null;
record = null;
throw new TaskCorruptedException(Collections.singleton(id));
}
+ } catch (final FailedProcessingException failedProcessingException){
+ // Do not keep the failed processing exception in the stack trace
+ handleException(failedProcessingException.getCause());
} catch (final StreamsException exception) {
record = null;
throw exception;
} catch (final RuntimeException e) {
- // Do not keep the failed processing exception in the stack trace
- final Throwable processingException = e instanceof FailedProcessingException ? e.getCause() : e;
-
- final StreamsException error = new StreamsException(
- String.format(
- "Exception caught in process. taskId=%s, processor=%s, topic=%s, partition=%d, offset=%d, stacktrace=%s",
- id(),
- processorContext.currentNode().name(),
- record.topic(),
- record.partition(),
- record.offset(),
- getStacktraceString(processingException)
- ),
- processingException
- );
- record = null;
-
- throw error;
+ handleException(e);
} finally {
processorContext.setCurrentNode(null);
}
return true;
}
+ private void handleException(final Throwable e) {
Review Comment:
Why did you change `private void handleException(final Throwable e)` to `private void handleException(final Exception e)`? The additional wrapping into an `Exception` is not worth changing the signature. Please revert to `private void handleException(final Throwable e)` and pass in `failedProcessingException.getCause()`.
See my comment above.
--
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: Fix processing exception handler [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #16663:
URL: https://github.com/apache/kafka/pull/16663#discussion_r1687778090
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -800,35 +800,39 @@ record = null;
record = null;
throw new TaskCorruptedException(Collections.singleton(id));
}
+ } catch (final FailedProcessingException failedProcessingException){
+ // Do not keep the failed processing exception in the stack trace
+ handleException(failedProcessingException.getCause());
} catch (final StreamsException exception) {
record = null;
throw exception;
} catch (final RuntimeException e) {
- // Do not keep the failed processing exception in the stack trace
- final Throwable processingException = e instanceof FailedProcessingException ? e.getCause() : e;
-
- final StreamsException error = new StreamsException(
- String.format(
- "Exception caught in process. taskId=%s, processor=%s, topic=%s, partition=%d, offset=%d, stacktrace=%s",
- id(),
- processorContext.currentNode().name(),
- record.topic(),
- record.partition(),
- record.offset(),
- getStacktraceString(processingException)
- ),
- processingException
- );
- record = null;
-
- throw error;
+ handleException(e);
} finally {
processorContext.setCurrentNode(null);
}
return true;
}
+ private void handleException(final Throwable e) {
Review Comment:
@sebastienviale found that `failedProcessingException.getCause()` returns a `Throwable`. I think casting would be overhead. Let's keep `private void handleException(final Throwable 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: Fix processing exception handler [kafka]
Posted by "loicgreffier (via GitHub)" <gi...@apache.org>.
loicgreffier commented on code in PR #16663:
URL: https://github.com/apache/kafka/pull/16663#discussion_r1689427013
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -800,35 +800,39 @@ record = null;
record = null;
throw new TaskCorruptedException(Collections.singleton(id));
}
+ } catch (final FailedProcessingException failedProcessingException){
+ // Do not keep the failed processing exception in the stack trace
+ handleException(failedProcessingException.getCause());
} catch (final StreamsException exception) {
record = null;
throw exception;
} catch (final RuntimeException e) {
- // Do not keep the failed processing exception in the stack trace
- final Throwable processingException = e instanceof FailedProcessingException ? e.getCause() : e;
-
- final StreamsException error = new StreamsException(
- String.format(
- "Exception caught in process. taskId=%s, processor=%s, topic=%s, partition=%d, offset=%d, stacktrace=%s",
- id(),
- processorContext.currentNode().name(),
- record.topic(),
- record.partition(),
- record.offset(),
- getStacktraceString(processingException)
- ),
- processingException
- );
- record = null;
-
- throw error;
+ handleException(e);
} finally {
processorContext.setCurrentNode(null);
}
return true;
}
+ private void handleException(final Throwable e) {
Review Comment:
Reverted. Probably a mistake in a recent commit
--
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: Fix processing exception handler [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #16663:
URL: https://github.com/apache/kafka/pull/16663
--
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