What does ProcessorContext.forward() do, and how do you control where a forwarded record goes?

**Context.** A **Processor** in the Kafka Streams Processor API processes one input record at a time in its `process(Record<KIn,VIn> record)` method. To produce output, it does not `return` a value — instead it calls **`context.forward(...)`** on the **ProcessorContext** it received in `init()`. **What forward does.** `ProcessorContext<KOut,VOut>.forward(Record<KOut,VOut> record)` hands the record to the **downstream children** of the current processor node in the topology graph. A child may be another processor (which immediately runs its `process()` on this record, on the same thread) or a **sink node** (which serializes the record and writes it to an output Kafka topic). Forwarding is **synchronous and depth-first**: control flows down the sub-topology and returns before `forward` returns. **Multiplicity — the superpower over the DSL.** The DSL's `mapValues` is one-in-one-out; `flatMap` is one-in-many-out. PAPI gives you full control directly: - **Drop a record:** don't call forward at all (filter). - **Transform:** call forward once with a new Record. - **Fan out / flat-map:** call forward in a loop, emitting many records from one input. - **Emit nothing on input but later:** buffer in a state store and forward from a **punctuator** (time-driven). **Targeting a specific child — manual branching.** `forward(Record, String childName)` sends the record to **only the named child** instead of all children. The `childName` is the processor/sink name you gave in `Topology.addProcessor("name", ...)` / `addSink("name", ...)`. This lets one processor route different records to different downstream branches (e.g., valid → A, invalid → B) — the imperative analog of the DSL's `KStream.split()/branch()`. **Building the output Record.** A `Record<K,V>` is immutable and carries `key`, `value`, `timestamp`, and `Headers`. You construct `new Record<>(key, value, timestamp)` or derive one from the input with `record.withValue(newVal)`, `withKey(newKey)`, `withTimestamp(ts)`, `withHeaders(h)`. Choosing the **timestamp** matters: it becomes the stream-time contribution of the output record and affects downstream windowing/punctuation. **Where you may call forward.** Only from the **stream thread** that owns the task: inside `process()`, or inside a **punctuator** scheduled via `context.schedule(...)`. You must **not** forward from another thread (e.g., a callback from an external async client) — that violates the single-threaded task model and can corrupt state/ordering. If you need async, stage results in a queue/store and forward them from the next `process()`/punctuation on the stream thread. **Type safety (api package).** `ProcessorContext` is parameterized `<KOut,VOut>`, so the compiler enforces that you only forward records of the processor's declared output types. The old `org.apache.kafka.streams.processor.Processor`/`ProcessorContext` were untyped (`forward(Object,Object)`), which is error-prone — prefer the `api` package. **Edge cases.** - Forwarding to a child that doesn't exist (wrong name) throws — names must match topology wiring. - A processor with no children that never forwards is effectively a terminal sink-of-nothing; usually you forward to a sink or it's a dead end. - Forwarding many records per input increases downstream load and can affect commit/processing latency; the whole chain runs before the next input record is processed.