From a267bfeb8e03c38b70556a5fba7154f5f10add4a Mon Sep 17 00:00:00 2001 From: junaiddshaukat Date: Fri, 14 Aug 2026 19:27:17 +0500 Subject: [PATCH 1/2] [GSoC 2026] Kafka Streams runner: an application for measuring instances 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. --- .../kafka-streams/measurement/build.gradle | 72 +++++ .../measurement/docker-compose.yml | 27 ++ .../measurement/RescalingMeasurement.java | 259 ++++++++++++++++++ .../streams/measurement/package-info.java | 26 ++ settings.gradle.kts | 1 + 5 files changed, 385 insertions(+) create mode 100644 runners/kafka-streams/measurement/build.gradle create mode 100644 runners/kafka-streams/measurement/docker-compose.yml create mode 100644 runners/kafka-streams/measurement/src/main/java/org/apache/beam/runners/kafka/streams/measurement/RescalingMeasurement.java create mode 100644 runners/kafka-streams/measurement/src/main/java/org/apache/beam/runners/kafka/streams/measurement/package-info.java diff --git a/runners/kafka-streams/measurement/build.gradle b/runners/kafka-streams/measurement/build.gradle new file mode 100644 index 000000000000..5e094d94f5f4 --- /dev/null +++ b/runners/kafka-streams/measurement/build.gradle @@ -0,0 +1,72 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * License); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an AS IS BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/** + * An application for measuring the Kafka Streams runner's behaviour when instances come and go. + * + * Not part of the build's verification: it is something a person runs against a Kafka, several + * copies at once, and watches. + */ + +apply plugin: 'org.apache.beam.module' +apply plugin: 'application' +mainClassName = "org.apache.beam.runners.kafka.streams.measurement.RescalingMeasurement" + +applyJavaNature( + automaticModuleName: 'org.apache.beam.runners.kafka.streams.measurement', + publish: false, + exportJavadoc: false, +) + +description = "Apache Beam :: Runners :: Kafka Streams :: Measurement" + +def kafkaStreamsRunnerProject = ":runners:kafka-streams" + +evaluationDependsOn(kafkaStreamsRunnerProject) + +// Same pin as the runner and the job server: applyJavaNature forces the versions in library.java, +// which includes an older kafka-clients than the runner is compiled against. +def kafka_version = project(kafkaStreamsRunnerProject).kafka_version + +configurations.configureEach { + resolutionStrategy.eachDependency { details -> + if (details.requested.group == "org.apache.kafka") { + details.useVersion(kafka_version) + details.because("Kafka Streams runner is developed against Kafka ${kafka_version}.") + } + } +} + +dependencies { + implementation project(kafkaStreamsRunnerProject) + implementation project(path: ":sdks:java:core", configuration: "shadow") + implementation project(path: ":model:pipeline", configuration: "shadow") + implementation project(":runners:java-fn-execution") + // On the compile classpath to resolve PortablePipelineRunner, which KafkaStreamsPipelineRunner + // implements; no class of it is named here, so the dependency analysis does not see it used. + implementation project(":runners:java-job-service") + permitUnusedDeclared project(":runners:java-job-service") + implementation project(":runners:core-java") + permitUnusedDeclared project(":runners:core-java") + // The pipeline's own code runs in this process, so the Java SDK harness has to be present. + runtimeOnly project(":sdks:java:harness") + implementation library.java.joda_time + // Without a binding the application starts and says nothing, which is unhelpful for something + // whose whole purpose is to be watched while it runs. + runtimeOnly library.java.slf4j_simple +} diff --git a/runners/kafka-streams/measurement/docker-compose.yml b/runners/kafka-streams/measurement/docker-compose.yml new file mode 100644 index 000000000000..836f121ca331 --- /dev/null +++ b/runners/kafka-streams/measurement/docker-compose.yml @@ -0,0 +1,27 @@ +# One Kafka for the measurement application. One broker is enough: what gets run several times is +# the runner instance, not the broker. +# +# docker compose -f runners/kafka-streams/measurement/docker-compose.yml up -d +# +# group.min.session.timeout.ms is lowered because a broker refuses a session timeout below it, and +# how quickly the group notices a departed instance is the floor on how quickly its work moves. The +# default of 6s would put a floor under every measurement of recovery. +services: + kafka: + image: apache/kafka:4.0.0 + container_name: ks-measurement-kafka + ports: + - "9092:9092" + environment: + KAFKA_NODE_ID: 1 + KAFKA_PROCESS_ROLES: broker,controller + KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093 + KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 + KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093 + KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER + KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT + KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 + KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 + KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 + KAFKA_GROUP_MIN_SESSION_TIMEOUT_MS: 1000 + KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 diff --git a/runners/kafka-streams/measurement/src/main/java/org/apache/beam/runners/kafka/streams/measurement/RescalingMeasurement.java b/runners/kafka-streams/measurement/src/main/java/org/apache/beam/runners/kafka/streams/measurement/RescalingMeasurement.java new file mode 100644 index 000000000000..7f6fc4077094 --- /dev/null +++ b/runners/kafka-streams/measurement/src/main/java/org/apache/beam/runners/kafka/streams/measurement/RescalingMeasurement.java @@ -0,0 +1,259 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.runners.kafka.streams.measurement; + +import java.util.concurrent.atomic.AtomicLong; +import org.apache.beam.model.pipeline.v1.RunnerApi; +import org.apache.beam.runners.fnexecution.provisioning.JobInfo; +import org.apache.beam.runners.kafka.streams.KafkaStreamsPipelineOptions; +import org.apache.beam.runners.kafka.streams.KafkaStreamsPipelineRunner; +import org.apache.beam.runners.kafka.streams.KafkaStreamsRunner; +import org.apache.beam.sdk.Pipeline; +import org.apache.beam.sdk.coders.KvCoder; +import org.apache.beam.sdk.coders.StringUtf8Coder; +import org.apache.beam.sdk.coders.VarLongCoder; +import org.apache.beam.sdk.io.CountingSource; +import org.apache.beam.sdk.io.Read; +import org.apache.beam.sdk.options.Default; +import org.apache.beam.sdk.options.Description; +import org.apache.beam.sdk.options.PipelineOptionsFactory; +import org.apache.beam.sdk.options.PortablePipelineOptions; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.beam.sdk.transforms.GroupByKey; +import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.transforms.windowing.FixedWindows; +import org.apache.beam.sdk.transforms.windowing.Window; +import org.apache.beam.sdk.util.construction.Environments; +import org.apache.beam.sdk.util.construction.PipelineOptionsTranslation; +import org.apache.beam.sdk.util.construction.PipelineTranslation; +import org.apache.beam.sdk.util.construction.SplittableParDo; +import org.apache.beam.sdk.values.KV; +import org.joda.time.Duration; + +/** + * One instance of a streaming pipeline, run as an ordinary application, for measuring what happens + * when instances are added and removed. + * + *

