mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1735366368
##########
streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java:
##########
@@ -1622,6 +1626,72 @@ public void testInvalidProcessingExceptionHandler() {
);
}
+ @Test
+ public void
shouldSetAndGetDeserializationExceptionHandlerWhenOnlyNewConfigIsSet() {
+
props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
LogAndContinueExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class,
streamsConfig.deserializationExceptionHandler().getClass());
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void
shouldUseNewDeserializationExceptionHandlerWhenBothConfigsAreSet() {
+
props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
LogAndContinueExceptionHandler.class);
+
props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
LogAndFailExceptionHandler.class);
+
+ try (LogCaptureAppender streamsConfigLogs =
LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
+ streamsConfigLogs.setClassLogger(StreamsConfig.class, Level.WARN);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class,
streamsConfig.deserializationExceptionHandler().getClass());
+
+ final long warningMessageWhenBothConfigsAreSet =
streamsConfigLogs.getMessages().stream()
+ .filter(m -> m.contains("Both the deprecated and new config
for deserialization exception handler are configured."))
+ .count();
+ assertEquals(1, warningMessageWhenBothConfigsAreSet);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void
shouldUseOldDeserializationExceptionHandlerWhenOnlyOldConfigIsSet() {
+
props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
LogAndContinueExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class,
streamsConfig.deserializationExceptionHandler().getClass());
+ }
+
+ @Test
+ public void
shouldSetAndGetProductionExceptionHandlerWhenOnlyNewConfigIsSet() {
+ props.put(StreamsConfig.PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG,
DefaultProductionExceptionHandler.class);
Review Comment:
Well, given that the default is well, the default, setting the default does
not give us any signal if it was picked up... Maybe use
`ProductionExceptionHandlerMock` or implement a new dummy handler for this test
itself.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
