KAFKA-16448: Add ErrorHandlerContext in deserialization exception handler by sebastienviale · Pull Request #16432 · apache/kafka

sebastienviale · 2024-06-24T08:03:21Z

This PR is part of KAFKA-16448 which aims to bring a ProcessingExceptionHandler to Kafka Streams in order to deal with exceptions that occur during processing.

This PR expose the new ErrorHandlerContext as a parameter to the Deserialization exception handlers and deprecate the previous handle signatur

Jira: https://issues.apache.org/jira/browse/KAFKA-16448.

Contributors
@Dabz
@sebastienviale
@loicgreffier

cadonna

Thanks for the PR @sebastienviale , @loicgreffier, and @Dabz!

Here my feeback!

cadonna · 2024-07-23T09:18:39Z

+        );
+
+        final StreamsException e = assertThrows(StreamsException.class, () -> recordDeserializer.deserialize(context, rawRecord));
+        assertEquals(e.getMessage(), "Deserialization exception handler is set to fail upon a deserialization error. If you would rather have the streaming pipeline continue after a deserialization error, please set the default.deserialization.exception.handler appropriately.");


This line is definitely too long. Please add some new lines.

cadonna · 2024-07-23T09:19:09Z

+        "true, false",
+        "false, true"
+    })
+    public void shouldThrowStreamsExceptionWhenDeserializationFailsAndDeserializationExceptionHandlerRepliesWithFail(final boolean keyThrowsException,


nit:

Suggested change

public void shouldThrowStreamsExceptionWhenDeserializationFailsAndDeserializationExceptionHandlerRepliesWithFail(final boolean keyThrowsException,

public void shouldThrowStreamsExceptionWhenDeserializationFailsAndExceptionHandlerRepliesWithFail(final boolean keyThrowsException,

cadonna · 2024-07-23T09:19:50Z

+        "true, false",
+        "false, true"
+    })
+    public void shouldNotThrowStreamsExceptionWhenDeserializationFailsAndDeserializationExceptionHandlerRepliesWithContinue(final boolean keyThrowsException,


nit:

Suggested change

public void shouldNotThrowStreamsExceptionWhenDeserializationFailsAndDeserializationExceptionHandlerRepliesWithContinue(final boolean keyThrowsException,

public void shouldNotThrowStreamsExceptionWhenDeserializationFailsAndExceptionHandlerRepliesWithContinue(final boolean keyThrowsException,

cadonna · 2024-07-23T09:26:53Z

+    @ParameterizedTest
+    @CsvSource({
+        "true, false",
+        "false, true"


Do you also need to test "true, true"?

true, true should fail and exit on the deserialization of the key just like true, false. As the test is parametized, it does not cost so much to add this scenario.

cadonna · 2024-07-23T09:32:02Z

+                sourceNodeName,
+                processorContext.taskId());
+            response = deserializationExceptionHandler.handle(errorHandlerContext, rawRecord, deserializationException);
        } catch (final Exception fatalUserException) {


I realized now that you did not consider that the exception handler can also throw an exception for the processing exception handler (and I as the reviewer also missed it). Could you open a separate PR that catches exceptions from the processing exception handler and re-throws them as it is done here?
Sorry for not noticing it earlier!

@cadonna As processing exception handler handles exception at ProcessorNode, indeed exceptions during deserialization are not caught right now. Is it your point?

Maybe we should consider catching and handling exceptions when getting the nextRecord at StreamTask level right before processing it:

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

Line 770 in a5bfc21

record = partitionGroup.nextRecord(recordInfo, wallClockTime);

Going to sleep on it

I think, we misunderstood each other. On this line an exception originating from the user-specified handler is caught. The exception can be anything that is thrown by user code. We did not consider exceptions thrown from user code for the processing exception handler.

done in this PR #16675

done in this PR #16300 for Punctuating part

cadonna

Thanks for the updates @sebastienviale, @loicgreffier, and @Dabz!

LGTM!

mjsax

Overall LGTM. A few minor things.

mjsax · 2024-07-26T18:07:03Z

     * @param context processor context
     * @param record record that failed deserialization
     * @param exception the actual exception
+     * @deprecated Please {@link #handle(ErrorHandlerContext, ConsumerRecord, Exception)}


Suggested change

* @deprecated Please {@link #handle(ErrorHandlerContext, ConsumerRecord, Exception)}

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

Not sure about already merged PRs, but if they also have deprecation notice, this the the format we usually use. Including the version helps user to see how far behind they are and when the can expect the old API to be remove, so we want to include it.

Can you double check other PRs so we can update stuff if necessary?

done
I wil check other PRs to be sure

mjsax · 2024-07-26T18:17:47Z

+    default DeserializationHandlerResponse handle(final ProcessorContext context,
+                                                  final ConsumerRecord<byte[], byte[]> record,
+                                                  final Exception exception) {
+        throw new UnsupportedOperationException();


Nit. KIP says NotImplementedException() -- we should update the KIP accordingly.

done KIP has been updated

mjsax · 2024-07-26T18:20:45Z

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


Why do we still need this one? Can't we update GlobalStateManagerImpl to also pass in a sourceNode.name ?

@sebastienviale What about this comments? Merge the PR for now to make progress.

@mjsax I've been looking for a way to update GlobalStateManagerImpl so we can remove this method. But it looks like we have no source node name in GlobalStateManagerImpl. There's a "source" processor:

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

Line 315 in 010ab19

source.process(new Record<>(

But it doesn't look to be named. E.g., in the following scenario:

streamsBuilder .addGlobalStore(storeBuilder, "PERSON_TOPIC", Consumed.with(Serdes.String(), SerdesUtils.specificAvroValueSerdes()), () -> (Processor<String, SpecificRecord, Void, Void>) record -> log.info("Processing record with key = {}, value = {}", record.key(), record.value())); // Source var is this processor ⬆️

Well, the "helper" method also only calls the second one, setting sourceNodeName to null. Why can't we just call the second one from GlobalStateManagerImpl directly also setting it to null, and thus avoiding the need for the overload? The overload does not really add much value, or does it?

@mjsax Addressing it in #16746

mjsax · 2024-07-26T18:22:28Z

        this.taskId = taskId;
    }

+    public DefaultErrorHandlerContext(final ProcessorContext processorContext,


Should this constructor replace the other one? And we just pass in null for processorContext?

Or we use setRecordContext() -- but having constructor overloads is always something I think is better to avoid if we can.

mjsax · 2024-07-26T18:22:52Z

        return this.taskId;
    }
+
+    public ProcessorContext processorContext() {


Should this return Optional<ProcessorContext> given that it could be null ?

…ndler Co-authored-by: Dabz <d.gasparina@gmail.com> Co-authored-by: loicgreffier <loic.greffier@michelin.com>

…ndler remove rawKey and rawValue Co-authored-by: Dabz <d.gasparina@gmail.com> Co-authored-by: loicgreffier <loic.greffier@michelin.com>

…ndler Co-authored-by: Dabz <d.gasparina@gmail.com> Co-authored-by: loicgreffier <loic.greffier@michelin.com>

…dler (#16432) This PR is part of KIP1033 which aims to bring a ProcessingExceptionHandler to Kafka Streams in order to deal with exceptions that occur during processing. This PR expose the new ErrorHandlerContext as a parameter to the Deserialization exception handlers and deprecate the previous handle signature. Co-authored-by: Dabz <d.gasparina@gmail.com> Co-authored-by: loicgreffier <loic.greffier@michelin.com> Reviewers: Bruno Cadonna <bruno@confluent.io>, Matthias J. Sax <matthias@confluent.io>

mjsax · 2024-07-30T04:24:33Z

Thanks for the PR.

Merged to trunk and cherry-picked to 3.9 branch.

…dler (apache#16432) This PR is part of KIP1033 which aims to bring a ProcessingExceptionHandler to Kafka Streams in order to deal with exceptions that occur during processing. This PR expose the new ErrorHandlerContext as a parameter to the Deserialization exception handlers and deprecate the previous handle signature. Co-authored-by: Dabz <d.gasparina@gmail.com> Co-authored-by: loicgreffier <loic.greffier@michelin.com> Reviewers: Bruno Cadonna <bruno@confluent.io>, Matthias J. Sax <matthias@confluent.io>

loicgreffier mentioned this pull request Jul 19, 2024

KAFKA-16448: Catch and handle processing exceptions #16093

Merged

3 tasks

loicgreffier force-pushed the KAFKA-16448-Deserialization-Add-Error-Handler-Context branch from ade2811 to 0674502 Compare July 22, 2024 07:22

sebastienviale changed the title ~~Kafka 16448 deserialization add error handler context~~ KAFKA-16448: Add ErrorHandlerContext in deserialization exception handler Jul 22, 2024

loicgreffier force-pushed the KAFKA-16448-Deserialization-Add-Error-Handler-Context branch 2 times, most recently from 32c8edf to 8f5decd Compare July 22, 2024 22:27

cadonna reviewed Jul 23, 2024

View reviewed changes

loicgreffier force-pushed the KAFKA-16448-Deserialization-Add-Error-Handler-Context branch 2 times, most recently from 589b42b to 7add8c7 Compare July 23, 2024 15:24

cadonna approved these changes Jul 24, 2024

View reviewed changes

loicgreffier force-pushed the KAFKA-16448-Deserialization-Add-Error-Handler-Context branch from a9d57f8 to c4c781c Compare July 24, 2024 08:23

cadonna added streams kip Requires or implements a KIP labels Jul 24, 2024

mjsax reviewed Jul 26, 2024

View reviewed changes

sebastienviale and others added 4 commits July 29, 2024 16:17

KAFKA-16448 Add Error Handler Context in Deserialization Exception Ha…

8d5d266

…ndler Co-authored-by: Dabz <d.gasparina@gmail.com> Co-authored-by: loicgreffier <loic.greffier@michelin.com>

KAFKA-16448 Add Error Handler Context in Deserialization Exception Ha…

7afd77f

…ndler Co-authored-by: Dabz <d.gasparina@gmail.com> Co-authored-by: loicgreffier <loic.greffier@michelin.com>

KAFKA-16448 Add Error Handler Context in Deserialization Exception Ha…

2ebb123

…ndler remove rawKey and rawValue Co-authored-by: Dabz <d.gasparina@gmail.com> Co-authored-by: loicgreffier <loic.greffier@michelin.com>

KAFKA-16448 Add Error Handler Context in Deserialization Exception Ha…

c5fecd7

…ndler remove rawKey and rawValue Co-authored-by: Dabz <d.gasparina@gmail.com> Co-authored-by: loicgreffier <loic.greffier@michelin.com>

sebastienviale force-pushed the KAFKA-16448-Deserialization-Add-Error-Handler-Context branch from 7ffcc19 to c5fecd7 Compare July 29, 2024 14:17

KAFKA-16448 Add Error Handler Context in Deserialization Exception Ha…

afcbadd

…ndler Co-authored-by: Dabz <d.gasparina@gmail.com> Co-authored-by: loicgreffier <loic.greffier@michelin.com>

mjsax merged commit faaef52 into apache:trunk Jul 30, 2024

loicgreffier mentioned this pull request Jul 30, 2024

KAFKA-16448: Fix raw record not being cached in store #16684

Closed

loicgreffier mentioned this pull request Jul 31, 2024

KAFKA-16448: Unifiy deserialization failure handling #16746

Closed

	public void shouldThrowStreamsExceptionWhenDeserializationFailsAndDeserializationExceptionHandlerRepliesWithFail(final boolean keyThrowsException,
	public void shouldThrowStreamsExceptionWhenDeserializationFailsAndExceptionHandlerRepliesWithFail(final boolean keyThrowsException,

	public void shouldNotThrowStreamsExceptionWhenDeserializationFailsAndDeserializationExceptionHandlerRepliesWithContinue(final boolean keyThrowsException,
	public void shouldNotThrowStreamsExceptionWhenDeserializationFailsAndExceptionHandlerRepliesWithContinue(final boolean keyThrowsException,

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

Conversation

sebastienviale commented Jun 24, 2024

Uh oh!

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

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!

cadonna left a comment

Choose a reason for hiding this comment

Uh oh!

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

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!

loicgreffier Jul 30, 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!

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!

mjsax commented Jul 30, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

loicgreffier Jul 30, 2024 •

edited

Loading