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/08/02 08:49:20 UTC
[PR] KAFKA-16448 Update documentation [kafka]
sebastienviale opened a new pull request, #16776:
URL: https://github.com/apache/kafka/pull/16776
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 updates the documentation
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 Update documentation [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #16776:
URL: https://github.com/apache/kafka/pull/16776#issuecomment-2265797834
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 Update documentation [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #16776:
URL: https://github.com/apache/kafka/pull/16776#discussion_r1702087746
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -395,83 +396,88 @@ <h4><a class="toc-backref" href="#id23">num.standby.replicas</a><a class="header
<td colspan="2">The maximum time in milliseconds to wait before triggering a rebalance to probe for warmup replicas that have sufficiently caught up.</td>
<td>600000 milliseconds (10 minutes)</td>
</tr>
- <tr class="row-even"><td>processing.guarantee</td>
+ <tr class="row-even"><td>processing.exception.handler</td>
+ <td>Medium</td>
+ <td colspan="2">Exception handling class that implements the <code class="docutils literal"><span class="pre">ProcessingExceptionHandler</span></code> interface.</td>
+ <td><code class="docutils literal"><span class="pre">ProcessingExceptionHandler</span></code></td>
Review Comment:
```suggestion
<td><code class="docutils literal"><span class="pre">LogAndFailProcessingExceptionHandler</span></code></td>
```
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -924,6 +931,56 @@ <h4><a class="toc-backref" href="#id30">probing.rebalance.interval.ms</a><a clas
them to active tasks if ready. They will continue to be triggered as long as there are warmup tasks, and until the assignment is balanced. Must be at least 1 minute.
</div></blockquote>
</div>
+ <div class="section" id="processing-exception-handler">
+ <span id="streams-developer-guide-proceh"></span><h4><a class="toc-backref" href="#id41">processing.exception.handler</a><a class="headerlink" href="#processing-exception-handler" title="Permalink to this headline"></a></h4>
+ <blockquote>
+ <div><p>The processing exception handler allows you to manage exceptions triggered during the processing of a record. The implemented exception
+ handler needs to return a <code>FAIL</code> or <code>CONTINUE</code> depending on the record and the exception thrown. Returning
+ <code>FAIL</code> will signal that Streams should shut down and <code>CONTINUE</code> will signal that Streams should ignore the issue
+ and continue processing. The following library built-in exception handlers are available:</p>
+ <ul class="simple">
+ <li><a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/streams/errors/LogAndContinueProcessingExceptionHandler.html">LogAndContinueProcessingExceptionHandler</a>:
+ This handler logs the processing exception and then signals the processing pipeline to continue processing more records.
+ This log-and-skip strategy allows Kafka Streams to make progress instead of failing if there are records that fail to be processed.</li>
+ <li><a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/streams/errors/LogAndFailProcessingExceptionHandler.html">LogAndFailProcessingExceptionHandler</a>.
+ This handler logs the processing exception and then signals the processing pipeline to stop processing more records.</li>
+ </ul>
+
+ <p>You can also provide your own customized exception handler besides the library provided ones to meet your needs. For example, you can choose to forward corrupt
+ records into a quarantine topic (think: a "dead letter queue") for further processing. To do this, use the Producer API to write a corrupted record directly to
+ the quarantine topic. To be more concrete, you can create a separate <code>KafkaProducer</code> object outside the Streams client, and pass in this object
+ as well as the dead letter queue topic name into the <code>Properties</code> map, which then can be retrieved from the <code>configure</code> function call.
+ The drawback of this approach is that "manual" writes are side effects that are invisible to the Kafka Streams runtime library,
+ so they do not benefit from the end-to-end processing guarantees of the Streams API:</p>
+
+ <pre class="line-numbers"><code class="language-java">public class SendToDeadLetterQueueExceptionHandler implements ProcessingExceptionHandler {
+ KafkaProducer<byte[], byte[]> dlqProducer;
+ String dlqTopic;
+
+ @Override
+ public ProcessingHandlerResponse handle(final ErrorHandlerContext context,
+ final Record<?, ?> record,
+ final Exception exception) {
Review Comment:
```suggestion
final Exception exception) {
```
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -924,6 +931,56 @@ <h4><a class="toc-backref" href="#id30">probing.rebalance.interval.ms</a><a clas
them to active tasks if ready. They will continue to be triggered as long as there are warmup tasks, and until the assignment is balanced. Must be at least 1 minute.
</div></blockquote>
</div>
+ <div class="section" id="processing-exception-handler">
+ <span id="streams-developer-guide-proceh"></span><h4><a class="toc-backref" href="#id41">processing.exception.handler</a><a class="headerlink" href="#processing-exception-handler" title="Permalink to this headline"></a></h4>
+ <blockquote>
+ <div><p>The processing exception handler allows you to manage exceptions triggered during the processing of a record. The implemented exception
+ handler needs to return a <code>FAIL</code> or <code>CONTINUE</code> depending on the record and the exception thrown. Returning
+ <code>FAIL</code> will signal that Streams should shut down and <code>CONTINUE</code> will signal that Streams should ignore the issue
+ and continue processing. The following library built-in exception handlers are available:</p>
+ <ul class="simple">
+ <li><a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/streams/errors/LogAndContinueProcessingExceptionHandler.html">LogAndContinueProcessingExceptionHandler</a>:
+ This handler logs the processing exception and then signals the processing pipeline to continue processing more records.
+ This log-and-skip strategy allows Kafka Streams to make progress instead of failing if there are records that fail to be processed.</li>
+ <li><a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/streams/errors/LogAndFailProcessingExceptionHandler.html">LogAndFailProcessingExceptionHandler</a>.
+ This handler logs the processing exception and then signals the processing pipeline to stop processing more records.</li>
+ </ul>
+
+ <p>You can also provide your own customized exception handler besides the library provided ones to meet your needs. For example, you can choose to forward corrupt
+ records into a quarantine topic (think: a "dead letter queue") for further processing. To do this, use the Producer API to write a corrupted record directly to
+ the quarantine topic. To be more concrete, you can create a separate <code>KafkaProducer</code> object outside the Streams client, and pass in this object
+ as well as the dead letter queue topic name into the <code>Properties</code> map, which then can be retrieved from the <code>configure</code> function call.
+ The drawback of this approach is that "manual" writes are side effects that are invisible to the Kafka Streams runtime library,
+ so they do not benefit from the end-to-end processing guarantees of the Streams API:</p>
+
+ <pre class="line-numbers"><code class="language-java">public class SendToDeadLetterQueueExceptionHandler implements ProcessingExceptionHandler {
+ KafkaProducer<byte[], byte[]> dlqProducer;
+ String dlqTopic;
+
+ @Override
+ public ProcessingHandlerResponse handle(final ErrorHandlerContext context,
+ final Record<?, ?> record,
Review Comment:
```suggestion
final Record<?, ?> record,
```
--
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 Update documentation [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #16776:
URL: https://github.com/apache/kafka/pull/16776
--
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