junaiddshaukat commented on code in PR #39578:
URL: https://github.com/apache/beam/pull/39578#discussion_r3700100405
##########
runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ExecutableStageProcessor.java:
##########
@@ -253,6 +285,18 @@ private FnDataReceiver<WindowedValue<?>>
mainInputReceiver() {
}
private void closeBundleAndFlush(Record<byte[], KStreamsPayload<?>> record) {
+ byte[] key = record.key();
+ closeBundleAndFlush(key == null ? lastKey : key, record.timestamp());
Review Comment:
You're right, and it's worse than unnecessary standing in a different
record's key there would just be wrong.
That parameter existed only because I had the bundle also being closed by a
wall-clock punctuator, which has no record to take a key from, so it passed the
last key it had seen. I dropped the punctuator from this PR when it turned out
to duplicate output, and I left the substitution behind. It's gone now — the
close takes the record again and carries that record's own key and timestamp,
exactly as before this PR.
And yes, to your question: an ExecutableStage is unkeyed. It runs stateless,
with StateRequestHandler.unsupported() and no timer receivers, so the Kafka
record key means nothing to it and is only carried along. Where the key does
matter, downstream sets it — ShuffleByKeyProcessor derives it from the Beam key
before a GroupByKey. I've added that to the javadoc so the next reader doesn't
have to ask.
--
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]
