junaiddshaukat commented on code in PR #39578:
URL: https://github.com/apache/beam/pull/39578#discussion_r3702097246
##########
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:
Yes, the time-based close will be its own PR. I pulled it out of this one
because it duplicated output on a real broker and I didn't want to ship a
mechanism whose failure mode I couldn't explain. Still hoping for your read on
whether a state store's flush() as a pre-commit hook is the right shape for it.
And you're right about the key, it should be null. Today Impulse and Read
emit new byte[0] as a placeholder and every stage just carries that along,
which isn't the same thing at all: an empty array is a real key, so murmur2
hashes it to one fixed partition, whereas a null key is the one Kafka spreads.
So a keyless record reaching a sink would pile onto a single partition instead
of being distributed — which is exactly the wrong behaviour for the rescaling
numbers we want.
It's masked at the moment because the only sink we have is the GroupByKey
repartition topic, and data reaching it goes through ShuffleByKeyProcessor,
which overwrites the key with the encoded Beam key. So the placeholder never
actually reaches a partitioner as a data key today. It would start to matter as
soon as a stage output goes to a topic directly.
One thing I'd want to fix as part of that change: in
GroupByKeyBroadcastPartitioner, partition() null-checks the key but
partitions() — the one Kafka Streams actually calls — does murmur2(key)
unguarded, so a null key would NPE there today.
I'd rather do that as a separate PR than widen this one, since it touches
Impulse, Read and the partitioner rather than bundling. Shall I file an issue
and pick it up after the time bound, or would you prefer it folded in here?
--
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]
