[GSoC 2026] Kafka Streams runner: shorten comments, and fix the license header and Python formatting

runners/kafka-streams/measurement/docker-compose.yml

@@ -1,3 +1,20 @@
+#
+# 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.
+#
+
 # One Kafka for the measurement application. One broker is enough: what gets run several times is
 # the runner instance, not the broker.
 #

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

@@ -45,56 +45,41 @@
 
 /**
  * One instance of a streaming pipeline, run as an ordinary application, for measuring what happens
- * when instances are added and removed.
+ * when instances come and go.
  *
- * <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>Run several against one Kafka. They share an application id, so the consumer group divides the
+ * work between them and stopping one hands its share to the others. It is an application rather
+ * than a test because the numbers only mean something under a realistic load: a grouping over
+ * thousands of keys, fed fast enough that no partition sits idle holding a watermark back.
  *
- * <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.
+ * <p>The source runs at a fixed rate over a fixed key space, so a complete window is known before
+ * the run starts — one line per key, the same count on each — which is what makes a shortfall
+ * legible as one.
  *
  * <pre>
  *   docker compose -f runners/kafka-streams/measurement/docker-compose.yml up -d
- *   ./gradlew :runners:kafka-streams:measurement:installDist
- * </pre>
+ *   ./gradlew -Pwith-kafka-streams-runner :runners:kafka-streams:measurement:installDist
  *
- * <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.
+ * <p>Each instance needs its own {@code --stateDir}; sharing one fails with a {@code
+ * LockException}. Output is one line per key per window, counted by the pipeline itself rather than
+ * beside it, so the tally does not depend on how many instances are running:
  *
  * <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.
+ * <p>{@code skew_ms} is the gap between the window's event time and the wall clock when it came
+ * out. A pipeline that cannot keep up should report its groups later and later while still
+ * reporting all of them, so climbing skew with complete windows is congestion and missing groups
+ * are something else. To watch a handover, kill one instance and watch the other; the delay before
+ * it reports the dead instance's share is dominated by {@code --sessionTimeoutMs}.
  */
+
 public final class RescalingMeasurement {
 
   private RescalingMeasurement() {}

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

@@ -48,17 +48,13 @@ public KafkaStreamsPipelineRunner(KafkaStreamsPipelineOptions pipelineOptions) {
 
   @Override
   public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo) {
-    // Surface a clear error if an option this runner needs is missing, instead of letting
-    // Properties.put fail with a raw NullPointerException further down. Only the options that are
-    // meaningful here are checked, rather than validating the whole interface: this runs on the job
-    // server, executing a pipeline that has already been submitted, so the client-side options
-    // PortablePipelineOptions marks required — jobEndpoint above all — do not apply. Flink's
-    // equivalent PortablePipelineRunner does not validate here either.
+    // Only the options meaningful here are checked, not the whole interface: this runs on the job
+    // server, so the client-side options PortablePipelineOptions marks required — jobEndpoint above
+    // all — do not apply. Flink's PortablePipelineRunner does not validate here either.
     checkRequiredOption("applicationId", pipelineOptions.getApplicationId());
     checkRequiredOption("bootstrapServers", pipelineOptions.getBootstrapServers());
-    // A topic cannot have fewer than one partition, and the value is also the number of watermark
-    // reports a shuffle's consumer waits for, so a non-positive value would leave it waiting
-    // forever rather than failing.
+    // Also the number of watermark reports a shuffle's consumer waits for, so a non-positive value
+    // would leave it waiting forever rather than failing.
     if (pipelineOptions.getInternalParallelism() < 1) {
       throw new IllegalArgumentException(
           "--internalParallelism must be at least 1, but was "
@@ -81,31 +77,26 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
         topology.describe());
 
     KafkaStreams kafkaStreams = new KafkaStreams(topology, streamsConfig(jobInfo));
-    // Kafka Streams reports a failed task by moving the client to ERROR and keeping the exception
-    // to itself, which left a failed job with nothing to say beyond "unknown error". Hold on to the
-    // first failure so this method can rethrow it: the job service turns what run() throws into the
-    // job's error message.
+    // Kafka Streams moves the client to ERROR and keeps the exception to itself, which left failed
+    // jobs saying only "unknown error". Keep the first failure so run() can rethrow it.
     AtomicReference<@Nullable Throwable> failure = new AtomicReference<>();
     kafkaStreams.setUncaughtExceptionHandler(
         throwable -> {
           failure.compareAndSet(null, throwable);
           LOG.error("Pipeline {} failed", jobInfo.jobId(), throwable);
-          // The pipeline is a job with an owner waiting on it, not a service to keep alive, so a
-          // failure stops the client rather than replacing the thread and carrying on.
+          // A job with an owner waiting on it, not a service: a failure stops the client.
           return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
         });
-    // Build the result before starting: it registers a state listener, and Kafka Streams only
-    // accepts one while the application is still in the CREATED state.
+    // Before start(): Kafka Streams only accepts a state listener while still in CREATED.
     KafkaStreamsPortablePipelineResult result =
         new KafkaStreamsPortablePipelineResult(
             kafkaStreams,
             context.getMetricsContainerStepMap(),
-            // Only once every task is initialized are the processors that have registered the whole
-            // set, and only then can "all of them are finished" mean the pipeline is finished.
+            // Only once every task is initialized is the registered set complete, so that "all
+            // finished" can mean the pipeline is finished.
             context.getTerminationTracker()::started);
-    // A bounded pipeline finishes; Kafka Streams has no notion of that, so the runner stops the
-    // client itself once every processor has reached the terminal watermark. Registered before
-    // start(), so a pipeline that drains quickly cannot finish before anything is listening.
+    // Kafka Streams has no notion of a finished pipeline, so the runner stops the client once every
+    // processor reaches the terminal watermark. Registered before start() so a fast drain is seen.
     context
         .getTerminationTracker()
         .onAllTerminated(