Uh oh!
There was an error while loading. Please reload this page.
[GSoC 2026] Kafka Streams runner: an application for measuring instances coming and going - #39752
Conversation
7784f20 to
f7f487cCompareAssigning reviewers: R: @shunping added as fallback since no labels match configuration Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
je-ik
left a comment
There was a problem hiding this comment.
I'd try to convert the local counters to a beam-native aggregation mechanism that works independently on the number of workers.
| .setCoder(KvCoder.of(StringUtf8Coder.of(), VarLongCoder.of())) | ||
| .apply("window", Window.into(FixedWindows.of(Duration.millis(options.getWindowMs())))) | ||
| .apply("group", GroupByKey.create()) | ||
| .apply("count", ParDo.of(new CountGroupsFn())); |
There was a problem hiding this comment.
This counts groups only locally. Can we rather use
MapElements.into()
.via( .. extractthekeyonly ....)
.apply(Window.into(
FixedWindows.of(Duration.millis(1000) /* e.g. */))
.apply(Count.perElement())
.apply(ParDo. .. /* log the output and processing time vs event time skew */);?
There was a problem hiding this comment.
Done, and following your sketch as written. Count.perElement() over the windowed keys, then a ParDo logging each group with the skew. No counters in the application at all now.
Worth flagging what I got wrong first: I put a Combine.globally after Count.perElement() to get one line per window instead of one per key. That funnels everything onto a single key, so one partition of three had all the data and the other two were idle, and the run intermittently produced nothing at all — 40 windows on one attempt and 0 on the next with identical settings. Your chain keeps every partition fed and both runs since have been complete.
The source now runs at a fixed rate over a fixed key space, so a complete window is one line per key with the same count on each. At the defaults that is 2000 keys at 10 elements, every window.
| .apply("group", GroupByKey.create()) | ||
| .apply("count", ParDo.of(new CountGroupsFn())); | ||
| SplittableParDo.convertReadBasedSplittableDoFnsToPrimitiveReads(pipeline); |
There was a problem hiding this comment.
Hm, looks like we should do this inside the runner, because we don't support SDF.
There was a problem hiding this comment.
Agreed it belongs in the runner. I don't think I can do it where this line sits, though: convertReadBasedSplittableDoFnsToPrimitiveReads takes a Pipeline, and by the time KafkaStreamsPipelineRunner.run has it the pipeline is already a proto built on the client. So it would need either a proto-level override rewriting the SDF read back to a primitive one, or use_deprecated_read required of the client, which is a contract rather than a fix.
This application can call it because it builds the pipeline in its own process, but a Python pipeline through the job server can't, which is the case that actually matters. Shall I file it and do it separately? I didn't want to put a half-thought-through runner change in this PR.
There was a problem hiding this comment.
Can we enforce --use_deprecated_read in the runner's wrapper (for Java)?
There was a problem hiding this comment.
Done in #39766.
I should correct what I said earlier on this thread. I claimed the runner could not do the conversion because it receives a proto — that is true of KafkaStreamsPipelineRunner, on the job server side, but not of KafkaStreamsRunner, the Java wrapper you were pointing at. That one gets a real Pipeline, and the conversion is a replaceAll over it, so it works exactly where you suggested.
Being precise about the two halves, because they are not equally load-bearing: Beam already converts unless a pipeline asked for splittable reads, so the change that matters is that the runner calls the conversion at all. The experiment covers the case where a pipeline asks for use_sdf_read, which this runner cannot honour. There is a test for each and each fails if its half is removed.
This application still calls the conversion itself, because it hands a proto to KafkaStreamsPipelineRunner directly rather than going through the wrapper.
fda3e56 to
fbf8eb9Compare…ces coming and going Runs one instance of a windowed grouping pipeline and prints what it is reading and producing once a second. Several run against one Kafka under a shared application id, so the consumer group divides the work between them and stopping one hands its share to the others. An application rather than a test: the question is how long a handover takes and what throughput does across it, which is a thing you run and watch. It is not wired into any build task. Two counters rather than one, because a stall before the shuffle and a stall after it look identical if you only count output.
… than beside it Follows the review: the groups in a window are now the pipeline's own output, counted per key by Count.perElement rather than tallied in a local counter, so the number does not depend on how many instances are running. The source produces a fixed rate over a fixed key space, so a complete window is known before the run starts and a shortfall is legible as one. Each group is logged with the gap between its window's event time and the wall clock, which is what falling behind should look like: groups arriving later while still all arriving. SpotBugs is turned off for this module. It runs the pipeline in process, so the SDK harness and its dependencies are on the classpath and SpotBugs reports on those instead of on the four classes here. The it/ modules do the same for the same reason.
fbf8eb9 to
7dd77e0Compare10ff557
into
apache:feat/18479-kafka-streams-runner-skeletonUh oh!
There was an error while loading. Please reload this page.
Summary
Part of #18479. An application for measuring what the runner does when instances come and go.
This is an application rather than a test, following the review on #39745. A test asserts something and passes; the question here is how long a handover takes and what throughput does across it, and that is a thing you run against a real Kafka with several instances and watch. It is not wired into any build task.
What it runs
A fixed rate over a fixed key space, into fixed windows and a per-key count:
Both halves of that matter for the measurement. The rate and the key space are fixed, so a complete window is known before the run starts — one line per key, the same count on each — and a shortfall is legible as a shortfall rather than as one of the many rates a pipeline might happen to be running at. And nothing is counted beside the pipeline: the groups in a window are its own output, so the tally does not depend on how many instances are running or on which of them is doing the work.
skew_msis the gap between the window's event time and the wall clock when the group came out. A pipeline that cannot keep up should report its groups later and later while still reporting all of them, so a climbing skew with complete windows is congestion, and missing groups are something else.Running it
One broker is enough — what gets run several times is the runner instance, not the broker.
The instances share an application id, so Kafka's consumer group divides the work between them. Each needs its own
--stateDir; sharing one fails with aLockException.At the defaults — 20000 elements/sec over 2000 keys into 1s windows — every window reports 2000 distinct keys at 10 elements each, which is what says the window was complete.
What it measures
Killing an instance with
kill -9, so there is no gracefulLeaveGroup, and timing how long before the survivor takes the work over:--sessionTimeoutMsHandover is dominated by how long the consumer group takes to notice, which is
session.timeout.ms— the option added in #39748. The rest is the work of moving the load, and it is not constant: about 3s when the group notices after 6s, about 10.7s when it notices after 45s. A longer blind period leaving more to catch up on would explain that, but this does not show it, since the two are varied together here.Those numbers were taken by killing the instance carrying the source read, which is the slower and more interesting case: the source is not split, so the read sits on one instance and has to move wholesale. Reassignment of the grouping tasks alone appears to complete sooner, which the numbers above do not separate out.
These are numbers from a laptop with one broker and two instances, and are meant to show the shape of the thing — detection dominating, the rest following — rather than to be quoted as the runner's performance.
Notes
Reading concentrates on one instance rather than spreading, because the source is not split.
The read rate is worth choosing rather than leaving alone, and the application defaults it below the runner's own. Read in large enough turns, the source starves the rest of the topology: the poll punctuator is scheduled every 50ms, and a turn that takes longer than that is already overdue when it returns, so it fires again immediately and the same thread never gets back to the grouping tasks. Measured on the earlier revision of this application, a turn of 200 elements took 3ms and held the thread 6% of the time; a turn of 5000 took 57ms and held it 89%, and no groups came out at all. That is a count bound where a time bound is wanted, and is worth fixing separately in the runner.