lihaosky commented on code in PR #14426:
URL: https://github.com/apache/kafka/pull/14426#discussion_r1397715022
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java:
##########
@@ -104,20 +103,16 @@ public void init(final ProcessorContext<K, VOut> context)
{
internalProcessorContext = (InternalProcessorContext<K, VOut>)
context;
final StreamsMetricsImpl metrics = (StreamsMetricsImpl)
context.metrics();
- droppedRecordsSensor =
droppedRecordsSensor(Thread.currentThread().getName(),
context.taskId().toString(), metrics);
+ droppedRecordsSensor =
droppedRecordsSensor(Thread.currentThread().getName(),
context.taskId().toString(),
+ metrics);
Review Comment:
ditto
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamLeftJoinTest.java:
##########
@@ -609,17 +842,18 @@ public void testOrdering() {
inputTopic1.pipeInput(1, "A1", 100L);
processor.checkAndClearProcessResult();
- // push one item to the other window that has a join; this should
produce non-joined records with a closed window first, then
- // the joined records
+ // push one item to the other window that has a join;
+ // this should produce the joined record first;
+ // then non-joined record with a closed window
// by the time they were produced before
// w1 = { 0:A0 (ts: 0), 1:A1 (ts: 100) }
// w2 = { }
// --> w1 = { 0:A0 (ts: 0), 1:A1 (ts: 100) }
// --> w2 = { 1:a1 (ts: 110) }
inputTopic2.pipeInput(1, "a1", 110L);
processor.checkAndClearProcessResult(
- new KeyValueTimestamp<>(0, "A0+null", 0L),
- new KeyValueTimestamp<>(1, "A1+a1", 110L)
+ new KeyValueTimestamp<>(1, "A1+a1", 110L),
+ new KeyValueTimestamp<>(0, "A0+null", 0L)
Review Comment:
This is because previously we look at outer store and then join. But this
change make it we join first and then look at outer store. The ts in outer
store and other store is hard to reason. If we change the ts of 0 to be 100 and
ts of 1 to be 50, the original test would still produce 0 first which has
larger ts... So unless we compare the ts of join and outer at the same time
when we output, we can guarantee the order of ts when output.
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java:
##########
@@ -60,24 +61,21 @@ class KStreamKStreamJoin<K, V1, V2, VOut> implements
ProcessorSupplier<K, V1, K,
private final TimeTrackerSupplier sharedTimeTrackerSupplier;
- KStreamKStreamJoin(final boolean isLeftSide,
- final String otherWindowName,
- final JoinWindowsInternal windows,
- final ValueJoinerWithKey<? super K, ? super V1, ? super
V2, ? extends VOut> joiner,
- final boolean outer,
- final Optional<String> outerJoinWindowName,
- final TimeTrackerSupplier sharedTimeTrackerSupplier) {
+ KStreamKStreamJoin(final boolean isLeftSide, final String otherWindowName,
final JoinWindowsInternal windows,
+ final ValueJoinerWithKey<? super K, ? super V1, ? super V2, ?
extends VOut> joiner, final boolean outer,
+ final Optional<String> outerJoinWindowName, final
TimeTrackerSupplier sharedTimeTrackerSupplier) {
Review Comment:
Change this back? I think for Kafka streams, the convention is to align with
first param's indentation
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java:
##########
@@ -165,40 +155,52 @@ public void process(final Record<K, V1> record) {
// problem:
//
// Say we have a window size of 5 seconds
- // 1. A non-joined record with time T10 is seen in the
left-topic (maxLeftStreamTime: 10)
- // The record is not processed yet, and is added to
the outer-join store
- // 2. A non-joined record with time T2 is seen in the
right-topic (maxRightStreamTime: 2)
- // The record is not processed yet, and is added to
the outer-join store
- // 3. A joined record with time T11 is seen in the
left-topic (maxLeftStreamTime: 11)
- // It is time to look at the expired records. T10 and
T2 should be emitted, but
- // because T2 was late, then it is not fetched by the
window store, so it is not processed
+ // 1. A non-joined record with time T10 is seen in the
left-topic
+ // (maxLeftStreamTime: 10)
+ // The record is not processed yet, and is added to the
outer-join store
+ // 2. A non-joined record with time T2 is seen in the
right-topic
+ // (maxRightStreamTime: 2)
+ // The record is not processed yet, and is added to the
outer-join store
+ // 3. A joined record with time T11 is seen in the
left-topic
+ // (maxLeftStreamTime: 11)
+ // It is time to look at the expired records. T10 and T2
should be emitted, but
+ // because T2 was late, then it is not fetched by the
window store, so it is not
+ // processed
//
// See KStreamKStreamLeftJoinTest.testLowerWindowBound()
tests
//
- // This condition below allows us to process the
out-of-order records without the need
+ // This condition below allows us to process the
out-of-order records without
+ // the need
// to hold it in the temporary outer store
if (!outerJoinStore.isPresent() || timeTo <
sharedTimeTracker.streamTime) {
context().forward(record.withValue(joiner.apply(record.key(), record.value(),
null)));
} else {
sharedTimeTracker.updatedMinTime(inputRecordTimestamp);
outerJoinStore.ifPresent(store -> store.put(
- TimestampedKeyAndJoinSide.make(isLeftSide,
record.key(), inputRecordTimestamp),
- LeftOrRightValue.make(isLeftSide,
record.value())));
+ TimestampedKeyAndJoinSide.make(isLeftSide,
record.key(), inputRecordTimestamp),
+ LeftOrRightValue.make(isLeftSide,
record.value())));
Review Comment:
ditto
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java:
##########
@@ -104,20 +103,16 @@ public void init(final ProcessorContext<K, VOut> context)
{
internalProcessorContext = (InternalProcessorContext<K, VOut>)
context;
final StreamsMetricsImpl metrics = (StreamsMetricsImpl)
context.metrics();
- droppedRecordsSensor =
droppedRecordsSensor(Thread.currentThread().getName(),
context.taskId().toString(), metrics);
+ droppedRecordsSensor =
droppedRecordsSensor(Thread.currentThread().getName(),
context.taskId().toString(),
+ metrics);
otherWindowStore = context.getStateStore(otherWindowName);
sharedTimeTracker =
sharedTimeTrackerSupplier.get(context.taskId());
if (enableSpuriousResultFix) {
outerJoinStore =
outerJoinWindowName.map(context::getStateStore);
- sharedTimeTracker.setEmitInterval(
- StreamsConfig.InternalConfig.getLong(
- context.appConfigs(),
-
EMIT_INTERVAL_MS_KSTREAMS_OUTER_JOIN_SPURIOUS_RESULTS_FIX,
- 1000L
- )
- );
+
sharedTimeTracker.setEmitInterval(StreamsConfig.InternalConfig.getLong(context.appConfigs(),
+
EMIT_INTERVAL_MS_KSTREAMS_OUTER_JOIN_SPURIOUS_RESULTS_FIX, 1000L));
Review Comment:
ditto
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java:
##########
@@ -134,29 +129,24 @@ public void process(final Record<K, V1> record) {
final long timeTo = Math.max(0L, inputRecordTimestamp +
joinAfterMs);
sharedTimeTracker.advanceStreamTime(inputRecordTimestamp);
-
- // Emit all non-joined records which window has closed
- if (inputRecordTimestamp == sharedTimeTracker.streamTime) {
- outerJoinStore.ifPresent(store ->
emitNonJoinedOuterRecords(store, record));
- }
try (final WindowStoreIterator<V2> iter =
otherWindowStore.fetch(record.key(), timeFrom, timeTo)) {
while (iter.hasNext()) {
needOuterJoin = false;
final KeyValue<Long, V2> otherRecord = iter.next();
final long otherRecordTimestamp = otherRecord.key;
- outerJoinStore.ifPresent(store -> {
- // use putIfAbsent to first read and see if there's
any values for the key,
- // if yes delete the key, otherwise do not issue a put;
- // we may delete some values with the same key early
but since we are going
- // range over all values of the same key even after
failure, since the other window-store
- // is only cleaned up by stream time, so this is okay
for at-least-once.
-
store.putIfAbsent(TimestampedKeyAndJoinSide.make(!isLeftSide, record.key(),
otherRecordTimestamp), null);
- });
-
- context().forward(
- record.withValue(joiner.apply(record.key(),
record.value(), otherRecord.value))
- .withTimestamp(Math.max(inputRecordTimestamp,
otherRecordTimestamp)));
+ outerJoinStore.ifPresent(store ->
+ // Use putIfAbsent to first read and see if there's any
values for the key,
+ // if yes delete the key, otherwise do not issue a put;
+ // we may delete some values with the same key early but
since we are going
+ // range over all values of the same key even after
failure, since the other
+ // window-store
+ // is only cleaned up by stream time, so this is okay for
at-least-once.
+
store.putIfAbsent(TimestampedKeyAndJoinSide.make(!isLeftSide, record.key(),
otherRecordTimestamp),
+ null));
+
+
context().forward(record.withValue(joiner.apply(record.key(), record.value(),
otherRecord.value))
+ .withTimestamp(Math.max(inputRecordTimestamp,
otherRecordTimestamp)));
Review Comment:
ditto
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java:
##########
@@ -94,7 +92,8 @@ public Processor<K, V1, K, VOut> get() {
private class KStreamKStreamJoinProcessor extends ContextualProcessor<K,
V1, K, VOut> {
private WindowStore<K, V2> otherWindowStore;
private Sensor droppedRecordsSensor;
- private Optional<KeyValueStore<TimestampedKeyAndJoinSide<K>,
LeftOrRightValue<V1, V2>>> outerJoinStore = Optional.empty();
+ private Optional<KeyValueStore<TimestampedKeyAndJoinSide<K>,
LeftOrRightValue<V1, V2>>> outerJoinStore = Optional
+ .empty();
Review Comment:
revert this change? Does it cause checkstyle problem previously?
--
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
