KAFKA-16448: Unify error-callback exception handling by mjsax · Pull Request #16745 · apache/kafka

mjsax · 2024-07-31T07:33:44Z

@sebastienviale @loicgreffier -- as discussed on other PR review, I think we should apply some cleanup and unification for the different handler. Would love to get your feedback.

\cc @cadonna @lucasbru @bbejeck

mjsax · 2024-07-31T07:34:15Z

+            ", processorNodeId='" + processorNodeId + '\'' +
+            ", taskId=" + taskId +
+            '}';
+    }


Omitting headers on purpose, as we should not log user-data.

I know it's late in the game to ask this, but what about a timestamp? Just a thought, I don't have a strong opinion here.

Good question. Sounds like a miss... \cc @sebastienviale @loicgreffier

Side comment: I would not block this PR on this question, but do a follow up one, if we want to add ts.

Yep I agree it shouldn't block, adding a timestamp in a follow on PR is fine.

mjsax · 2024-07-31T07:34:40Z

                                globalProcessorContext.metrics()
-                            )
+                            ),
+                            null


Side cleanup as discussed on some other PR.

mjsax · 2024-07-31T07:35:00Z

            // Rethrow exceptions that should not be handled here
            throw e;
-        } catch (final RuntimeException e) {
+        } catch (final RuntimeException processingException) {


side cleanup: use proper variable names

mjsax · 2024-07-31T07:35:40Z

                keySerializer,
                exception);
