muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1735795478
##########
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:
Thought about it, but missed. Now updated with
ProductionExceptionHandlerMock, and also had to add a no args constructor.
--
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
