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

sebastienviale · 2024-06-24T08:05:58Z

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 Production exception handler and deprecate the previous handle signature

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

Contributors
@Dabz
@sebastienviale
@loicgreffier

cadonna

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

Here my feedback!

cadonna · 2024-07-23T09:36:10Z

 public class DefaultProductionExceptionHandler implements ProductionExceptionHandler {
    @Override
-    public ProductionExceptionHandlerResponse handle(final ProducerRecord<byte[], byte[]> record,
+    public ProductionExceptionHandlerResponse handle(final ErrorHandlerContext context,


I do not think we can modify this without deprecating the old method and mentioning it in the KIP since it is part of the public API.
I am aware that this method is only called internally.

True, we did it for the deserialization exception handler implementations. I'm wondering why we did not do the same for the production.. 😄

@cadonna I updated KIP-1033 with a word regarding deserialization and production handler implementations

cadonna · 2024-07-23T09:40:05Z

+     * @param record the record that failed to serialize
+     * @param exception the exception that occurred during serialization


Could you please revert the formatting changes in this two lines?

cadonna · 2024-07-23T10:34:11Z

+            final ProductionExceptionHandler.SerializationExceptionOrigin origin;
+            if (keyBytes == null) {
+                origin = ProductionExceptionHandler.SerializationExceptionOrigin.KEY;
+            } else {
+                origin = ProductionExceptionHandler.SerializationExceptionOrigin.VALUE;
+            }


I think, it would be better to have something like:

try { keyBytes = keySerializer.serialize(topic, headers, key); } catch (final ClassCastException exception) { handleClassCastException(...) } catch (final Exception exception) { handleException(ProductionExceptionHandler.SerializationExceptionOrigin.KEY, ...) } try { valBytes = valueSerializer.serialize(topic, headers, value); } catch (final ClassCastException exception) { handleClassCastException(...) } catch (final Exception exception) { handleException(ProductionExceptionHandler.SerializationExceptionOrigin.VALUE, ...) }

Done.

I had to remove the final from keyBytes and valueBytes and initialize them when moving the ClassCastException code to a dedicated method

If you want to keep the final and make the code a bit better readable, you could rename handleClassCastException() to createStreamsExceptionForClassCastException() and return the StreamsException from there. Then you can do

} catch (final ClassCastException exception) { throw createStreamsExceptionForClassCastException()( topic, key, value, keySerializer, valueSerializer, exception);

With this it is clear that the catch-branch terminates the method.

The update has been pushed!

cadonna · 2024-07-23T10:41:36Z

+            assertEquals(expectedContext.recordContext().rawRecord().key(), context.sourceRawKey());
+            assertEquals(expectedContext.recordContext().rawRecord().value(), context.sourceRawValue());


I think you need to use assertArrayEquals() here to verify the byte arrays.

cadonna · 2024-07-24T08:51:30Z

    }

+    @Test
+    public void shouldThrowStreamsExceptionWhenValueSerializationFailedAndProductionExceptionHandlerRepliesWithFail() {


I am missing the version of this test where the key serialization fails.

done, unit test added

cadonna

Thanks for the updates!

LGTM!

mjsax · 2024-07-25T23:27:31Z

- * Production exception handler that always instructs streams to continue when an exception
- * happens while attempting to produce result records.
- */
-public class AlwaysContinueProductionExceptionHandler implements ProductionExceptionHandler {


Why does this PR remove this class?

@mjsax

We have created a parameterized ProductionExceptionHandler (ProductionExceptionHandlerMock) in the RecordCollectorTest class that additionally does some assertions, including the origin of the exception.

Thus AlwaysContinueProductionExceptionHandler was redundant with ProductionExceptionHandlerMock, it has been replaced in this PR while updating and adding some additional unit tests

But AlwaysContinueProductionExceptionHandler is a public built-in handler users can configure... -- it's public API, and thus has nothing to do with ProductionExceptionHandlerMock and our internal testing.

It's a build-in alternative to the default handler.

If we really want to no have it any longer, we would need to deprecate -- but we cannot just remove public API. But I don't see why we would want to deprecate and remove it?

@mjsax Unless I'm wrong, AlwaysContinueProductionExceptionHandler belongs to src/test/java and is only used for testing purpose in the RecordCollectorTest.

Given that it is only use for testing, should it be consider public? If yes, we can obviously restore the class

Ah. You are right. My bad. I guess I mixed it up with the two built-in handler for DeserializationExceptionHandler, but we only have the "alway fail" build-in handler for the production case...

mjsax · 2024-07-26T18:25:32Z

     *
     * @param record The record that failed to produce
     * @param exception The exception that occurred during production
+     * @deprecated Use {@link #handle(ErrorHandlerContext, ProducerRecord, Exception)} instead


Suggested change

* @deprecated Use {@link #handle(ErrorHandlerContext, ProducerRecord, Exception)} instead

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

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

     *
     * @param record        the record that failed to serialize
     * @param exception     the exception that occurred during serialization
+     * @deprecated Use {@link #handle(ErrorHandlerContext, ProducerRecord, Exception)} instead


Suggested change

* @deprecated Use {@link #handle(ErrorHandlerContext, ProducerRecord, Exception)} instead

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

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

+    @Deprecated
+    default ProductionExceptionHandlerResponse handle(final ProducerRecord<byte[], byte[]> record,
+                                                      final Exception exception) {
+        throw new UnsupportedOperationException();


KIP says NotImplementedException -- we should update the KIP -- same for the other method below.

KIP has been updated

mjsax · 2024-07-26T18:29:00Z

+        droppedRecordsSensor.record();
+    }
+
+    private <K, V> StreamsException createStreamsExceptionForClassCastException(final String topic,


Given that we introduce a helper and have two dedicated try-catch blocks for key and value now, it seems we can tailor this method to key or value similar to what we do for handleException instead of this generic one covering both at once?

done, I hope this what you are waiting !

Yes, and I merged this PR to make progress. However, I think, we can unify both methods into a single one, passing in "key" or "value" as parameter to avoid code duplication (similar to the "origin" parameter use in handleException(...))

The generic type would could be updated to <KV> for "key or value type" or just to <T> if you prefer.

@mjsax Addressing it in #16736

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

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

mjsax · 2024-07-30T03:21:39Z

+
+        droppedRecordsSensor.record();
+    }
+    private <K> StreamsException createStreamsExceptionForKeyClassCastException(final String topic,


Nit: missing newline

@mjsax Addressing it in #16736

…16433) This PR is part of KIP-1033 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 Production exception handler 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:38Z

Thanks for the PR.

Merged to trunk and cherry-picked to 3.9 branch.

…pache#16433) This PR is part of KIP-1033 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 Production exception handler 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-Production-Add-Error-Handler-Context branch from f512db8 to c60b79e Compare July 22, 2024 07:23

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

loicgreffier force-pushed the KAFKA-16448-Production-Add-Error-Handler-Context branch from 36098ac to 3fe0f59 Compare July 22, 2024 22:28

cadonna reviewed Jul 23, 2024

View reviewed changes

loicgreffier force-pushed the KAFKA-16448-Production-Add-Error-Handler-Context branch 2 times, most recently from d6dc8f1 to 162d75b Compare July 23, 2024 15:03

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

loicgreffier force-pushed the KAFKA-16448-Production-Add-Error-Handler-Context branch from 94edf82 to 697e0b7 Compare July 24, 2024 08:43

cadonna reviewed Jul 24, 2024

View reviewed changes

cadonna approved these changes Jul 24, 2024

View reviewed changes

loicgreffier mentioned this pull request Jul 25, 2024

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

Closed

mjsax reviewed Jul 25, 2024

View reviewed changes

mjsax reviewed Jul 26, 2024

View reviewed changes

sebastienviale and others added 7 commits July 29, 2024 16:12

KAFKA-16448 Add Error Handler Context for Production Exception Handler

4f367f7

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

KAFKA-16448 Add Error Handler Context for Production Exception Handler

94242d8

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

KAFKA-16448 Add Error Handler Context for Production Exception Handler

1fd0a66

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

KAFKA-16448 Add Error Handler Context for Production Exception Handler

057464f

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

KAFKA-16448 Add Error Handler Context for Production Exception Handle…

1efb3b2

…r 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 for Production Exception Handle…

419155c

…r 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 for Production Exception Handle…

902ae98

…r 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-Production-Add-Error-Handler-Context branch from 42839a4 to 902ae98 Compare July 29, 2024 14:14

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

mjsax reviewed Jul 30, 2024

View reviewed changes

		* @param record the record that failed to serialize
		* @param exception the exception that occurred during serialization

		assertEquals(expectedContext.recordContext().rawRecord().key(), context.sourceRawKey());
		assertEquals(expectedContext.recordContext().rawRecord().value(), context.sourceRawValue());

	* @deprecated Use {@link #handle(ErrorHandlerContext, ProducerRecord, Exception)} instead
	* @deprecated Since 3.9. Use {@link #handle(ErrorHandlerContext, ProducerRecord, 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!

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!

mjsax commented Jul 30, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants