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. + *
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. * - *
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. + *
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. * *
* docker compose -f runners/kafka-streams/measurement/docker-compose.yml up -d - * ./gradlew :runners:kafka-streams:measurement:installDist - *+ * ./gradlew -Pwith-kafka-streams-runner :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 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. + *
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: * *
* <millis> <instance> window_end=<millis> key=<key> count=<n> skew_ms=<n> ** - *
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. + *
{@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() {} diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineRunner.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineRunner.java index baa62f31aa4a..486cc05d8b54 100644 --- a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/KafkaStreamsPipelineRunner.java +++ b/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( diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ExecutableStageProcessor.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ExecutableStageProcessor.java index fcb8b2ff27e6..5cb0672ff9fc 100644 --- a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ExecutableStageProcessor.java +++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ExecutableStageProcessor.java @@ -48,39 +48,32 @@ import org.slf4j.LoggerFactory; /** - * Kafka Streams {@link Processor} that executes a fused {@link ExecutableStage} (stateless user - * code such as ParDo) in the Beam SDK harness over the Fn API. + * Kafka Streams {@link Processor} that executes a fused {@link ExecutableStage} — stateless user + * code such as ParDo — in the Beam SDK harness over the Fn API. * - *
For each {@link KStreamsPayload#isData() data} payload it unwraps the {@link WindowedValue} - * and feeds it to the harness through the stage's main input {@link FnDataReceiver}. Harness - * outputs are collected on the harness threads into {@link #pendingOutputs} and then flushed - * downstream on the Kafka Streams processing thread when the bundle closes — Kafka Streams' {@link - * ProcessorContext#forward} must only be called from the processing thread, so outputs are never - * forwarded directly from a harness callback. + *
Each {@link KStreamsPayload#isData() data} payload is unwrapped and fed to the harness through + * the stage's main input {@link FnDataReceiver}. Harness outputs are collected on the harness + * threads into {@link #pendingOutputs} and flushed downstream when the bundle closes, because + * {@link ProcessorContext#forward} may only be called from the processing thread. * - *
A {@link KStreamsPayload#isWatermark() watermark} payload is a report from one partition of - * one upstream transform and marks a bundle boundary: the open bundle (if any) is closed (flushing - * outputs), the report is fed to the {@link WatermarkAggregator}, and the stage's output watermark - * is forwarded downstream — stamped with this stage's own transform id — only when the aggregate - * across the upstream transform's partitions actually advances. Until every partition has reported, - * the watermark is held and nothing is forwarded — but data is still processed in the meantime. + *
A {@link KStreamsPayload#isWatermark() watermark} payload marks a bundle boundary: the open + * bundle is closed and flushed, the report goes to the {@link WatermarkAggregator}, and the stage's + * output watermark is forwarded — stamped with this stage's transform id — only once the aggregate + * across the upstream partitions advances. Until every partition has reported the watermark is + * held, though data is still processed meanwhile. * - *
A bundle is also bounded in size, by {@code --maxBundleSize}, and closed once that many - * elements have been fed to it. Without the bound a bundle stays open until the next watermark, - * which on a stream that produces steadily lets it grow without limit. The bound is checked as - * elements arrive. A time bound ({@code --maxBundleTimeMs}) is not applied yet — see the option's - * own documentation. + *
A bundle is also bounded by {@code --maxBundleSize}, checked as elements arrive; without it a + * bundle would stay open until the next watermark and grow without limit on a steady stream. The + * time bound {@code --maxBundleTimeMs} is not applied yet, see that option's documentation. * - *
Closing a bundle asks Kafka Streams to commit, so the elements a bundle consumed and the - * records it produced are committed together and a restart replays either all of the bundle or none - * of it. Note that this aligns commits to bundle boundaries but does not stop Kafka - * Streams from committing on its own interval part-way through a bundle; closing the bundle first - * from a pre-commit hook would be needed to rule that out entirely. + *
Closing a bundle asks Kafka Streams to commit, so the elements consumed and the records + * produced commit together and a restart replays all of a bundle or none. This aligns commits to + * bundle boundaries but does not stop Kafka Streams committing on its own interval mid-bundle; + * ruling that out needs a pre-commit hook. * - *
This is the Kafka Streams analogue of Flink's {@code ExecutableStageDoFnOperator} and Spark's - * {@code SparkExecutableStageFunction}. State, timers, and side inputs are out of scope for this - * first version: the stage is executed with {@link StateRequestHandler#unsupported()} and no timer - * receivers. + *
The analogue of Flink's {@code ExecutableStageDoFnOperator} and Spark's {@code
+ * SparkExecutableStageFunction}. State, timers and side inputs are out of scope here: the stage
+ * runs with {@link StateRequestHandler#unsupported()} and no timer receivers.
*/
class ExecutableStageProcessor
implements Processor Data records are forwarded straight through unchanged — the merge of the N parents'
- * data streams is the flatten.
+ * Data records pass straight through — merging the parents' streams is the flatten. The work is
+ * in the watermark, which Flatten owns as GroupByKey does: a {@link WatermarkAggregator} over its
+ * inputs, forwarding its own watermark only when the minimum across them advances and stamping it
+ * as a single source. That holds the output back until every branch has reported, so a downstream
+ * GroupByKey cannot fire before all branches are drained.
*
- * Watermark reports are where Flatten does real work, and it owns its output watermark
- * the same way GroupByKey does: it runs a {@link WatermarkAggregator} over its inputs, forwards its
- * own watermark only when the {@code min()} across them advances, and stamps that as a single
- * source ({@code 0 of 1}) to its downstream. This holds the output watermark back until
- * every input branch has reported, so a downstream GroupByKey does not fire before all
- * flattened branches are drained.
- *
- * The {@link WatermarkAggregator} tells the input branches apart by the transform id each
- * branch's producer stamps on its watermark (Kafka Streams does not tell a processor which parent
- * forwarded a record). Each producer stamps its own identity regardless of who consumes it, so a
- * PCollection feeding several Flattens reports one identity and every Flatten still waits only for
- * the upstream transforms it expects — the set handed to it at construction from the pipeline
- * graph.
+ * Branches are told apart by the transform id each producer stamps, since Kafka Streams does not
+ * say which parent forwarded a record. A producer stamps its own identity regardless of who
+ * consumes it, so a PCollection feeding several Flattens reports one identity and each Flatten
+ * still waits only for the upstream transforms handed to it at construction.
*/
class FlattenProcessor
implements Processor Windowing and triggering are executed by Beam's {@link
- * org.apache.beam.runners.core.ReduceFnRunner} inside {@link WindowedGroupByKeyProcessor}, the same
- * way the Flink and Spark portable runners do it — so fixed/sliding windows, the default trigger,
- * allowed lateness and timestamp combiners all work. The input PCollection's windowing strategy is
- * hydrated from the pipeline proto and handed to the processor.
+ * Windowing and triggering run through Beam's {@link
+ * org.apache.beam.runners.core.ReduceFnRunner} inside {@link WindowedGroupByKeyProcessor}, as the
+ * Flink and Spark portable runners do, so fixed and sliding windows, the default trigger, allowed
+ * lateness and timestamp combiners all work. The input's windowing strategy is hydrated from the
+ * pipeline proto and handed to the processor.
*
- * Topology added (the Beam key becomes the Kafka record key so Kafka Streams shuffles by it):
- *
- * The repartition topic is expected to exist on the broker before the job starts (same
- * pre-create assumption as the Impulse bootstrap topic); auto-creation lands with the AdminClient
- * wiring in a follow-up.
+ * The Beam key becomes the Kafka record key so Kafka Streams shuffles by it. The topology is a
+ * {@link ShuffleByKeyProcessor} that sets that key and passes watermark reports through, a sink to
+ * an internal repartition topic using a {@link GroupByKeyBroadcastPartitioner} that hashes data by
+ * key and fans watermarks out to every partition, a source reading that topic back, and the {@link
+ * WindowedGroupByKeyProcessor} with its state and timer stores.
*/
class GroupByKeyTranslator implements PTransformTranslator {
diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ImpulseProcessor.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ImpulseProcessor.java
index 7c4590a0e5a1..9a84035bf8bb 100644
--- a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ImpulseProcessor.java
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ImpulseProcessor.java
@@ -34,28 +34,17 @@
/**
* Kafka Streams {@link Processor} implementing Beam's {@code Impulse} transform.
*
- * For each task instance, emits exactly two {@link KStreamsPayload}s downstream:
+ * Each task emits exactly two payloads: a {@link KStreamsPayload#data data} payload wrapping an
+ * empty {@code byte[]} in the {@link org.apache.beam.sdk.transforms.windowing.GlobalWindow} at
+ * {@link BoundedWindow#TIMESTAMP_MIN_VALUE}, then a {@link KStreamsPayload#watermark watermark} at
+ * {@link BoundedWindow#TIMESTAMP_MAX_VALUE} to say the source is done.
*
- * A persistent state store records whether the data element was already emitted, so a restart
+ * does not duplicate it. The terminal watermark is re-emitted on every restart instead, so
+ * downstream watermark holds still release after recovery.
*
- * A persistent state store records whether the data element has already been emitted so that
- * task restarts do not duplicate the data. The terminal watermark, on the other hand, is re-emitted
- * on every restart so downstream watermark holds release correctly after recovery (per Jan's review
- * on PR #38689).
- *
- * The trigger comes from a wall-clock punctuator scheduled on {@link #init} — this lets the
- * processor fire even when the dedicated bootstrap source topic is empty, which is the expected
- * production state.
- *
- * Kafka Streams disallows negative record timestamps, so the forwarded {@link Record} carries
- * the Unix epoch ({@code 0L}). The Beam event-time lives inside the {@link KStreamsPayload}
- * variant: inside the {@link WindowedValue} for data, or as the explicit watermark millis.
+ * A wall-clock punctuator scheduled in {@link #init} drives it, so the processor fires even
+ * though its bootstrap topic is empty, which is the normal production state.
*/
class ImpulseProcessor implements Processor Adds three nodes to the Kafka Streams {@link Topology}:
+ * Adds three nodes: a {@code byte[]} source bound to a per-transform bootstrap topic, which
+ * exists only because Kafka Streams refuses to start a topology with no source topic and whose
+ * records {@link ImpulseProcessor} ignores; the processor itself, which fires a one-shot wall-clock
+ * punctuator and emits one empty data payload followed by a terminal watermark; and a persistent
+ * state store recording whether it already fired, so a restart does not duplicate the impulse.
*
- * The processor's output PCollection is registered with the translation context so subsequent
- * translators can wire themselves to this node by id.
- *
- * Bootstrap topic lifecycle: this translator does not auto-create the bootstrap
- * topic. The topic is expected to exist on the broker before the job starts; otherwise Kafka
- * Streams raises {@code MissingSourceTopicException} on startup. The auto-create-vs-pre-create
- * decision (design doc §12.1) is deferred to a follow-up sub-issue along with the {@code
- * AdminClient} wiring; pre-creation is sufficient for the {@code TopologyTestDriver}-based unit
- * tests in this PR.
+ * The output PCollection is registered with the translation context so later translators can
+ * wire to this node by id. The bootstrap topic itself is created before startup by {@link
+ * org.apache.beam.runners.kafka.streams.KafkaStreamsTopicManager}.
*/
class ImpulseTranslator implements PTransformTranslator {
diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayload.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayload.java
index c165f0e875d4..647c64953b06 100644
--- a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayload.java
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KStreamsPayload.java
@@ -24,26 +24,13 @@
import org.checkerframework.checker.nullness.qual.Nullable;
/**
- * Sum-type envelope flowing between Kafka Streams processors in the Beam Kafka Streams runner.
+ * Envelope for every record value passed between the runner's processors. It is either a {@link
+ * #isData() data} element wrapping a {@link WindowedValue}, or a {@link #isWatermark() watermark}
+ * report carrying an event time plus the partition fields the downstream {@link WatermarkManager}
+ * needs.
*
- * Every record value emitted by a runner-introduced processor is one of:
- *
- * The envelope lets a single Kafka Streams output channel carry both Beam data and the watermark
- * / synchronization primitives that Kafka Streams does not natively support. Future control
- * messages (e.g. the {@code (epoch, assigned_partitions)} propagation from design doc §5) can be
- * added here as additional variants.
- *
- * This class is intentionally in-JVM only for now; serialization across topic boundaries
- * (repartition or sink topics) will be introduced when the first translator that emits to a topic
- * lands, at which point a corresponding Kafka {@link org.apache.kafka.common.serialization.Serde}
- * will be added.
+ * One channel therefore carries both Beam data and the watermark coordination Kafka Streams has
+ * no notion of. Across topic boundaries it is encoded by {@link KStreamsPayloadSerde}.
*
* @param Kafka Streams has no per-key timer service, so timers are persisted like any other state, in
- * two stores that serve the two ways a timer is looked up:
+ * two stores serving the two ways a timer is looked up. The identity store, keyed by {@code key |
+ * domain | timerFamily | timerId | namespace}, is how {@link #setTimer} overwrites and {@link
+ * #deleteTimer} removes exactly one timer as the contract requires; its value is the index key, so
+ * an overwritten timer's index entry can be removed without knowing what time it was set for. The
+ * index store, keyed by {@code domain | fireTimestamp | identity}, is how due timers are found: the
+ * timestamp is in the sortable form described on {@link StoreKeys}, so every event-time timer due
+ * at a watermark is one range scan rather than a scan of every timer of every key, and its value is
+ * the {@link TimerData} so firing needs no second lookup.
*
- * Firing is driven by {@link WindowedGroupByKeyProcessor}: on a watermark advance it range-scans
- * the index for event-time timers that are due and replays them through {@link
- * org.apache.beam.runners.core.ReduceFnRunner#onTimers}.
- *
- * This instance reports the times it was constructed with; it never fires timers itself.
+ * Firing is driven by {@link WindowedGroupByKeyProcessor}, which range-scans the index on a
+ * watermark advance and replays due timers through {@link
+ * org.apache.beam.runners.core.ReduceFnRunner#onTimers}. This instance only reports the times it
+ * was constructed with; it never fires timers itself.
*/
class KafkaStreamsTimerInternals implements TimerInternals {
diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KafkaStreamsTranslationContext.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KafkaStreamsTranslationContext.java
index 4a5463677163..cc8f57e0370c 100644
--- a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KafkaStreamsTranslationContext.java
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/KafkaStreamsTranslationContext.java
@@ -52,18 +52,13 @@ public class KafkaStreamsTranslationContext {
* has not been registered is produced by a single instance; only a shuffle raises the count.
*/
private final Map This is the {@code totalSourcePartitions} its watermark reports carry, and what a downstream
- * {@link WatermarkAggregator} waits to hear from before it lets the watermark advance. It changes
- * only at a shuffle: everything fused downstream of one runs at the shuffle topic's partition
- * count, and everything else runs as a single instance.
+ * The {@code totalSourcePartitions} this PCollection's watermark reports carry, which a
+ * downstream {@link WatermarkAggregator} waits on. It changes only at a shuffle.
*/
public void registerPCollectionPartitionCount(String pCollectionId, int partitionCount) {
pCollectionIdToPartitionCount.put(pCollectionId, partitionCount);
}
- /**
- * How many partitions the transform producing {@code pCollectionId} runs across; one unless a
- * shuffle upstream raised it.
- *
- * Always at least one: an unregistered PCollection is produced by a single instance, and the
- * only value ever registered is {@code --internalParallelism}, which the runner rejects below one
- * before translating.
- */
+ /** One unless a shuffle upstream raised it; never less, as --internalParallelism is validated. */
public int getPartitionCount(String pCollectionId) {
return pCollectionIdToPartitionCount.getOrDefault(pCollectionId, 1);
}
@@ -167,10 +140,8 @@ public String getProcessorNameForPCollection(String pCollectionId) {
}
/**
- * Returns the dedicated bootstrap topic name for one Impulse transform. Keyed by transform id
- * (sanitized to Kafka's legal topic-name character set) because a pipeline can contain several
- * Impulses (e.g. an empty {@code Create} plus the dummy branch {@code PAssert} adds), and Kafka
- * Streams rejects registering the same topic on two source nodes.
+ * Keyed by transform id, sanitized to Kafka's legal topic characters: a pipeline can hold several
+ * Impulses, and Kafka Streams rejects the same topic on two source nodes.
*/
public String getImpulseBootstrapTopic(String transformId) {
String sanitizedTransformId = ILLEGAL_TOPIC_CHARS.matcher(transformId).replaceAll("_");
@@ -180,11 +151,7 @@ public String getImpulseBootstrapTopic(String transformId) {
+ sanitizedTransformId;
}
- /**
- * Returns the dedicated bootstrap topic name a primitive Read reads from. Keyed by transform id
- * (sanitized to Kafka's legal topic-name character set) so multiple Reads — and Impulse — never
- * register the same topic on two source nodes, which Kafka Streams rejects.
- */
+ /** Keyed by transform id, for the same reason as {@link #getImpulseBootstrapTopic}. */
public String getReadBootstrapTopic(String transformId) {
String sanitizedTransformId = ILLEGAL_TOPIC_CHARS.matcher(transformId).replaceAll("_");
return READ_BOOTSTRAP_TOPIC_PREFIX
diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ReadProcessor.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ReadProcessor.java
index a6ef768ae2a1..a80376cdefc9 100644
--- a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ReadProcessor.java
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/ReadProcessor.java
@@ -42,37 +42,26 @@
* Kafka Streams {@link Processor} implementing Beam's deprecated primitive {@code Read}
* (beam:transform:read:v1) over a {@link BoundedSource}.
*
- * For each task instance, reads the whole {@link BoundedSource} once and emits, in order:
+ * Each task reads the whole source once, emitting one {@link KStreamsPayload#data data} payload
+ * per element in the {@link org.apache.beam.sdk.transforms.windowing.GlobalWindow} at its own event
+ * time, then a {@link KStreamsPayload#watermark watermark} at {@link
+ * BoundedWindow#TIMESTAMP_MAX_VALUE} to say the source is done.
*
- * Wire form. A Read produces decoded Java objects, but the harness's main-input receiver
+ * expects the runner-side wire form: a raw object for a model coder, a length-prefixed {@code
+ * byte[]} for a coder the runner does not know. Stage-to-stage edges already carry that form, so
+ * each element is transcoded here, encoded with the SDK-side wire coder and decoded with the
+ * runner-side one. The two are byte-compatible by construction, so this yields exactly what the
+ * receiver expects, nesting and all.
*
- * Wire form. Unlike Impulse (whose element is already an opaque {@code byte[]}), a Read
- * produces decoded Java objects. Downstream {@link ExecutableStageProcessor} feeds
- * whatever it receives straight into the SDK harness, whose main-input receiver expects each
- * element in the runner-side wire form — a raw object for a model coder, but a length-prefixed
- * {@code byte[]} for a coder the runner does not know (e.g. {@code VarIntCoder}). Stage-to-stage
- * edges already carry that wire form because harness outputs are decoded with the runner-side wire
- * coder; this processor reproduces it for the source edge by transcoding each element through the
- * SDK-side wire coder (encode) and back through the runner-side wire coder (decode). The two are
- * byte-compatible by construction, so the transcode yields exactly the object the receiver expects,
- * nesting and all.
+ * As in {@link ImpulseProcessor}, a state store records whether the elements were already
+ * emitted so a restart does not duplicate them, while the terminal watermark is re-emitted on every
+ * restart so downstream holds still release. A wall-clock punctuator scheduled in {@link #init}
+ * drives it, since the bootstrap topic is empty.
*
- * This mirrors {@link ImpulseProcessor}: a persistent state store records whether the elements
- * have already been emitted so task restarts do not duplicate them, while the terminal watermark is
- * re-emitted on every restart so downstream watermark holds still release after recovery. The
- * trigger is a wall-clock punctuator scheduled on {@link #init} so the processor fires even though
- * its bootstrap source topic is empty.
- *
- * The source is read in a single instance with no splitting — parallelism across the source's
- * splits arrives with the topic-based shuffle work (#18479). Kafka Streams disallows negative
- * record timestamps, so each forwarded {@link Record} carries the Unix epoch ({@code 0L}); the Beam
- * event time lives inside the {@link WindowedValue}.
+ * The source is read single-instance without splitting; parallel reads arrive with #18479. Kafka
+ * Streams rejects negative record timestamps, so each {@link Record} carries the Unix epoch and the
+ * Beam event time travels inside the {@link WindowedValue}.
*/
class ReadProcessor The runner forces every {@code Read.Bounded} (including the one {@code Create} of two or more
- * elements expands to) into this primitive read before translation — see {@code
- * KafkaStreamsTestRunner.translate}, which applies {@code
- * SplittableParDo.convertReadBasedSplittableDoFnsToPrimitiveReads}. This deliberately avoids the
- * default {@code BoundedSourceAsSDFWrapperFn} splittable-DoFn expansion, which the runner cannot
- * execute yet (no SDF restriction protocol), as agreed with the mentor.
+ * The runner converts every {@code Read} into this primitive before translation, rather than
+ * letting it expand into the default splittable-DoFn wrapper, which it cannot execute; see {@code
+ * KafkaStreamsRunner.prepareForTranslation}.
*
- * Adds the same three-node shape as {@link ImpulseTranslator}:
+ * The topology is the same three-node shape as {@link ImpulseTranslator}: a {@code byte[]}
+ * source on a per-transform bootstrap topic, since Kafka Streams will not start a topology with no
+ * source topic and the records on it are ignored; the {@link ReadProcessor}; and a persistent state
+ * store recording whether the read already fired, so a restart does not duplicate elements.
*
- * The processor emits elements in the runner-side wire form the downstream stage's SDK harness
- * expects, so it is handed the SDK-side and runner-side wire coders for the read's output
- * PCollection (see {@link ReadProcessor} for why). Only {@link
- * org.apache.beam.model.pipeline.v1.RunnerApi.IsBounded.Enum#BOUNDED bounded} sources are
- * supported; {@link ReadTranslation#boundedSourceFromProto} rejects an unbounded payload.
+ * Elements are emitted in the runner-side wire form the downstream harness expects, so the
+ * processor is handed both wire coders for the output PCollection — see {@link ReadProcessor}.
+ * Bounded and unbounded sources are both supported, by {@link ReadProcessor} and {@link
+ * UnboundedReadProcessor} respectively.
*/
class ReadTranslator implements PTransformTranslator {
diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/TerminationReporter.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/TerminationReporter.java
index abc10500d211..6597b277770c 100644
--- a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/TerminationReporter.java
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/TerminationReporter.java
@@ -25,25 +25,19 @@
import org.checkerframework.checker.nullness.qual.Nullable;
/**
- * The bit of every watermark-emitting processor that reports it has finished, so a bounded pipeline
- * can stop itself. See {@link TerminationTracker} for why the runner has to work this out at all.
+ * The part of every watermark-emitting processor that reports it has finished, so a bounded
+ * pipeline can stop itself; see {@link TerminationTracker} for why that has to be worked out at
+ * all. A processor calls {@link #init} from {@code Processor#init}, passes every watermark it emits
+ * to {@link #watermarkEmitted}, and calls {@link #close} from {@code Processor#close}.
*
- * A processor creates one of these, calls {@link #init} from {@code Processor#init}, passes
- * every watermark it emits to {@link #watermarkEmitted}, and calls {@link #close} from {@code
- * Processor#close}.
+ * The report is scheduled rather than made inline, because reporting from inside {@code
+ * process()} would announce the processor finished while it is still handling the record that
+ * carried the terminal watermark; deferring it lets flushing, forwarding and committing happen
+ * first.
*
- * Reporting from inside {@code process()} would announce the processor as finished while it is
- * still in the middle of handling the record that carried the terminal watermark. Scheduling a
- * punctuator instead defers the report until the current processing has completed, so anything that
- * has to happen after the final watermark — flushing a bundle, forwarding downstream, committing —
- * still runs first.
- *
- * The punctuator is {@link PunctuationType#WALL_CLOCK_TIME} rather than stream time: no further
- * records arrive after the terminal watermark, so stream time would never advance and a stream-time
- * punctuator would never fire. The interval is the smallest Kafka Streams accepts — it rejects
- * anything below a millisecond with "The minimum supported scheduling interval is 1 millisecond."
+ * It uses {@link PunctuationType#WALL_CLOCK_TIME}: no records arrive after the terminal
+ * watermark, so stream time would never advance and a stream-time punctuator would never fire. The
+ * interval is 1ms, the smallest Kafka Streams accepts.
*/
class TerminationReporter {
diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/TerminationTracker.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/TerminationTracker.java
index 6c570db1aff1..d0a04e38126a 100644
--- a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/TerminationTracker.java
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/TerminationTracker.java
@@ -26,39 +26,24 @@
/**
* Decides when a bounded pipeline has finished, so the Kafka Streams client can be stopped.
*
- * Kafka Streams has no notion of a processor being finished: a topology runs until something
- * closes the client. A bounded Beam pipeline does finish, though, and the runner already knows
- * when: every processor emits a watermark of {@link
+ * Kafka Streams has no notion of a finished processor; a topology runs until something closes
+ * the client. A bounded pipeline does finish, and the runner already knows when, because every
+ * processor emits {@link
* org.apache.beam.sdk.transforms.windowing.BoundedWindow#TIMESTAMP_MAX_VALUE} once its input is
- * exhausted. This class collects those reports and fires a callback when there is nothing left to
- * do.
+ * exhausted. This collects those reports and fires a callback when nothing is left to do.
*
- * No coordination between instances is needed: a watermark crossing a repartition topic is
+ * broadcast to every partition (see {@link GroupByKeyBroadcastPartitioner}), so every task observes
+ * the terminal watermark itself and all instances reach the same conclusion independently.
*
- * A watermark that crosses a repartition topic is broadcast to every partition (see
- * {@link GroupByKeyBroadcastPartitioner}), so every task of every downstream transform observes the
- * terminal watermark on its own, whichever instance it happens to run on. Each instance can
- * therefore decide to stop from what it sees locally, and they all reach the same conclusion
- * without talking to each other.
+ * Every local processor is counted, not just the first. One instance can own tasks from both
+ * sides of a repartition topic, and the upstream side goes terminal as soon as it has written to
+ * the topic while the downstream side still has to consume it. Stopping at the first would drop
+ * that work and still report success.
*
- * One instance can own tasks from both sides of a repartition topic. The upstream side goes
- * terminal as soon as it has written its data to the topic, while the downstream side still has to
- * consume it. Stopping the client when the first processor finishes would cut that downstream work
- * off and report the pipeline as done having silently dropped it. So the callback only fires once
- * every processor instance registered here has terminated.
- *
- * An instance that happens to own only upstream tasks still terminates on its own, which is
- * correct: what it wrote is durable in the topic for whichever instance reads it.
- *
- * One tracker belongs to one pipeline, not to the JVM. The job server runs many jobs in a single
- * process, so a shared static tracker would let one pipeline finishing tear down another.
- *
- * A pipeline with an unbounded source never produces a terminal watermark, so the callback never
- * fires and the client keeps running — which is the intended behaviour for a streaming job.
+ * A tracker belongs to one pipeline rather than to the JVM: the job server runs many jobs in one
+ * process, and a static tracker would let one job stop another. An unbounded pipeline never
+ * produces a terminal watermark, so the callback never fires and the client keeps running.
*/
public class TerminationTracker {
diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/UnboundedReadProcessor.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/UnboundedReadProcessor.java
index 4f8c5f105566..21fc392990ea 100644
--- a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/UnboundedReadProcessor.java
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/UnboundedReadProcessor.java
@@ -44,29 +44,24 @@
* Reads an {@link UnboundedSource} and forwards its elements and watermark downstream.
*
* Where the bounded {@link ReadProcessor} drains its source once and jumps the watermark to the
- * end of time, an unbounded source never finishes: it is polled repeatedly, and its watermark
- * advances gradually as the reader reports progress. That difference is what makes this a streaming
- * runner rather than a batch one — downstream windows close because the source says time has moved
- * on, not because the input ran out.
+ * end of time, an unbounded source never finishes: it is polled repeatedly and its watermark
+ * advances as the reader reports progress. That is what makes downstream windows close because time
+ * moved on rather than because the input ran out.
*
- * Polling happens on a wall-clock punctuator rather than in {@code process}, because the
- * processor's bootstrap topic is empty and nothing else would drive it. Each turn reads at most
- * {@link #maxElementsPerPoll} elements so a busy source cannot monopolise the Kafka Streams thread
- * and starve the rest of the topology, then forwards the reader's watermark if it advanced.
+ * Polling runs on a wall-clock punctuator, since the bootstrap topic is empty and nothing else
+ * would drive it. A turn is bounded by {@link #maxElementsPerPoll} and by {@link #maxPollTimeMs} so
+ * a busy source cannot hold the Kafka Streams thread and starve the rest of the topology.
*
- * Restart is what the checkpoint mark is for. {@link UnboundedReader#getCheckpointMark()}
- * describes the position the reader has consumed to; it is written to a persistent state store, and
- * on {@link #init} the reader is created from the stored mark rather than from scratch, so a task
- * that moves or restarts resumes where it left off instead of re-reading from the beginning. The
- * store is changelogged and, under exactly-once, its writes commit atomically with the records the
- * processor forwarded, so the mark can never be ahead of the data that was actually emitted.
+ * The checkpoint mark is what makes restart work. {@link UnboundedReader#getCheckpointMark()} is
+ * written to a persistent state store and the reader is recreated from it in {@link #init}, so a
+ * task that moves or restarts resumes where it left off. The store is changelogged and, under
+ * exactly-once, commits atomically with the records forwarded, so the mark cannot run ahead of the
+ * data actually emitted.
*
- * The source handed to this processor has already been split by {@link ReadTranslator}, which is
- * where splitting belongs: it happens once for the pipeline rather than once per task instance, and
- * the contract does not define splitting an already-split source. Reading several splits in
- * parallel arrives with the topic-based shuffle work (#18479). As in the bounded processor, Kafka
- * Streams disallows negative record timestamps, so each forwarded {@link Record} carries the Unix
- * epoch and the Beam event time travels inside the {@link WindowedValue}.
+ * The source is split once by {@link ReadTranslator} rather than per task; reading several
+ * splits in parallel arrives with #18479. Kafka Streams rejects negative record timestamps, so each
+ * {@link Record} carries the Unix epoch and the event time travels inside the {@link
+ * WindowedValue}.
*/
class UnboundedReadProcessor A batch is capped at {@link #maxElementsPerPoll} so that the checkpoint mark and the
- * watermark are updated as the reader progresses rather than only at the end. Batches run back to
- * back while the source keeps filling them, since returning after every batch would cap
- * throughput at one batch per punctuation interval.
+ * A batch is capped at {@link #maxElementsPerPoll} so the checkpoint mark and the watermark
+ * move as the reader progresses, and batches run back to back while the source keeps filling
+ * them, since returning after each would cap throughput at one batch per punctuation.
*
- * The run is bounded all the same. A source that always has data — which is the normal case
- * for one that is keeping up — would otherwise never let this method return, and the Kafka
- * Streams thread would never get back to committing or to the rest of the topology. So at most
- * {@link #checkpointEveryNPolls} batches are taken before yielding, which is also where the
- * checkpoint mark is stored, and the next punctuation carries on from there.
- *
- * That batch bound is a count, and a count cannot bound the time: how long an element takes is
- * decided by the pipeline underneath it, which the source knows nothing about. A punctuator is
- * expected to be quick, and this one runs on the thread that also serves the rest of the
- * topology, so a turn that overruns its own {@link #POLL_INTERVAL} is due again as soon as it
- * returns and runs once more instead of the tasks below it. Measured on a grouping pipeline, a
- * turn of 200 elements took 3ms and held the thread 6% of the time, while a turn of 5000 took
- * 57ms and held it 89%, and the pipeline read tens of millions of elements while emitting none.
- * {@link #maxPollTimeMs} bounds the turn in time as well, and whichever bound is reached first
- * ends it.
+ * Both bounds exist because this runs on the thread that also serves the rest of the topology.
+ * At most {@link #checkpointEveryNPolls} batches are taken before yielding, and {@link
+ * #maxPollTimeMs} bounds the turn in time — a count cannot, since how long an element takes is
+ * decided by the pipeline below the source. A turn that overruns {@link #POLL_INTERVAL} is due
+ * again the moment it returns and runs instead of the tasks beneath it, which shows up as a
+ * pipeline that reads steadily and emits nothing.
*/
private void poll() {
if (exhausted) {
diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/WatermarkAggregator.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/WatermarkAggregator.java
index c9af03df26a9..e081d2e9db6b 100644
--- a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/WatermarkAggregator.java
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/WatermarkAggregator.java
@@ -26,29 +26,20 @@
import org.joda.time.Instant;
/**
- * Computes a transform's input watermark from the watermark reports of its upstream transforms.
+ * Computes a transform's input watermark from the reports of its upstream transforms.
*
- * A watermark report carries three orthogonal pieces of information (see {@link
- * WatermarkPayload}): which transform produced it, which partition (physical
- * instance) of that transform it is for, and how many partitions that transform has. A
- * producer stamps its own identity without regard to who consumes the report. This aggregator is
- * the consuming side, used by every transform that aggregates a watermark — ExecutableStage,
- * GroupByKey, Flatten (and CombinePerKey later):
+ * A report says which transform produced it, which partition of that transform it is for, and
+ * how many partitions that transform has (see {@link WatermarkPayload}); a producer stamps its own
+ * identity without regard to who consumes it. This is the consuming side, used by every transform
+ * that aggregates a watermark — ExecutableStage, GroupByKey, Flatten.
*
- * It is constructed with the upstream transform ids the consumer expects, known from the
+ * pipeline graph at translation time, and tracks each with its own {@link WatermarkManager}. The
+ * input watermark is the minimum across them, defined only once every expected upstream is ready;
+ * until then {@link #advance()} returns {@link BoundedWindow#TIMESTAMP_MIN_VALUE} and the caller
+ * emits nothing.
*
- * Not thread-safe; the caller (a single Kafka Streams processor thread) serializes access.
+ * Not thread-safe; the calling Kafka Streams processor thread serializes access.
*/
final class WatermarkAggregator {
diff --git a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/WatermarkManager.java b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/WatermarkManager.java
index cd0fca3654ae..39b9f11bb57d 100644
--- a/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/WatermarkManager.java
+++ b/runners/kafka-streams/src/main/java/org/apache/beam/runners/kafka/streams/translation/WatermarkManager.java
@@ -24,44 +24,23 @@
import org.joda.time.Instant;
/**
- * In-memory tracker of a single fused stage's input watermark, computed from the committed
- * watermarks reported by the upstream source partitions that feed it (the output /
- * repartition-topic partitions of the parent stage).
+ * Tracks one fused stage's input watermark from the committed watermarks reported by the upstream
+ * source partitions feeding it. Kept free of Kafka wiring so it can be unit-tested on its own.
*
- * This is the core of the Kafka Streams runner's watermark propagation, decoupled from the Kafka
- * wiring so it can be unit-tested in isolation. The wiring that produces the reports (flushing
- * {@code (sourcePartition, committedWatermark, totalSourcePartitions)} atomically with each offset
- * commit and fanning it out to every downstream partition) and consumes them lands in a follow-up.
+ * It counts source partitions rather than producer instances. A partition count is fixed, known,
+ * and travels in-band with every report, whereas instances come and go on every rebalance and can
+ * die without notice; a dead instance's partitions are reassigned and the new owner keeps
+ * reporting.
*
- * Until every source partition has reported, the stage's input watermark is undefined and {@link
+ * #advance()} returns {@link BoundedWindow#TIMESTAMP_MIN_VALUE}. A change in the partition count
+ * clears the reports and re-opens that hold, which subsumes an explicit epoch rule.
*
- * The question a stage has to answer is "have I received the watermark from every upstream
- * producer, so that {@code min()} across them is meaningful?". Counting producer instances
- * is hard: an instance can be killed without notice, leaving stale state, and the number changes on
- * every rebalance. Counting source partitions is robust instead, because the partition count
- * is fixed and known: it travels in-band with every report ({@code totalSourcePartitions}), a
- * partition is always owned by exactly one live instance, and when an instance dies its partitions
- * are reassigned and the new owner keeps reporting. So the manager only ever reasons about
- * partitions, never about instances. (Design agreed with the mentor; see the watermark
- * coordination-channel PoC findings.)
+ * Watermarks must not go backwards, so each partition's watermark is held monotonic and the
+ * emitted one is clamped against the last emitted — a newly appeared partition may report an older
+ * watermark than the stage has already reached.
*
- * Until a committed watermark has been seen for every source partition, the stage's input
- * watermark is undefined and {@link #advance()} returns {@link BoundedWindow#TIMESTAMP_MIN_VALUE} —
- * i.e. the stage emits no meaningful watermark downstream. A change in {@code
- * totalSourcePartitions} (e.g. a repartition) clears the accumulated reports and re-opens this hold
- * until the new full set has reported, which subsumes the "new epoch / revert" rule without an
- * explicit epoch.
- *
- * Beam watermarks must be non-decreasing. Each source partition's watermark is held monotonic (a
- * lower report is ignored), and the emitted stage watermark is additionally clamped so it never
- * regresses below the previously emitted value — relevant if a newly appeared partition reports an
- * older watermark after the stage had already advanced.
- *
- * Not thread-safe; the caller (a single Kafka Streams processor thread) serializes access.
+ * Not thread-safe; the calling Kafka Streams processor thread serializes access.
*/
public final class WatermarkManager {
diff --git a/sdks/python/apache_beam/runners/portability/kafka_streams_runner.py b/sdks/python/apache_beam/runners/portability/kafka_streams_runner.py
index a2d043ca174e..890ec41164a6 100644
--- a/sdks/python/apache_beam/runners/portability/kafka_streams_runner.py
+++ b/sdks/python/apache_beam/runners/portability/kafka_streams_runner.py
@@ -89,8 +89,7 @@ def path_to_jar(self):
'using `./gradlew runners:kafka-streams:job-server:shadowJar`.' %
self._jar)
return self._jar
- return self.path_to_beam_jar(
- ':runners:kafka-streams:job-server:shadowJar')
+ return self.path_to_beam_jar(':runners:kafka-streams:job-server:shadowJar')
def java_arguments(
self, job_port, artifact_port, expansion_port, artifacts_dir):
- *
- *
- *
- *
+ *
- *
- *
- *
- *
- *
- *
- *
- *
- *
- *
+ *
- *
- *
- * Why termination is scheduled rather than reported inline
- *
- * Why no coordination between instances is needed
+ * Why every local processor has to be counted, not just the first
- *
- * Scope
- *
- *
- *
+ * Why source partitions, not producer instances
+ * Holding until ready
- *
- * Monotonicity
- *
- *