mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1735367470
##########
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);
+
+ try (LogCaptureAppender streamsConfigLogs =
LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
+ streamsConfigLogs.setClassLogger(StreamsConfig.class, Level.WARN);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(DefaultProductionExceptionHandler.class,
streamsConfig.productionExceptionHandler().getClass());
+
+ final long warningMessageWhenBothConfigsAreSet =
streamsConfigLogs.getMessages().stream()
+ .filter(m -> m.contains("Both the deprecated and new config
for production exception handler are configured."))
+ .count();
+ assertEquals(1, warningMessageWhenBothConfigsAreSet);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void shouldUseOldProductionExceptionHandlerWhenOnlyOldConfigIsSet()
{
+
props.put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG,
DefaultProductionExceptionHandler.class);
Review Comment:
as above
--
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