Run several of these against one Kafka. They share an application id, so Kafka's consumer + * group divides the work between them, and stopping one hands its share to the others. Each prints + * how much it is processing once a second, which is what makes a handover visible: the instance + * that is stopped goes silent, and the others pick its work up. + * + *

This is an application rather than a test on purpose. The numbers only mean something if the + * pipeline is doing a realistic amount of work — a grouping over thousands of keys, fed fast enough + * that every partition has something to do. A pipeline that trickles produces idle partitions, and + * an idle partition holds a watermark back for reasons that have nothing to do with rescaling. + * + * 

+ *   docker compose -f runners/kafka-streams/measurement/docker-compose.yml up -d
+ *   ./gradlew :runners:kafka-streams:measurement:installDist
+ *
+ * + *

Then start two instances, sharing an application id and differing in everything local to the + * instance. Each needs its own {@code --stateDir}: two instances sharing one directory fail with a + * {@code LockException}, because Kafka Streams locks the state it keeps on disk. + * + * 

+ *   BIN=runners/kafka-streams/measurement/build/install/measurement/bin/measurement
+ *   $BIN --applicationId=demo --instanceName=one --stateDir=/tmp/ks-one &
+ *   $BIN --applicationId=demo --instanceName=two --stateDir=/tmp/ks-two &
+ *
+ * + *

