# Streaming Pipeline Topologies > Composable stage shapes for event-driven pipelines — copier, filter, splitter, sharder, merger — with ordering, semantic parallelism, state, shuffle and recovery boundaries; exactly-once scope across source, state and sinks; bounded joins and windows; watermarks, late-data policy, backlog versus flow control, and reproducible replay. Use when a stage is parallelised, when a join grows state without bound, when a stage re-keys the stream, when late events arrive after a window closed, when a windowed test uses wall-clock, or when reprocessing gives a different answer. Not whether to be event-driven (event-driven-architecture), ordering scope (message-ordering-and-partitioning), barriers (distributed-aggregation-and-barriers), skew (hot-partitions-and-rebalancing), the consumer (kafka-consumers-in-java), or in-process demand (reactive-backpressure). - Skill: `robsonkades/streaming-pipeline-topologies` (Agent Skill, multi-file: 4 files) - Install (CLI): `npx skillmds@latest add robsonkades/streaming-pipeline-topologies` - Raw SKILL.md: https://api.skillmd.com/api/skills/robsonkades/streaming-pipeline-topologies/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Product & Planning - Author: robsonkades (https://skillmd.com/u/robsonkades) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/robsonkades/streaming-pipeline-topologies --- # Streaming Pipeline Topologies ## Purpose Give a pipeline a vocabulary of stage shapes, and decide for each whether it may run at concurrency above 1 — the single question where correctness is silently traded for throughput. The shapes are small: **copier** (fan-out to independent consumers), **filter** (drop by predicate), **splitter** (one input, many outputs), **sharder** (re-partition by a new key), **merger/join** (combine streams; a union can interleave without keyed join state). Semantic parallelism is safe when operations commute/order does not matter, or keyed state and effects have one current owner with recovery/fencing. Stateless code can still emit ordered or non-idempotent effects; stateful frameworks can safely parallelize by key. Repartitioning is a shuffle boundary: old-key order no longer defines order among records sharing a new key, state must migrate/rebuild and skew changes. It does **not** inherently end exactly-once—some engines include repartition topics/state in one transaction or checkpoint. The two failures prevented are the stage parallelised because it "looked stateless", and the join that retains every key it has ever seen — which passes every load test and dies in week three of memory, not of throughput. ## Workflow Inspect deployed engine/API, connector and Java versions, topology configuration and state backend before using version-sensitive guarantees; no upgrade is implied. Missing watermark, checkpoint or sink evidence is unknown. Deliver a stage map with ordering/authority boundaries, state/backlog budgets and the tested recovery contract. Replay, resets and failure injection must use isolated targets or existing explicit authorization for their effects. 1. **Name each stage by shape** before drawing arrows. If an operator combines shapes, model each semantic step even when the implementation fuses them; this exposes separate ordering, state and failure boundaries without forcing an unnecessary network hop. 2. **For each stage, answer four questions:** what ordering does it preserve, is it safe above concurrency 1, what state does it hold, and how does it fail. The table is `references/stage-catalogue.md`. 3. **Mark every shuffle/repartition explicitly.** State old/new key, partitioner/count/epoch, ordering semantics, framework transaction/checkpoint boundary and recovery. Never inherit an exactly-once label across a sink the engine does not control. 4. **For any stateful stage, bound the state.** Specify key count, bytes/events per key, multiplicity, and the clock/progress that drives cleanup. A window or TTL alone is not a physical bound when progress stalls. See `references/stateful-stages.md`. 5. **Write the late-data policy down** — drop, side stream, or correction. Not deciding is accepting an unverified framework default; inspect and instrument it. 6. **Trace flow control and backlog separately.** Operator queues/credits can backpressure upstream within a job; a durable log usually decouples producers, so consumer lag measures backlog without slowing production. Bound both internal buffers and log retention/replay. 7. **Specify replay semantics**: use event time when historical event-time answers are required, pin timestamp/watermark/late-data rules, and test with controlled time rather than sleeping. ## Decision block ```text Run a stage above concurrency 1 when: - records/effects commute or sequence/version checks tolerate completion reordering, or - state is keyed by the partition key, one current owner is enforced, and checkpoint, rebalance and stale-owner behavior are defined Keep a stage at one worker per partition when: - downstream state is order-sensitive per key: a state machine, a CDC apply, an event-sourced projection. Unordered execution/emission can reorder it; ordered per-key lanes or validated ordered-async operators may preserve the required contract Do not treat parallelism as a knob when: - the stage changes the partitioning (sharder) or combines partitions (merger, join). Evaluate key ownership, migration and ordering; changing parallelism need not change the key Push the stage upstream instead when: - it is a filter with high selectivity and the source can evaluate the predicate Split into separate pipelines instead when: - two branches need different parallelism, ordering or retention. One topology forced to satisfy both is sized for the stricter and pays for it twice ``` ## Rules - Statelessness alone does not make effects order-insensitive. Parallelize freely only when output/effect composition tolerates completion order and duplicates; otherwise preserve a serial lane or version/sequence guard. Stateful keyed ownership also needs checkpoint, rebalance and stale-task fencing semantics. - Filter cost and removable upstream work depend on payloads, batching and predicate cost. Push a pure predicate earlier only if null/type/time semantics and required audit/security observations remain equivalent; a 99% record drop does not imply 99% byte or CPU savings. - A **splitter** raises one question and it is transactional: are the N outputs atomic? If not, a crash after output 1 leaves consumers of output 2 with a gap they must tolerate. What a transaction covers, and what it does not, is `delivery-semantics`. - A **sharder/shuffle** changes the key and forces three reviews: per-key ordering (records sharing a new key may arrive from inputs with no relative order), the guarantee/checkpoint scope (which may or may not include the shuffle), and skew profile (a new key is a new distribution — `hot-partitions-and-rebalancing`). - A stream-stream join without eviction can retain unmatched records indefinitely. Table/latest- value joins may retain current state per live key and tombstones may remove it; fixed-size aggregates need less per-key bytes than raw-event joins. Bound by semantic retention and measure distinct keys, unmatched events, bytes and compaction/checkpoint amplification. - A **copier**—a second consumer group—decouples offsets/failure but adds broker read/network/ cache and downstream cost. Kafka groups have independent offsets, but share topic retention and compaction; another group does not preserve expired input or create independent retention. - **Say which window and implementation you mean.** Naive sliding windows replicate each record across `size/step` windows; pane/incremental aggregation can reduce storage/CPU depending on whether the function is algebraically mergeable. Continuous sessions may never finalize, but state growth depends on accumulator versus raw-event/join storage. - A watermark is an engine/source assertion about event-time progress, commonly the minimum across active partitions plus out-of-orderness/idleness policy—not a guarantee. It encodes how long to wait for stragglers; the late-data policy is a separate decision about the one that arrives anyway. Name both — "it probably won't happen" is insufficient evidence for a loss policy. Replay behavior depends on how progress is rebuilt. - **In a log-based boundary, lag is durable backlog, not backpressure to producers.** A slow consumer reads later while producers may continue. Retention can make old input unavailable; internal queues/state can still OOM before that. Alert on age/bytes/catch-up capacity against SLO and effective retention. In-process demand signalling—`request(n)`, credits, bounded buffers strategies — is a different mechanism inside one JVM (`reactive-backpressure`). - Processing-time windows generally produce different buckets on replay; use them only when current processing behavior is the intended semantics. Event time improves reproducibility only with stable timestamp extraction, watermark/idleness rules, late policy, input snapshot and deterministic operators/sinks. - Replay is not merely "start at offset 0": a full recomputation needs isolated/reset state, while checkpoint recovery restores state and source positions consistently. Choose versioned output/cutover or a sink protocol that tolerates replay (`idempotency`) before running it. - Never size a state store from the average key: size it from distinct keys × per-key state × window multiplicity, and export the real number as a metric. A store whose size is visible only in a heap dump has already taken the outage. ## Exactly-once scope For each engine, enumerate source offsets, shuffle topics, state changelog/checkpoint and sinks inside one atomic recovery boundary. Kafka Streams `exactly_once_v2` can transactionally couple Kafka input/output/state changelog, but an external database call is outside. Flink checkpoints need a replayable source and checkpoint-aware/idempotent/transactional sink; checkpoint success does not make an arbitrary side effect exactly once. Upgrades, rescaling and savepoint/state- serializer compatibility are part of the contract. ## Security and operability - Authenticate/authorize internal topics/state stores and protect replay tools; topology duplication can bypass the API's tenant controls. - Minimize PII in repartition keys/changelogs and apply retention/deletion to derived copies. - Expose topology version, partition/key distribution, watermark per input, idle partitions, late/drop/correction count, state bytes/entries, checkpoint duration/failure and restore time. ## References - [The stage catalogue](references/stage-catalogue.md) — every shape with its ordering effect, parallel-safety condition, state requirement and characteristic failure, plus the composition rules and the shapes that are two stages pretending to be one. Read when designing a topology or reviewing whether a stage may be parallelised. - [Stateful stages](references/stateful-stages.md) — window types with their state cost, watermarks and late-data policy options, the unbounded-state failure with the metrics that catch it before OOM, state store sizing, and how to test a windowed join deterministically on controlled event time. Read before building a join, or when state is growing.