Kafka Streams: give a Streams topology PC's per-key parallelism #255

Goal

Let a Kafka Streams topology process records with Parallel Consumer's parallelism, so a Streams app
gets per-key concurrency beyond its partition count instead of one record at a time per task.

Today Kafka Streams has no parallel processing of its own, and no prospect of it: KIP-311 is
explicitly abandoned and KIP-408's discussion thread has been dead since January 2019. PC is the only
thing on the JVM that does per-key concurrency cleanly. The two have never been joined up.

Why this, and why now

Nobody has ever filed an issue for this - upstream has 249 issues and a full-text sweep over title,
body and comments for "kafka streams", "streams integration", "topology", "DSL", "async processing",
"KIP-311", "KIP-408" and "stream processor" returns nothing. Upstream PR confluentinc#390 exists but
linked no issue and its body is the untouched template.

The demand is real, it just lives in Discussions, which is why an issue mirror never surfaced it:

Read the titles: what people asked for is state stores with PC. That is worth noting, because
state is the single most expensive thing to build from scratch - and the approach below gets it for
free by staying inside Kafka Streams rather than reimplementing it.

In confluentinc#596 maintainer rkolesnev reached the same conclusion from the other side: "it would
be easier to do it the other way around and figure out a way to add processing parallelisation to
Kafka Streams instead."

Background

Full assessment, with source citations against Apache Kafka 3.9 and trunk:
docs/plans/2026-08-07-002-investigate-kafka-streams-on-pc-report.md.

Two things it settles up front:

• Upstream PR Features/streams confluentinc/parallel-consumer#390 is not this. It sketches a PC-native Streams-like DSL, never
  touches Kafka Streams, and has never compiled. Useful as an API-shape sketch, not as a code port.
• Swapping in a PC-backed Consumer via KafkaClientSupplier does not work. Not an API problem -
  Streams serialises work above the consumer. StreamTask.process() pulls one record at a time via
  partitionGroup.nextRecord(). It also degrades to a silent no-op under KIP-1071's streams rebalance
  protocol on Kafka 4.1+, where StreamThread bypasses the supplier and only logs a warning.

Approach

Cut the seam one layer lower: replace record selection with PC's WorkManager, and execute the
processor chain on PC's worker pool. This needs a fork of
streams/src/main/java/org/apache/kafka/streams/processor/internals/, but the diff is bounded.

The fit is better than expected:

• PC's WorkManager already exposes the right four methods with the right types - registerWork,
  getWorkIfAvailable, onSuccessResult/onFailureResult, and
  collectCommitDataForDirtyPartitions(), which already returns
  Map<TopicPartition, OffsetAndMetadata>.
• PC already guarantees at most one in-flight record per key under KEY ordering
  (ProcessingShard:149-154). The per-key ordering queues comparable systems had to build are the
  shard manager's core invariant.
• Kafka already ships SynchronizedPartitionGroup, selected at StreamTask.java:199-207 behind an
  internal flag. The selection side is solved upstream for free.

The diff

One change is load-bearing: AbstractProcessorContext.java:47-48 holds recordContext and
currentNode as a single mutable slot per task, non-volatile, read ambiently by every store wrapper.
Under concurrency the failure mode is silent corruption, not a crash. Make these thread-local and most
of the rest is mechanical.

The caching layer gets deleted from the parallel path, not fixed. CachingKeyValueStore has a
whole-store exclusive lock, ThreadCache's eviction budget is per-thread across every task, and
eviction runs downstream user topology inline via context.forward on whichever thread called put.
withCachingDisabled() removes the whole class of problem in one line. It costs a real semantic
change: the DSL emits every update rather than one per cache flush.

Honest costs

• Stays serial: punctuators, stream-stream joins (sharedTimeTracker is shared across both sides
  of a join), and every operator-local observedStreamTime - those are plain non-volatile longs
  doing read-modify-write, so concurrency corrupts them rather than merely reordering.
• Changes semantics: windowed aggregations, suppress, segment retention - windowCloseTime is
  derived from observed stream time, so reordering changes which records are dropped as late.
• Cleanly safe: non-windowed aggregations - count, reduce, KStreamAggregate,
  KTableAggregate - do not read stream time at all.
• The parallel/serial boundary is a build-time, per-source-node-DAG decision, not a mid-subtopology
  split: forwardInternal calls child.process(record) directly with no queue, and the only
  buffering boundary Streams has is a repartition topic. InternalTopologyBuilder is the deciding
  class.
• It is a fork to maintain against upstream.

Suggested first step

A spike, not the full thing: Tier 1 plus the thread-local change, and find out whether it runs at all.
Nothing larger should be committed to before that answers.

Stretch goal

If the Web GUI (#215) has landed by then, extend it to show the Kafka Streams side: per-task
shard depth, which source-node DAGs were tagged parallel-safe versus left serial, and where records are
actually spending time. The GUI's whole point is showing what a running instance is doing, and "which
half of my topology is actually running in parallel" is exactly the question this feature creates and
nothing else can answer.

Not in scope

A PC-native Streams-like DSL (the other reading of confluentinc#390). Viable while stateless, but
reaching state parity means building the embedded-store-plus-changelog-plus-repartition subsystem from
scratch - and per the Discussions above, state is precisely what users asked for. Separate issue if
ever wanted.

Open question

RocksDB's own JNI guarantee for concurrent Get/Put on one handle is unverified. Kafka asserts
it nowhere - a thread-safety string search over RocksDBStore.java returns zero hits. Needs sourcing
from RocksDB upstream before anything leans on it.