The read rate is worth knowing about even though it is defaulted here. Reading far faster than + * the grouping keeps up with does not produce groups sooner, it produces none at all: at the + * runner's own default this pipeline emits one window's worth of groups and then stops, while the + * source goes on reading millions of elements, and at 20000 elements per poll it emits nothing at + * all. Output stopping altogether rather than falling behind gradually is the thing to watch for + * when changing {@code --readMaxElementsPerPoll}. + * + *

To watch a handover, kill the instance that is reading — {@code elements_read} in the output + * says which one that is, since reading concentrates on one instance — and watch {@code + * elements_read} on the other. The delay before it starts climbing is dominated by {@code + * --sessionTimeoutMs}, which is how long the consumer group waits before deciding the instance is + * gone. + */ +public final class RescalingMeasurement { + + /** What this instance has processed, printed once a second. */ + private static final AtomicLong PROCESSED = new AtomicLong(); + + /** + * Elements read from the source, so a stall before the grouping can be told from one after it. + */ + private static final AtomicLong READ = new AtomicLong(); + + private RescalingMeasurement() {} + + /** Whether the command line mentioned an option, so that a default is not applied over it. */ + private static boolean given(String[] args, String name) { + for (String arg : args) { + if (arg.equals("--" + name) || arg.startsWith("--" + name + "=")) { + return true; + } + } + return false; + } + + /** + * Applies the defaults this measurement needs, where they differ from the runner's own. + * + *

The runner's defaults are meant for a pipeline, not for this. Left alone, the source reads + * far faster than the grouping keeps up with, and the result is not output arriving late but + * output stopping: one window's worth of groups is emitted and then nothing, while the source + * goes on reading millions of elements. That reads as a broken runner rather than as a source + * being read too fast, which is the wrong thing for a measurement to suggest when someone runs it + * for the first time. The parallelism is raised for a related reason: a measurement of work + * moving between instances needs more than the single partition the runner defaults to, since + * with one partition there is nothing to divide. + */ + private static void applyMeasurementDefaults(String[] args, MeasurementOptions options) { + if (!given(args, "readMaxElementsPerPoll")) { + options.setReadMaxElementsPerPoll(200); + } + if (!given(args, "internalParallelism")) { + options.setInternalParallelism(3); + } + } + + /** Options of the measurement itself, on top of the runner's own. */ + public interface MeasurementOptions extends KafkaStreamsPipelineOptions { + + @Description("Name for this instance in the output, so several can be told apart.") + @Default.String("instance") + String getInstanceName(); + + void setInstanceName(String instanceName); + + @Description( + "How many distinct keys the grouping runs over. Thousands, so that every partition of the" + + " shuffle has work and no partition sits idle holding a watermark back.") + @Default.Integer(2_000) + int getNumKeys(); + + void setNumKeys(int numKeys); + + @Description("Window size in milliseconds; how often each key's group is emitted.") + @Default.Integer(1_000) + int getWindowMs(); + + void setWindowMs(int windowMs); + } + + /** Spreads elements over many keys so the shuffle divides the work evenly. */ + private static class ToKeyedFn extends DoFn> { + private final int numKeys; + + ToKeyedFn(int numKeys) { + this.numKeys = numKeys; + } + + @ProcessElement + public void processElement(@Element Long value, OutputReceiver> out) { + READ.incrementAndGet(); + out.output(KV.of("key-" + (value % numKeys), value)); + } + } + + /** Counts the groups this instance produced. */ + private static class CountGroupsFn extends DoFn>, Void> { + @ProcessElement + public void processElement() { + PROCESSED.incrementAndGet(); + } + } + + /** Prints throughput once a second, so a handover shows up as a gap and a recovery. */ + private static void reportEverySecond(String instanceName) { + Thread reporter = + new Thread( + () -> { + long previous = 0; + while (!Thread.currentThread().isInterrupted()) { + try { + Thread.sleep(1_000L); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + long total = PROCESSED.get(); + System.out.printf( + "%d %s groups_per_second=%d groups_total=%d elements_read=%d%n", + System.currentTimeMillis(), instanceName, total - previous, total, READ.get()); + previous = total; + } + }, + "measurement-reporter"); + reporter.setDaemon(true); + reporter.start(); + } + + public static void main(String[] args) throws Exception { + PipelineOptionsFactory.register(MeasurementOptions.class); + // Deliberately not withValidation(): that enforces the options a pipeline needs when it is + // submitted to a job server, and --jobEndpoint above all, which means nothing here because this + // application runs the pipeline itself. + MeasurementOptions options = PipelineOptionsFactory.fromArgs(args).as(MeasurementOptions.class); + if (options.getApplicationId() == null || options.getApplicationId().isEmpty()) { + throw new IllegalArgumentException( + "--applicationId is required, and every instance of one measurement must share it: it is" + + " what puts them in the same consumer group and so divides the work between them."); + } + // Pipeline.create needs a runner class even though this application never calls pipeline.run() + // — it builds the pipeline proto and hands it to the runner below itself. + applyMeasurementDefaults(args, options); + options.setRunner(KafkaStreamsRunner.class); + // The user code runs in this same process, so no container or separate worker is needed. + options + .as(PortablePipelineOptions.class) + .setDefaultEnvironmentType(Environments.ENVIRONMENT_EMBEDDED); + + Pipeline pipeline = Pipeline.create(options); + pipeline + .apply("read", Read.from(CountingSource.unbounded())) + .apply("key", ParDo.of(new ToKeyedFn(options.getNumKeys()))) + .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())); + + SplittableParDo.convertReadBasedSplittableDoFnsToPrimitiveReads(pipeline); + RunnerApi.Pipeline proto = PipelineTranslation.toProto(pipeline); + JobInfo jobInfo = + JobInfo.create( + options.getApplicationId(), + options.getJobName(), + "", + PipelineOptionsTranslation.toProto(options)); + + System.out.printf( + "starting %s: application=%s keys=%d parallelism=%d window=%dms session_timeout=%dms" + + " read_per_poll=%d bundle=%d%n", + options.getInstanceName(), + options.getApplicationId(), + options.getNumKeys(), + options.getInternalParallelism(), + options.getWindowMs(), + options.getSessionTimeoutMs(), + options.getReadMaxElementsPerPoll(), + options.getMaxBundleSize()); + reportEverySecond(options.getInstanceName()); + + // Blocks until the instance is stopped; a streaming pipeline has no end of its own. + new KafkaStreamsPipelineRunner(options).run(proto, jobInfo); + } +} diff --git a/runners/kafka-streams/measurement/src/main/java/org/apache/beam/runners/kafka/streams/measurement/package-info.java b/runners/kafka-streams/measurement/src/main/java/org/apache/beam/runners/kafka/streams/measurement/package-info.java new file mode 100644 index 000000000000..41579676fb8d --- /dev/null +++ b/runners/kafka-streams/measurement/src/main/java/org/apache/beam/runners/kafka/streams/measurement/package-info.java @@ -0,0 +1,26 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/** + * An application for measuring the Kafka Streams runner's behaviour when instances come and go. + * + *

Not part of the build's verification: it is something a person runs against a Kafka, several + * copies at once, and watches. See {@link + * org.apache.beam.runners.kafka.streams.measurement.RescalingMeasurement} for how to run it. + */ +package org.apache.beam.runners.kafka.streams.measurement; diff --git a/settings.gradle.kts b/settings.gradle.kts index 050d97dc600e..de1e5ea6533d 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -153,6 +153,7 @@ if (startParameter.projectProperties.containsKey("with-kafka-streams-runner")) { include(":runners:kafka-streams") include(":runners:kafka-streams:proto") include(":runners:kafka-streams:job-server") + include(":runners:kafka-streams:measurement") } include(":runners:local-java") include(":runners:portability:java") From 7dd77e03ff2614819d53922f2597f7e62b92fc3d Mon Sep 17 00:00:00 2001 From: junaiddshaukat Date: Sat, 15 Aug 2026 15:57:31 +0500 Subject: [PATCH 2/2] [GSoC 2026] Kafka Streams runner: count groups in the pipeline rather 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. --- .../kafka-streams/measurement/build.gradle | 6 + .../measurement/RescalingMeasurement.java | 182 +++++++++--------- 2 files changed, 98 insertions(+), 90 deletions(-) diff --git a/runners/kafka-streams/measurement/build.gradle b/runners/kafka-streams/measurement/build.gradle index 5e094d94f5f4..5e7992a0916e 100644 --- a/runners/kafka-streams/measurement/build.gradle +++ b/runners/kafka-streams/measurement/build.gradle @@ -31,6 +31,12 @@ applyJavaNature( automaticModuleName: 'org.apache.beam.runners.kafka.streams.measurement', publish: false, exportJavadoc: false, + // This module runs the pipeline in its own process, so the SDK harness and its dependencies are + // on the classpath, and SpotBugs reports on those rather than on the four classes here — some + // eleven thousand warnings, none of them in this source tree. The same is done in the it/ + // modules, which are on the classpath of what they exercise for the same reason. Checkstyle, + // ErrorProne, spotless and the nullness checker all still run. + enableSpotbugs: false, ) description = "Apache Beam :: Runners :: Kafka Streams :: Measurement" diff --git a/runners/kafka-streams/measurement/src/main/java/org/apache/beam/runners/kafka/streams/measurement/RescalingMeasurement.java b/runners/kafka-streams/measurement/src/main/java/org/apache/beam/runners/kafka/streams/measurement/RescalingMeasurement.java index 7f6fc4077094..3ba9b22ca0ee 100644 --- a/runners/kafka-streams/measurement/src/main/java/org/apache/beam/runners/kafka/streams/measurement/RescalingMeasurement.java +++ b/runners/kafka-streams/measurement/src/main/java/org/apache/beam/runners/kafka/streams/measurement/RescalingMeasurement.java @@ -17,25 +17,22 @@ */ package org.apache.beam.runners.kafka.streams.measurement; -import java.util.concurrent.atomic.AtomicLong; import org.apache.beam.model.pipeline.v1.RunnerApi; import org.apache.beam.runners.fnexecution.provisioning.JobInfo; import org.apache.beam.runners.kafka.streams.KafkaStreamsPipelineOptions; import org.apache.beam.runners.kafka.streams.KafkaStreamsPipelineRunner; import org.apache.beam.runners.kafka.streams.KafkaStreamsRunner; import org.apache.beam.sdk.Pipeline; -import org.apache.beam.sdk.coders.KvCoder; -import org.apache.beam.sdk.coders.StringUtf8Coder; -import org.apache.beam.sdk.coders.VarLongCoder; -import org.apache.beam.sdk.io.CountingSource; -import org.apache.beam.sdk.io.Read; +import org.apache.beam.sdk.io.GenerateSequence; import org.apache.beam.sdk.options.Default; import org.apache.beam.sdk.options.Description; import org.apache.beam.sdk.options.PipelineOptionsFactory; import org.apache.beam.sdk.options.PortablePipelineOptions; +import org.apache.beam.sdk.transforms.Count; import org.apache.beam.sdk.transforms.DoFn; -import org.apache.beam.sdk.transforms.GroupByKey; +import org.apache.beam.sdk.transforms.MapElements; import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.transforms.windowing.FixedWindows; import org.apache.beam.sdk.transforms.windowing.Window; import org.apache.beam.sdk.util.construction.Environments; @@ -43,6 +40,7 @@ import org.apache.beam.sdk.util.construction.PipelineTranslation; import org.apache.beam.sdk.util.construction.SplittableParDo; import org.apache.beam.sdk.values.KV; +import org.apache.beam.sdk.values.TypeDescriptors; import org.joda.time.Duration; /** @@ -50,15 +48,18 @@ * when instances are added and removed. * *

Run several of these against one Kafka. They share an application id, so Kafka's consumer - * group divides the work between them, and stopping one hands its share to the others. Each prints - * how much it is processing once a second, which is what makes a handover visible: the instance - * that is stopped goes silent, and the others pick its work up. + * group divides the work between them, and stopping one hands its share to the others. * *

This is an application rather than a test on purpose. The numbers only mean something if the * pipeline is doing a realistic amount of work — a grouping over thousands of keys, fed fast enough * that every partition has something to do. A pipeline that trickles produces idle partitions, and * an idle partition holds a watermark back for reasons that have nothing to do with rescaling. * + *

The source produces a fixed number of elements per second over a fixed set of keys, so what a + * complete window looks like is known before the run starts: every window should report the same + * number of groups. That is what makes a shortfall legible as a shortfall, rather than as one of + * the many rates a pipeline could happen to be running at. + * * 

  *   docker compose -f runners/kafka-streams/measurement/docker-compose.yml up -d
  *   ./gradlew :runners:kafka-streams:measurement:installDist
@@ -74,29 +75,28 @@
  *   $BIN --applicationId=demo --instanceName=two --stateDir=/tmp/ks-two &
  *
* - *

The read rate is worth knowing about even though it is defaulted here. Reading far faster than - * the grouping keeps up with does not produce groups sooner, it produces none at all: at the - * runner's own default this pipeline emits one window's worth of groups and then stops, while the - * source goes on reading millions of elements, and at 20000 elements per poll it emits nothing at - * all. Output stopping altogether rather than falling behind gradually is the thing to watch for - * when changing {@code --readMaxElementsPerPoll}. + *

The pipeline logs one line per key per window. 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 happens to be doing the work. + * + * 

+ *   <millis> <instance> window_end=<millis> key=<key> count=<n> skew_ms=<n>
+ *
* - *

To watch a handover, kill the instance that is reading — {@code elements_read} in the output - * says which one that is, since reading concentrates on one instance — and watch {@code - * elements_read} on the other. The delay before it starts climbing is dominated by {@code - * --sessionTimeoutMs}, which is how long the consumer group waits before deciding the instance is - * gone. + *

Because the rate and the key space are both fixed, a complete window has one line per key and + * the same count on each, so counting the lines for a window says whether the window was complete. + * + *

{@code skew_ms} is the gap between the window's event time and the wall clock when the group + * came out. It is what falling behind should look like: a pipeline that cannot keep up ought to + * 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. + * + *

To watch a handover, kill one instance and watch the other's lines. The delay before the + * survivor reports the killed instance's share again is dominated by {@code --sessionTimeoutMs}, + * which is how long the consumer group waits before deciding the instance is gone. */ public final class RescalingMeasurement { - /** What this instance has processed, printed once a second. */ - private static final AtomicLong PROCESSED = new AtomicLong(); - - /** - * Elements read from the source, so a stall before the grouping can be told from one after it. - */ - private static final AtomicLong READ = new AtomicLong(); - private RescalingMeasurement() {} /** Whether the command line mentioned an option, so that a default is not applied over it. */ @@ -112,14 +112,11 @@ private static boolean given(String[] args, String name) { /** * Applies the defaults this measurement needs, where they differ from the runner's own. * - *

The runner's defaults are meant for a pipeline, not for this. Left alone, the source reads - * far faster than the grouping keeps up with, and the result is not output arriving late but - * output stopping: one window's worth of groups is emitted and then nothing, while the source - * goes on reading millions of elements. That reads as a broken runner rather than as a source - * being read too fast, which is the wrong thing for a measurement to suggest when someone runs it - * for the first time. The parallelism is raised for a related reason: a measurement of work - * moving between instances needs more than the single partition the runner defaults to, since - * with one partition there is nothing to divide. + *

The runner's defaults are meant for a pipeline, not for this. Left alone, the source is read + * in large enough turns that the read starves the rest of the topology and no groups come out at + * all. The parallelism is raised for a related reason: a measurement of work moving between + * instances needs more than the single partition the runner defaults to, since with one partition + * there is nothing to divide. */ private static void applyMeasurementDefaults(String[] args, MeasurementOptions options) { if (!given(args, "readMaxElementsPerPoll")) { @@ -141,67 +138,59 @@ public interface MeasurementOptions extends KafkaStreamsPipelineOptions { @Description( "How many distinct keys the grouping runs over. Thousands, so that every partition of the" - + " shuffle has work and no partition sits idle holding a watermark back.") + + " shuffle has work and no partition sits idle holding a watermark back. With a" + + " window long enough to contain them all, this is also how many groups a complete" + + " window has.") @Default.Integer(2_000) int getNumKeys(); void setNumKeys(int numKeys); - @Description("Window size in milliseconds; how often each key's group is emitted.") + @Description( + "How many elements the source produces per second. Fixed rather than as-fast-as-possible so" + + " that a window's contents are known in advance and a shortfall is visible.") + @Default.Integer(20_000) + int getElementsPerSecond(); + + void setElementsPerSecond(int elementsPerSecond); + + @Description("Window size in milliseconds; how often the groups are counted and reported.") @Default.Integer(1_000) int getWindowMs(); void setWindowMs(int windowMs); } - /** Spreads elements over many keys so the shuffle divides the work evenly. */ - private static class ToKeyedFn extends DoFn> { - private final int numKeys; + /** + * Logs each group the pipeline produces, with how far behind the wall clock its window was. + * + *

One line per key per window. With a fixed rate over a fixed key space every window holds the + * same groups, so counting the lines for a window says whether the window was complete, and no + * counter has to be kept anywhere for that to be true — the count is the pipeline's own output + * rather than a tally maintained beside it, which is what makes it independent of how many + * instances are running. + * + *

The skew is the point of the timestamp. A pipeline that cannot keep up should report its + * groups later and later rather than stop reporting them, so a skew that climbs while the groups + * stay complete is the pipeline falling behind, and groups going missing is something else. + */ + private static class ReportGroupFn extends DoFn, Void> { + private final String instanceName; - ToKeyedFn(int numKeys) { - this.numKeys = numKeys; + ReportGroupFn(String instanceName) { + this.instanceName = instanceName; } @ProcessElement - public void processElement(@Element Long value, OutputReceiver> out) { - READ.incrementAndGet(); - out.output(KV.of("key-" + (value % numKeys), value)); + public void processElement(@Element KV group, BoundedWindow window) { + long windowEnd = window.maxTimestamp().getMillis(); + long now = System.currentTimeMillis(); + System.out.printf( + "%d %s window_end=%d key=%s count=%d skew_ms=%d%n", + now, instanceName, windowEnd, group.getKey(), group.getValue(), now - windowEnd); } } - /** Counts the groups this instance produced. */ - private static class CountGroupsFn extends DoFn>, Void> { - @ProcessElement - public void processElement() { - PROCESSED.incrementAndGet(); - } - } - - /** Prints throughput once a second, so a handover shows up as a gap and a recovery. */ - private static void reportEverySecond(String instanceName) { - Thread reporter = - new Thread( - () -> { - long previous = 0; - while (!Thread.currentThread().isInterrupted()) { - try { - Thread.sleep(1_000L); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return; - } - long total = PROCESSED.get(); - System.out.printf( - "%d %s groups_per_second=%d groups_total=%d elements_read=%d%n", - System.currentTimeMillis(), instanceName, total - previous, total, READ.get()); - previous = total; - } - }, - "measurement-reporter"); - reporter.setDaemon(true); - reporter.start(); - } - public static void main(String[] args) throws Exception { PipelineOptionsFactory.register(MeasurementOptions.class); // Deliberately not withValidation(): that enforces the options a pipeline needs when it is @@ -213,23 +202,35 @@ public static void main(String[] args) throws Exception { "--applicationId is required, and every instance of one measurement must share it: it is" + " what puts them in the same consumer group and so divides the work between them."); } + applyMeasurementDefaults(args, options); // Pipeline.create needs a runner class even though this application never calls pipeline.run() // — it builds the pipeline proto and hands it to the runner below itself. - applyMeasurementDefaults(args, options); options.setRunner(KafkaStreamsRunner.class); // The user code runs in this same process, so no container or separate worker is needed. options .as(PortablePipelineOptions.class) .setDefaultEnvironmentType(Environments.ENVIRONMENT_EMBEDDED); + // A window holds every key as long as it is long enough for the rate to reach them all; below + // that the source has not got round to each key once and the window is short by construction. + long elementsPerWindow = (long) options.getElementsPerSecond() * options.getWindowMs() / 1_000L; + long expectedGroups = Math.min(options.getNumKeys(), elementsPerWindow); + + int numKeys = options.getNumKeys(); Pipeline pipeline = Pipeline.create(options); pipeline - .apply("read", Read.from(CountingSource.unbounded())) - .apply("key", ParDo.of(new ToKeyedFn(options.getNumKeys()))) - .setCoder(KvCoder.of(StringUtf8Coder.of(), VarLongCoder.of())) + .apply( + "read", + GenerateSequence.from(0) + .withRate(options.getElementsPerSecond(), Duration.standardSeconds(1))) + .apply( + "key", + // numKeys is read here rather than inside the lambda: reaching for it through options + // would capture the PipelineOptions in the transform, which cannot be serialized. + MapElements.into(TypeDescriptors.strings()).via((Long n) -> "key-" + (n % numKeys))) .apply("window", Window.into(FixedWindows.of(Duration.millis(options.getWindowMs())))) - .apply("group", GroupByKey.create()) - .apply("count", ParDo.of(new CountGroupsFn())); + .apply("countPerKey", Count.perElement()) + .apply("report", ParDo.of(new ReportGroupFn(options.getInstanceName()))); SplittableParDo.convertReadBasedSplittableDoFnsToPrimitiveReads(pipeline); RunnerApi.Pipeline proto = PipelineTranslation.toProto(pipeline); @@ -241,17 +242,18 @@ public static void main(String[] args) throws Exception { PipelineOptionsTranslation.toProto(options)); System.out.printf( - "starting %s: application=%s keys=%d parallelism=%d window=%dms session_timeout=%dms" - + " read_per_poll=%d bundle=%d%n", + "starting %s: application=%s keys=%d rate=%d/s parallelism=%d window=%dms" + + " session_timeout=%dms read_per_poll=%d bundle=%d expected_groups_per_window=%d%n", options.getInstanceName(), options.getApplicationId(), options.getNumKeys(), + options.getElementsPerSecond(), options.getInternalParallelism(), options.getWindowMs(), options.getSessionTimeoutMs(), options.getReadMaxElementsPerPoll(), - options.getMaxBundleSize()); - reportEverySecond(options.getInstanceName()); + options.getMaxBundleSize(), + expectedGroups); // Blocks until the instance is stopped; a streaming pipeline has no end of its own. new KafkaStreamsPipelineRunner(options).run(proto, jobInfo);