mjsax commented on code in PR #19303:
URL: https://github.com/apache/kafka/pull/19303#discussion_r2029558102
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionSendProcessorSupplierTest.java:
##########
@@ -316,6 +367,21 @@ public void innerJoinShouldPropagateDeletionOfPrimaryKey()
{
);
}
+ @Test
+ public void
innerJoinShouldNotPropagateDeletionOfPrimaryKeyWhenPreviousFKIsNull() {
+ final MockInternalProcessorContext<String,
SubscriptionWrapper<String>> context = new MockInternalProcessorContext<>();
+ innerJoinProcessor.init(context);
+ context.setRecordMetadata("topic", 0, 0);
+
+ innerJoinProcessor.process(new Record<>(pk, new Change<>(null, new
LeftValue(null)), 0));
+
+ assertThat(context.forwarded(), empty());
+
+ // test dropped-records sensors
+ assertEquals(1.0, getDroppedRecordsTotalMetric(context));
Review Comment:
That's is not really the definition of "dropped records" -- "dropped
records" is a metric for malformed records that could not be processed. -- Eg,
a `null`-key record for a table is not valid, because we need a valid PK, and
thus we "drop" such a record.
However for the case we discuss here, the input record is totally ok, and
it's operator semantics to decide/figure out that the join result is already
correct, and thus we don't need to do anything.
--
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]
