bbejeck commented on code in PR #17710:
URL: https://github.com/apache/kafka/pull/17710#discussion_r1833344761
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -2773,10 +2894,8 @@ public void
punctuateShouldThrowStreamsExceptionWhenProcessingExceptionHandlerRe
when(stateManager.taskId()).thenReturn(taskId);
when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
task = createStatelessTask(createConfig(
- AT_LEAST_ONCE,
"100",
- LogAndFailExceptionHandler.class.getName(),
Review Comment:
Why remove here?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -248,18 +251,54 @@ private static StreamsConfig createConfig(final String
enforcedProcessingValue)
}
private static StreamsConfig createConfig(final String eosConfig, final
String enforcedProcessingValue) {
- return createConfig(eosConfig, enforcedProcessingValue,
LogAndFailExceptionHandler.class.getName(),
LogAndFailProcessingExceptionHandler.class.getName());
+ return createConfig(
+ eosConfig,
+ enforcedProcessingValue,
+ LogAndFailExceptionHandler.class,
+ LogAndFailProcessingExceptionHandler.class,
+ FailOnInvalidTimestamp.class
+ );
+ }
+
+ private static StreamsConfig createConfig(final Class<? extends
DeserializationExceptionHandler> deserializationExceptionHandler) {
+ return createConfig(
+ AT_LEAST_ONCE,
+ "0",
+ deserializationExceptionHandler,
+ LogAndFailProcessingExceptionHandler.class,
+ FailOnInvalidTimestamp.class
+ );
+ }
+
+ private static StreamsConfig createConfigWithTsExtractor(final Class<?
extends TimestampExtractor> timestampExtractor) {
+ return createConfig(
+ AT_LEAST_ONCE,
Review Comment:
Also why creating configs with `AT_LEAST_ONCE` only?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -248,18 +251,54 @@ private static StreamsConfig createConfig(final String
enforcedProcessingValue)
}
private static StreamsConfig createConfig(final String eosConfig, final
String enforcedProcessingValue) {
- return createConfig(eosConfig, enforcedProcessingValue,
LogAndFailExceptionHandler.class.getName(),
LogAndFailProcessingExceptionHandler.class.getName());
+ return createConfig(
+ eosConfig,
+ enforcedProcessingValue,
+ LogAndFailExceptionHandler.class,
+ LogAndFailProcessingExceptionHandler.class,
+ FailOnInvalidTimestamp.class
+ );
+ }
+
+ private static StreamsConfig createConfig(final Class<? extends
DeserializationExceptionHandler> deserializationExceptionHandler) {
+ return createConfig(
+ AT_LEAST_ONCE,
+ "0",
+ deserializationExceptionHandler,
+ LogAndFailProcessingExceptionHandler.class,
+ FailOnInvalidTimestamp.class
+ );
+ }
+
+ private static StreamsConfig createConfigWithTsExtractor(final Class<?
extends TimestampExtractor> timestampExtractor) {
+ return createConfig(
+ AT_LEAST_ONCE,
+ "0",
Review Comment:
same as above
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -2795,10 +2914,8 @@ public void
punctuateShouldThrowStreamsExceptionWhenProcessingExceptionHandlerRe
when(stateManager.taskId()).thenReturn(taskId);
when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
task = createStatelessTask(createConfig(
- AT_LEAST_ONCE,
"100",
- LogAndFailExceptionHandler.class.getName(),
Review Comment:
same here and below
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -248,18 +251,54 @@ private static StreamsConfig createConfig(final String
enforcedProcessingValue)
}
private static StreamsConfig createConfig(final String eosConfig, final
String enforcedProcessingValue) {
- return createConfig(eosConfig, enforcedProcessingValue,
LogAndFailExceptionHandler.class.getName(),
LogAndFailProcessingExceptionHandler.class.getName());
+ return createConfig(
+ eosConfig,
+ enforcedProcessingValue,
+ LogAndFailExceptionHandler.class,
+ LogAndFailProcessingExceptionHandler.class,
+ FailOnInvalidTimestamp.class
+ );
+ }
+
+ private static StreamsConfig createConfig(final Class<? extends
DeserializationExceptionHandler> deserializationExceptionHandler) {
+ return createConfig(
+ AT_LEAST_ONCE,
+ "0",
Review Comment:
What is represented by the `"0"`? Maybe add a named variable to help
grokking what's going on.
--
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
