KAFKA-16508: Streams custom handler should handle the timeout exceptions #16450

aliehsaeedii · 2024-06-25T16:45:38Z

No description provided.

mjsax · 2024-06-25T18:37:33Z

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

@@ -306,6 +307,15 @@ private void recordSendError(final String topic, final Exception exception, fina
            sendException.set(new TaskMigratedException(errorMessage, exception));
        } else {
            if (exception instanceof RetriableException) {
+                if (exception instanceof TimeoutException && exception.getCause() != null) {
+                    if (exception.getCause() instanceof UnknownTopicOrPartitionException) {


Should we not use these three nested ifs, but add a new if (isRetriable(excption)) { ... } helper method?

Would also avoid to call the handler on two places, as isRetriable() return false we just re-use the code in the else path?

Would also avoid to call the handler on two places, as isRetriable() return false we just re-use the code in the else path?

I assumed if we have an UnknownTopicOrPartition exception and the handler instructs to CONTINUE, we must follow the old path which is sendException.set(new TaskCorruptedException(Collections.singleton(taskId))); BTW, I changed the code as you wished.

I don't think so. If the hander returns CONTINUE we should execute the "continues" code, pass, ie, the existing else to just drop the record and record it:
https://github.com/apache/kafka/pull/16450/files#diff-36dd8c03fa5252dbd39042bb49a0d6272728fbd46459c6a45f5189fa59749b32R322-R323

mjsax · 2024-06-25T22:35:54Z

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

@@ -1350,6 +1351,30 @@ public void shouldNotThrowStreamsExceptionOnSubsequentCallIfASendFailsWithContin
        collector.closeClean();
    }

+    @Test
+    public void shouldThrowStreamsExceptionOnUnknownTopicOrPartitionExceptionWithDefaultExceptionHandler() {


If we don't call the handler, what exception is thrown from flush()? It is just the plain/unwrapped TimeoutExcpetion?

If we don't call the default handler, it throws TaskCorruptedException with a null cause. This is what recordSendError() method determines.

mjsax · 2024-06-25T22:37:37Z

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

@@ -306,6 +307,15 @@ private void recordSendError(final String topic, final Exception exception, fina
            sendException.set(new TaskMigratedException(errorMessage, exception));
        } else {
            if (exception instanceof RetriableException) {
+                if (exception instanceof TimeoutException && exception.getCause() != null) {
+                    if (exception.getCause() instanceof UnknownTopicOrPartitionException) {
+                        if (productionExceptionHandler.handle(serializedRecord, exception) == ProductionExceptionHandlerResponse.FAIL) {


Wondering if we should hand in exception or exception.getCause()? Thoughts?

mjsax · 2024-06-25T23:57:05Z

streams/src/test/java/org/apache/kafka/streams/integration/CustomHandlerIntegrationTest.java

+                .to(NON_EXISTING_TOPIC, Produced.with(Serdes.Integer(), Serdes.String()));
+        produceRecords();
+        startApplication();
+        closeApplication();


I think there is a race condition? In the end, we don't know if the record was even processed? We could call KafkaStreams#close() before the record was processed?

Also, with the default production handler which return "FAIL", should we not expect Kafka Streams to go into ERROR state instead of NOT_RUNNING?

mjsax · 2024-06-28T15:35:31Z

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

@@ -305,14 +306,16 @@ private void recordSendError(final String topic, final Exception exception, fina
                "indicating the task may be migrated out";
            sendException.set(new TaskMigratedException(errorMessage, exception));
        } else {
-            if (exception instanceof RetriableException) {
+            final ProductionExceptionHandlerResponse response = productionExceptionHandler.handle(serializedRecord, exception);


I don't think we need to call the handler for all cases here, but only if the exception is non-retriable (old, existing condition), or if the exception is a timeout with root cause "unknown topic or partition"?

So something like:

if (!isRetriable(exception)) { // we only change this line but the rest stays the same? errorMessage += "\nThe broker is either slow or in bad state (like not having enough replicas) in responding the request, " + "or the connection to broker was interrupted sending the request or receiving the response. " + "\nConsider overwriting `max.block.ms` and /or " + "`delivery.timeout.ms` to a larger value to wait longer for such scenarios and avoid timeout errors"; sendException.set(new TaskCorruptedException(Collections.singleton(taskId))); } else { if (productionExceptionHandler.handle(serializedRecord, exception) == ProductionExceptionHandlerResponse.FAIL) { errorMessage += "\nException handler choose to FAIL the processing, no more records would be sent."; sendException.set(new StreamsException(errorMessage, exception)); } else { errorMessage += "\nException handler choose to CONTINUE processing in spite of this error but written offsets would not be recorded."; droppedRecordsSensor.record(); }

If the exception is Timeout with root cause UnknownTopicOrPartition and the handler instructs to CONTINUE, we must follow the old logic, which is having TaskCorruptedException. Therefore, I think we have to call handler for nearly all cases.

mjsax · 2024-06-28T16:00:05Z

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

@@ -325,6 +328,11 @@ private void recordSendError(final String topic, final Exception exception, fina
        log.error(errorMessage, exception);
    }

+    private boolean isMissingMetadata (final Exception exception) {


Should we rewrite this to:

private boolean isRetriable (final Exception exception) { return !(exception instanceOf TimeoutException && exception.getCause() != null && exception.getCause() instanceof UnknownTopicOrPartitionException()) || exception instanceOf RetriableExcepiton; }

Hope I got the logic right... Basically we want to say, it is retriable if it is a RetryableExcetpion but not if it is a TimeoutException with root-cause "unknonw topic or partition".

mjsax

Just a few nits. Overall LGTM.

mjsax · 2024-06-28T18:10:27Z

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

@@ -325,6 +326,11 @@ private void recordSendError(final String topic, final Exception exception, fina
        log.error(errorMessage, exception);
    }

+    private boolean isRetriable(final Exception exception) {
+        return (!(exception instanceof TimeoutException) || exception.getCause() == null


This logic is complex. Let's a comment about what it actually does

// We don't consider a `TimeoutException` with root cause `UnknownTopicOrPartitionException` retriable, // even if `TimeoutException extends RetriableException` and thus exclude this case expliclity

mjsax · 2024-06-28T18:15:00Z

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

@@ -325,6 +326,11 @@ private void recordSendError(final String topic, final Exception exception, fina
        log.error(errorMessage, exception);
    }

+    private boolean isRetriable(final Exception exception) {
+        return (!(exception instanceof TimeoutException) || exception.getCause() == null
+                || !(exception.getCause() instanceof UnknownTopicOrPartitionException)) && exception instanceof RetriableException;


Nit: Given how you wrote the predicate, might be easier to read if we put exception instanceof RetriableException && ... at the beginning, and have the complex expression at the end?

streams/src/test/java/org/apache/kafka/streams/integration/CustomHandlerIntegrationTest.java

mjsax · 2024-06-28T18:19:12Z

streams/src/test/java/org/apache/kafka/streams/integration/CustomHandlerIntegrationTest.java

+        while (true) {
+            if (caughtException.get() != null) {
+                assertInstanceOf(StreamsException.class, caughtException.get());
+                assertInstanceOf(TimeoutException.class, caughtException.get().getCause());


Should we add one more verifying the cause of TimeoutException, too?

mjsax · 2024-06-28T18:53:52Z

streams/src/test/java/org/apache/kafka/streams/integration/CustomHandlerIntegrationTest.java

+        IntegrationTestUtils.purgeLocalStreamsState(streamsConfiguration);
+        final long timeout = 60000;
+        TestUtils.waitForCondition(
+                () -> kafkaStreams.state() == State.NOT_RUNNING,


Given that the uncaught exception handler returns SHUTDOWN_CLIENT the final state should be ERROR IIRC?

mjsax

Thanks. LGTM assuming Jenkins passes.

KAFKA-16508: Streams custom handler should handle the timeout exceptions

fc4d155

mjsax added the streams label Jun 25, 2024

mjsax reviewed Jun 26, 2024

View reviewed changes

aliehsaeedii added 3 commits June 27, 2024 22:44

addressed reviews

add8c4d

add utest

c11260c

remove strange file

d970741

mjsax approved these changes Jun 28, 2024

View reviewed changes

aliehsaeedii added 3 commits June 28, 2024 18:12

add integration test

c115494

cs

2c2a3f1

correc the logic

857f8a2

mjsax approved these changes Jun 28, 2024

View reviewed changes

mjsax reviewed Jun 28, 2024

View reviewed changes

aliehsaeedii added 3 commits June 28, 2024 20:57

address reviews

22c54d4

assert ks error state

1a544d2

remove final state check

6a8eb68

mjsax approved these changes Jun 28, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-16508: Streams custom handler should handle the timeout exceptions #16450

KAFKA-16508: Streams custom handler should handle the timeout exceptions #16450

aliehsaeedii commented Jun 25, 2024 •

edited

Loading

mjsax Jun 25, 2024

aliehsaeedii Jun 27, 2024

mjsax Jun 28, 2024

mjsax Jun 25, 2024 •

edited

Loading

aliehsaeedii Jun 27, 2024 •

edited

Loading

mjsax Jun 25, 2024

mjsax Jun 25, 2024

mjsax Jun 28, 2024

aliehsaeedii Jun 28, 2024

mjsax Jun 28, 2024

mjsax left a comment

mjsax Jun 28, 2024

mjsax Jun 28, 2024

mjsax Jun 28, 2024

mjsax Jun 28, 2024 •

edited

Loading

aliehsaeedii Jun 28, 2024

mjsax left a comment

KAFKA-16508: Streams custom handler should handle the timeout exceptions #16450

Are you sure you want to change the base?

KAFKA-16508: Streams custom handler should handle the timeout exceptions #16450

Conversation

aliehsaeedii commented Jun 25, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax Jun 25, 2024 • edited Loading

Choose a reason for hiding this comment

aliehsaeedii Jun 27, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax Jun 28, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax left a comment

Choose a reason for hiding this comment

aliehsaeedii commented Jun 25, 2024 •

edited

Loading

mjsax Jun 25, 2024 •

edited

Loading

aliehsaeedii Jun 27, 2024 •

edited

Loading

mjsax Jun 28, 2024 •

edited

Loading