spena commented on a change in pull request #10917:
URL: https://github.com/apache/kafka/pull/10917#discussion_r661606393
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java
##########
@@ -96,11 +98,20 @@ public void init(final
org.apache.kafka.streams.processor.ProcessorContext conte
if (enableSpuriousResultFix
&& StreamsConfig.InternalConfig.getBoolean(
- context().appConfigs(),
+ context.appConfigs(),
ENABLE_KSTREAMS_OUTER_JOIN_SPURIOUS_RESULTS_FIX,
true
)) {
outerJoinWindowStore =
outerJoinWindowName.map(context::getStateStore);
+ sharedTimeTracker.nextTimeToEmit =
context.currentSystemTimeMs();
+
+ final Object emitIntervalConfig = context.appConfigs()
+
.get(InternalConfig.EMIT_INTERVAL_MS_KSTREAMS_OUTER_JOIN_SPURIOUS_RESULTS_FIX);
+ if (emitIntervalConfig instanceof Number) {
+ sharedTimeTracker.setEmitInterval(((Number)
emitIntervalConfig).longValue());
+ } else if (emitIntervalConfig instanceof String) {
+ sharedTimeTracker.setEmitInterval(Long.parseLong((String)
emitIntervalConfig));
+ }
Review comment:
Shouldn't be good to move this code inside
`StreamsConfig.InternalConfig`? I did that for the `getBoolean` so I could
re-use it in other places. This is a good candidate for internal configs.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
