mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1735367199
##########
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);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(DefaultProductionExceptionHandler.class,
streamsConfig.productionExceptionHandler().getClass());
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void shouldUseNewProductionExceptionHandlerWhenBothConfigsAreSet() {
+ props.put(StreamsConfig.PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG,
DefaultProductionExceptionHandler.class);
+
props.put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG,
ProductionExceptionHandler.class);
Review Comment:
Passing an interface does not seems to be right? For this case, the config
is ignored as the new config is used, but otherwise, when we try to create an
object we would crash.
--
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
