florin-akermann commented on code in PR #14174:
URL: https://github.com/apache/kafka/pull/14174#discussion_r1352140220
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamJoinWindow.java:
##########
@@ -51,8 +51,8 @@ public void init(final ProcessorContext<K, V> context) {
public void process(final Record<K, V> record) {
// if the key is null, we do not need to put the record into
window store
// since it will never be considered for join operations
+ context().forward(record);
if (record.key() != null) {
- context().forward(record);
// Every record basically starts a new window. We're using a
window store mostly for the retention.
window.put(record.key(), record.value(), record.timestamp());
Review Comment:
Should a null key ever match? I thought/think in this context `null != null`?
Plus, my guess would be that most `WindowStore` implementations throw upon
`null` as a key?
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java:
##########
@@ -124,17 +124,20 @@ public void init(final ProcessorContext<K, VOut> context)
{
@SuppressWarnings("unchecked")
@Override
public void process(final Record<K, V1> record) {
- if (StreamStreamJoinUtil.skipRecord(record, LOG,
droppedRecordsSensor, context())) {
- return;
- }
- boolean needOuterJoin = outer;
-
final long inputRecordTimestamp = record.timestamp();
final long timeFrom = Math.max(0L, inputRecordTimestamp -
joinBeforeMs);
final long timeTo = Math.max(0L, inputRecordTimestamp +
joinAfterMs);
-
sharedTimeTracker.advanceStreamTime(inputRecordTimestamp);
+ if (outer && record.key() == null && record.value() != null) {
+ context().forward(record.withValue(joiner.apply(record.key(),
record.value(), null)));
+ return;
+ } else if (record.key() == null || record.value() == null) {
Review Comment:
agree - adjusted
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/StreamStreamJoinUtil.java:
##########
@@ -39,22 +39,26 @@ public static <KIn, VIn, KOut, VOut> boolean skipRecord(
// furthermore, on left/outer joins 'null' in ValueJoiner#apply()
indicates a missing record --
// thus, to be consistent and to avoid ambiguous null semantics, null
values are ignored
if (record.key() == null || record.value() == null) {
- if (context.recordMetadata().isPresent()) {
- final RecordMetadata recordMetadata =
context.recordMetadata().get();
- logger.warn(
- "Skipping record due to null key or value. "
- + "topic=[{}] partition=[{}] offset=[{}]",
- recordMetadata.topic(), recordMetadata.partition(),
recordMetadata.offset()
- );
- } else {
- logger.warn(
- "Skipping record due to null key or value. Topic,
partition, and offset not known."
- );
- }
- droppedRecordsSensor.record();
+ dropRecord(logger, droppedRecordsSensor, context);
return true;
} else {
return false;
}
}
+
+ public static <KOut, VOut> void dropRecord(final Logger logger, final
Sensor droppedRecordsSensor, final ProcessorContext<KOut, VOut> context) {
Review Comment:
agree - adjusted
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java:
##########
@@ -124,17 +124,20 @@ public void init(final ProcessorContext<K, VOut> context)
{
@SuppressWarnings("unchecked")
@Override
public void process(final Record<K, V1> record) {
- if (StreamStreamJoinUtil.skipRecord(record, LOG,
droppedRecordsSensor, context())) {
- return;
- }
- boolean needOuterJoin = outer;
-
final long inputRecordTimestamp = record.timestamp();
final long timeFrom = Math.max(0L, inputRecordTimestamp -
joinBeforeMs);
final long timeTo = Math.max(0L, inputRecordTimestamp +
joinAfterMs);
-
sharedTimeTracker.advanceStreamTime(inputRecordTimestamp);
+ if (outer && record.key() == null && record.value() != null) {
Review Comment:
I don't quite follow. For inner join we would like to keep the current
behavior.
Which column in
https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Streams+Join+Semantics
does 'inner left' join refere to.
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java:
##########
@@ -351,8 +353,23 @@ private void optimizeTopology(final Properties props) {
LOG.debug("Optimizing the Kafka Streams graph for self-joins");
rewriteSingleStoreSelfJoin(root, new IdentityHashMap<>());
}
+ LOG.debug("Optimizing the Kafka Streams graph for null-key records");
+ rewriteRepartitionNodes();
}
+ private void rewriteRepartitionNodes() {
Review Comment:
This is related to this change:
https://github.com/apache/kafka/pull/14174/commits/d1301b1220a072af198539910b9d87e11706624a
Previously, as an optimization, null key records were filtered out upon
repartitioning.
Now we need null-key records to propagate if there is a left- or outer-join
operator further downstream.
@mjsax hinted in KAFKA-12317 that this optimization can no longer be applied
everywhere.
So now, by default, the optimization is no longer applied.
`rewriteRepartitionNodes` adds this optimization if no left- or outer-join
operation is downstream to the `partitionNode` in question.
In general, a manual repartition is covered by this as well.
Though, now I wonder whether there should be some kind of config/flag
whether you want this optimization to be applied at all.
Because, you could define some topology with left joins etc. expecting
null-key records to propagate all the way through.
However, if you chose to add a manual repartition at the very end of your
topology they would then still be filtered out because no left- or outer-join
operation is downstream of this last repartition.
I guess developers would like to be able to opt out of this optimization in
this case?
--
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