-        } catch (final Exception exception) {
+        } catch (final RuntimeException serializationException) {


side cleanup: use proper variable names

mjsax · 2024-07-31T07:36:11Z


    private void recordSendError(final String topic,
-                                 final Exception exception,
+                                 final Exception productionException,


side cleanup: use proper variable names

mjsax · 2024-07-31T07:37:24Z

-                                                    final ConsumerRecord<byte[], byte[]> rawRecord,
-                                                    final Logger log,
-                                                    final Sensor droppedRecordsSensor) {
-        handleDeserializationFailure(deserializationExceptionHandler, processorContext, deserializationException, rawRecord, log, droppedRecordsSensor, null);


Side cleanup as discussed on some other PR

mjsax · 2024-07-31T07:41:11Z


        try {
            maybeMeasureLatency(() -> punctuator.punctuate(timestamp), time, punctuateLatencySensor);
+        } catch (final TimeoutException timeoutException) {


Totally unrelated (might actually to good to put into different PR, but want to avoid that I forget it).

Seems we forgot to handle this case entirely... we might call context.forward() inside of punctuate() and thus need to handle the same error cases as for node.process()...

Why a TimeoutException here? Could a punctuate potentially throw it?

Seems we forgot to handle this case entirely... we might call context.forward() inside of punctuate() and thus need to handle the same error cases as for node.process()...

We could end up in a sink node, and thus in RecordCollectorImpl / producer.send() which could throw a TimeoutException.

mjsax · 2024-07-31T07:43:39Z

-            log.error("Fatal when handling serialization exception", e);
-            recordSendError(topic, e, null, context, processorNodeId);
-            return;
+            throw new FailedProcessingException("Fatal user code error in production error callback", fatalUserException);


fix: inside send() is seem incorrect to call recordSendError() which we only use in from the producer.send(..., Calback) which is called async. Here inside RecordCollector.send() (before we hit producer.send()) can should rather throw directly

mjsax · 2024-07-31T07:45:17Z

+            originalException);
+    }
+
+    private void handleException(final String errorMessage, final Throwable originalException) {


this is an attempt to preserve the error message we create when a handler throws by itself.

mjsax · 2024-07-31T07:45:53Z

+    private void handleException(final Throwable originalException) {
+        handleException(
            String.format(
-                "Exception caught in process. taskId=%s, processor=%s, topic=%s, partition=%d, offset=%d, stacktrace=%s",


given what we pass in originalException into `StreamsException, it seems not necessary to include the stracktrace in the error message.

this seems to break two unit tests in ProcessingExceptionHandlerIntegrationTest while testing exception.getMessage

Thanks! -- I did only run a few test locally, not the whole suite. Will take a look.

sebastienviale · 2024-07-31T08:22:49Z

thanks for the PR, you're just ahead of me

sebastienviale · 2024-07-31T08:25:52Z

I just notice a nit in ProductionExceptionHandler

 * @param record        the record that failed to serialize
 * @param exception     the exception that occurred during serialization
 * @deprecated Since 3.9. Use {@link #handle(ErrorHandlerContext, ProducerRecord, Exception)} instead.
 */
@Deprecated
default ProductionExceptionHandlerResponse handleSerializationException(final ProducerRecord record,
                                                                        final Exception exception) {
    return ProductionExceptionHandlerResponse.FAIL;
}

should be

@deprecated Since 3.9. Use {@link #handleSerializationException(ErrorHandlerContext, ProducerRecord, Exception, SerializationExceptionOrigin)} instead.

loicgreffier · 2024-07-31T22:12:35Z

A very little typo Use Please brought by KAFKA-16448 here:

kafka/streams/src/main/java/org/apache/kafka/streams/errors/DeserializationExceptionHandler.java

Line 40 in 8d8c367

    
                * @deprecated Since 3.9. Use Please {@link #handle(ErrorHandlerContext, ConsumerRecord, Exception)}

mjsax · 2024-08-01T05:07:32Z

Updates this PR, and also extended it to align all 5 cases we have:

deserialzation
production / serialization
production / send
processing
punctuation

We now handle both error cases "handler throws exception" and "handler returns null" (this one I just realized while working on this PR and just added it).

mjsax · 2024-08-01T05:14:16Z


    @Test
-    public void shouldStopProcessingWhenFatalUserExceptionInFailProcessingExceptionHandler() {
+    public void shouldStopProcessingWhenProcessingExceptionHandlerReturnsNull() {


Re-purposed this test case for NULL case.

It seem redundant to test "throw exception" twice as originally in the code (one for FAIL and once for CONTINUE), because if we throw, there is no response we return at all...

mjsax · 2024-08-01T05:14:56Z

    public static class FailProcessingExceptionHandlerMockTest implements ProcessingExceptionHandler {
        @Override
        public ProcessingExceptionHandler.ProcessingHandlerResponse handle(final ErrorHandlerContext context, final Record<?, ?> record, final Exception exception) {
-            if (((String) record.key()).contains("FATAL")) {


Not needed any longer -- we just need one handler that can throw an exception

mjsax · 2024-08-01T05:15:42Z

            taskId,
            getExceptionalStreamsProducerOnSend(exception),
-            new ProductionExceptionHandlerMock(ProductionExceptionHandler.ProductionExceptionHandlerResponse.CONTINUE),
+            new ProductionExceptionHandlerMock(Optional.of(ProductionExceptionHandlerResponse.CONTINUE)),


Want to test 4 cases, so added Optional to make this possible.

mjsax · 2024-08-01T05:16:54Z

-        assertEquals(rawRecord.timestamp(), record.timestamp());
-        assertEquals(TimestampType.CREATE_TIME, record.timestampType());
-        assertEquals(rawRecord.headers(), record.headers());
+        try (final Metrics metrics = new Metrics()) {


Side cleanup. Metrics must be close to avoid resource leaks. (A few more times below.)

mjsax · 2024-08-01T05:17:57Z


    @Test
-    public void shouldPunctuateNotHandleFailProcessingExceptionAndThrowStreamsException() {
+    public void punctuateShouldNotHandleFailProcessingExceptionAndThrowStreamsException() {


renaming to make it readable :) (same below)

Plus some formatting changes to make the code readable

mjsax · 2024-08-01T05:18:31Z

+            LogAndContinueProcessingExceptionHandler.class.getName()
+        ));

-        assertDoesNotThrow(() ->


No need for this... if it would throw, the test fails anyway.

mjsax · 2024-08-02T06:30:23Z

-            if (uncaughtException != null || !e.getMessage().contains("Injected test exception")) {
+            if (uncaughtException != null ||
+                !(e instanceof StreamsException) ||
+                !e.getCause().getMessage().equals("Injected test exception.")) {


This change is required only, because the original test conditions was not "correct" anyway. The outer exception was already a StreamsException wrapping the RuntimeException the test throws. Testing the output exception for type RuntimeException does not make any sense anyway, as all non-checked exception are RuntimeException and the check passes...

The only actual change we get in this PR, is that the error message of the outer StreamsException changes -- originally, it was not set explicitly, and thus was set as "RuntimeException: Injected test exception" -- thus, even if we did check the "wrong/outer" error message the test passed. This PR now sets the StreamsException error message explicitly (what I don't consider a backward compatibility issue), but to this check, we need to check the error message of the wrapped exception (bonus, we can test for equals instead of contains).

mjsax · 2024-08-02T06:31:05Z

-                    !(e.getMessage().contains("test exception"))) {
+                if (++exceptionCount > 2 ||
+                    !(e instanceof StreamsException) ||
+                    !(e.getCause().getMessage().endsWith(" test exception."))) {


As above. However, we cannot use equals as the use different error messages for different situation when we throw.

bbejeck

Thanks for the PR @mjsax overall this looks good to me, I have a couple of minor comments.

bbejeck · 2024-08-01T21:51:24Z

+            ", processorNodeId='" + processorNodeId + '\'' +
+            ", taskId=" + taskId +
+            '}';
+    }


I know it's late in the game to ask this, but what about a timestamp? Just a thought, I don't have a strong opinion here.

bbejeck · 2024-08-02T17:38:38Z


        try {
            maybeMeasureLatency(() -> punctuator.punctuate(timestamp), time, punctuateLatencySensor);
+        } catch (final TimeoutException timeoutException) {


Why a TimeoutException here? Could a punctuate potentially throw it?

bbejeck · 2024-08-02T18:03:28Z

                                                         final Exception exception) {
            assertInputs(context, exception);
-            return response;
+            if (response == null) {


I don't think the response can be null should this be response.isEmpty?

null is on purpose. It's how we setup the code (it's not something the handler return).

In the end response is "what should the handler return in this test run", and I miss use it to say "handler should throw an exception". I could also add a new boolean flag shouldThrowException instead to not miss use response for this case.

Will update the PR.

bbejeck · 2024-08-02T18:04:00Z

            assertInstanceOf(RuntimeException.class, exception);
            assertEquals("KABOOM!", exception.getMessage());
-            return response;
+            if (response == null) {


same as above here

bbejeck

LGTM

Follow up code cleanup for KIP-1033. This PR unifies the handling of both error cases for exception handlers: - handler throws an exception - handler returns null The unification happens for all 5 handler cases: - deserialzation - production / serialization - production / send - processing - punctuation Reviewers: Sebastien Viale <sebastien.viale@michelin.com>, Loic Greffier <loic.greffier@michelin.com>, Bill Bejeck <bill@confluent.io>

mjsax · 2024-08-03T20:18:55Z

Merged to trunk and cherry-picked to 3.9 branch.

mjsax added streams kip Requires or implements a KIP labels Jul 31, 2024

loicgreffier mentioned this pull request Jul 31, 2024

KAFKA-16448: Unifiy deserialization failure handling #16746

Closed

mjsax commented Jul 31, 2024

View reviewed changes

mjsax added 2 commits July 31, 2024 22:00

KAFKA-16448: Unify error-callback exception handling

e08da1a

fix typo

f13ae88

mjsax force-pushed the minor-kip-1033 branch from c239d50 to e08da1a Compare August 1, 2024 05:03

mjsax commented Aug 1, 2024

View reviewed changes

mjsax added 2 commits July 31, 2024 22:19

cleanup

87cef8f

adjust tests

507b980

mjsax commented Aug 2, 2024

View reviewed changes

bbejeck reviewed Aug 2, 2024

View reviewed changes

Github comments

646ca1e

bbejeck approved these changes Aug 2, 2024

View reviewed changes

mjsax merged commit 7c539f9 into apache:trunk Aug 3, 2024

mjsax deleted the minor-kip-1033 branch August 3, 2024 19:41

Conversation

mjsax commented Jul 31, 2024

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!

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!

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!

sebastienviale commented Jul 31, 2024

Uh oh!

sebastienviale commented Jul 31, 2024

Uh oh!

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

Uh oh!

Uh oh!

mjsax commented Aug 1, 2024

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

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

mjsax Aug 2, 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!

bbejeck left a comment

loicgreffier commented Jul 31, 2024 •

edited

Loading

mjsax Aug 2, 2024 •

edited

Loading