[GSoC 2026] Kafka Streams runner: an application for measuring instances coming and going

runners/kafka-streams/measurement/build.gradle

@@ -0,0 +1,78 @@
+/*
+ * 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,
+ // 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"
+
+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
+}

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

runners/kafka-streams/measurement/src/main/java/org/apache/beam/runners/kafka/streams/measurement/RescalingMeasurement.java

@@ -0,0 +1,261 @@
+/*
+ * 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 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.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.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;
+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.apache.beam.sdk.values.TypeDescriptors;
+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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <pre>
+ * docker compose -f runners/kafka-streams/measurement/docker-compose.yml up -d
+ * ./gradlew :runners:kafka-streams:measurement:installDist
+ * </pre>
+ *
+ * <p>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.
+ *
+ * <pre>
+ * BIN=runners/kafka-streams/measurement/build/install/measurement/bin/measurement
+ * $BIN --applicationId=demo --instanceName=one --stateDir=/tmp/ks-one &amp;
+ * $BIN --applicationId=demo --instanceName=two --stateDir=/tmp/ks-two &amp;
+ * </pre>
+ *
+ * <p>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.
+ *
+ * <pre>
+ * &lt;millis&gt; &lt;instance&gt; window_end=&lt;millis&gt; key=&lt;key&gt; count=&lt;n&gt; skew_ms=&lt;n&gt;
+ * </pre>
+ *
+ * <p>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.
+ *
+ * <p>{@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.
+ *
+ * <p>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 {
+
+ 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.
+ *
+ * <p>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")) {
+ 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. 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(
+ "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);
+ }
+
+ /**
+ * Logs each group the pipeline produces, with how far behind the wall clock its window was.
+ *
+ * <p>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.
+ *
+ * <p>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<KV<String, Long>, Void> {
+ private final String instanceName;
+
+ ReportGroupFn(String instanceName) {
+ this.instanceName = instanceName;
+ }
+
+ @ProcessElement
+ public void processElement(@Element KV<String, Long> 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);
+ }
+ }
+
+ 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.");
+ }
+ 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.
+ 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",
+ 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("countPerKey", Count.perElement())
+ .apply("report", ParDo.of(new ReportGroupFn(options.getInstanceName())));
+
+ 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 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(),
+ expectedGroups);
+
+ // Blocks until the instance is stopped; a streaming pipeline has no end of its own.
+ new KafkaStreamsPipelineRunner(options).run(proto, jobInfo);
+ }
+}

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.
+ *
+ * <p>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;