You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "florin-akermann (via GitHub)" <gi...@apache.org> on 2023/08/09 12:22:27 UTC
[GitHub] [kafka] florin-akermann opened a new pull request, #14174: Kafka 12317: Relax non-null key requirement in Kafka Streams
florin-akermann opened a new pull request, #14174:
URL: https://github.com/apache/kafka/pull/14174
https://cwiki.apache.org/confluence/display/KAFKA/KIP-962%3A+Relax+non-null+key+requirement+in+Kafka+Streams
Test Strategy:
Basic tests on minimal topologies with call to '.repartitioned()' prepended.
TODO: expand test cases - assert correct behavior for self joins, ...
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka 12317: Relax non-null key requirement in Kafka Streams [kafka]
Posted by "florin-akermann (via GitHub)" <gi...@apache.org>.
florin-akermann commented on PR #14174:
URL: https://github.com/apache/kafka/pull/14174#issuecomment-1793747649
@wcarlson5 thanks for the merge. I think it would be good to tackle https://github.com/apache/kafka/pull/14107 in the same release.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka 12317: Relax non-null key requirement in Kafka Streams [kafka]
Posted by "florin-akermann (via GitHub)" <gi...@apache.org>.
florin-akermann commented on code in PR #14174:
URL: https://github.com/apache/kafka/pull/14174#discussion_r1372233874
##########
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:
Ok, so basically we remove this optimization completley for now?
Developers could still just filter out null keys with a 'filter' operator to achieve the old behavior.
And then we make a separate ticket where developers can opt in to this optimization?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka 12317: Relax non-null key requirement in Kafka Streams [kafka]
Posted by "florin-akermann (via GitHub)" <gi...@apache.org>.
florin-akermann commented on code in PR #14174:
URL: https://github.com/apache/kafka/pull/14174#discussion_r1352231502
##########
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?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka 12317: Relax non-null key requirement in Kafka Streams [kafka]
Posted by "florin-akermann (via GitHub)" <gi...@apache.org>.
florin-akermann commented on PR #14174:
URL: https://github.com/apache/kafka/pull/14174#issuecomment-1749481836
Hey @wcarlson5
Great! Yes I am willing to push it.
Looking forward to your feedback.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka 12317: Relax non-null key requirement in Kafka Streams [kafka]
Posted by "wcarlson5 (via GitHub)" <gi...@apache.org>.
wcarlson5 merged PR #14174:
URL: https://github.com/apache/kafka/pull/14174
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka 12317: Relax non-null key requirement in Kafka Streams [kafka]
Posted by "wcarlson5 (via GitHub)" <gi...@apache.org>.
wcarlson5 commented on code in PR #14174:
URL: https://github.com/apache/kafka/pull/14174#discussion_r1366083940
##########
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:
That's likely true about the windowstore implemntations. Something to think about for later then. We should maybe clarify the semantics about null matching in the docs somewhere.
##########
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:
Hmm, I would think they would like to opt out. That would require an update to the kip. Maybe even a revote. I'm not sure what the odds are that someone manually repartitioning would be needing the null-keys to propagate. But its probably higher than you would think as manual repartitioner as are typically power-users.
I don't think we need to make it optional as we already filter all null keys and now we let some propagate. Maybe we should just make a ticket and we can come back to it. Being able to toggle the optimization should be pretty simple.
##########
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:
ah nvn, ignore this
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka 12317: Relax non-null key requirement in Kafka Streams [kafka]
Posted by "wcarlson5 (via GitHub)" <gi...@apache.org>.
wcarlson5 commented on PR #14174:
URL: https://github.com/apache/kafka/pull/14174#issuecomment-1749460142
Hey @florin-akermann I can take a look at this soon. I'll probably be able to take a first look next week if you are still willing to push this
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] florin-akermann commented on pull request #14174: Kafka 12317: Relax non-null key requirement in Kafka Streams
Posted by "florin-akermann (via GitHub)" <gi...@apache.org>.
florin-akermann commented on PR #14174:
URL: https://github.com/apache/kafka/pull/14174#issuecomment-1684251499
@mjsax @guozhangwang I'd say this PR is ready to be reviewd. May I ask for your input or do you know who to poke?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka 12317: Relax non-null key requirement in Kafka Streams [kafka]
Posted by "florin-akermann (via GitHub)" <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Kafka 12317: Relax non-null key requirement in Kafka Streams [kafka]
Posted by "wcarlson5 (via GitHub)" <gi...@apache.org>.
wcarlson5 commented on code in PR #14174:
URL: https://github.com/apache/kafka/pull/14174#discussion_r1347987570
##########
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:
why should null keys not enter the window?
##########
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:
I'm not a huge fan of splitting this out to a separate public method. I think you can just reuse the logic in skip record.
##########
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:
what about inner left joins? Those values go into the window? Why?
##########
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 to prevent null keys to go into reparation topics, right? Will that effect results if a manual reparation is added?
##########
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:
This is the check already in skip record, can you use that instead?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org