refactor(core) #225: put recovery on producerConfig alone, and move the factory, the derived id, redaction and the deprecation up the stack

This PR now stacks on #426 - producerConfig with the default constructor -
rather than on #420. Everything #420 carries beyond that rung leaves
this branch's tree here and returns above it, in #420 itself, which now stacks on this PR:
ProducerFactory and its contract exception, TransactionalIdDerivation, ProducerConfigRedaction (and
their tests), the deprecation of the instance option with its WARN and removal queue, the README
ownership section and migration note, and the four migrated example apps, which go back to master's.

What recovery is built on instead: the module's protected buildProducer(Map) seam from #426.
replacementProducerWrap() builds another wrapper through it from the same configuration, the caller's
transactional.id included, and hands the id to the ReplacementProducerSource so a failure to build can
still name it. The build runs through UserFunctions.carefullyRun as before: the seam is overridable,
and an Error from the constructor must surface as a build failure rather than escape the recovery
pass. The policy's terminal set loses the factory-contract clause with the factory; Authorization,
UnsupportedVersion and Error stay.

Tests move with it. ProducerRecoveryTest hands ParallelEoSStreamProcessor a PCModule whose seam hands
out the spied MockProducers the factory used to, records the id each build was handed, and asserts it
is the caller's, unchanged, on every build; the log-line and failure-message assertions key on a
per-instance id the test sets, as they keyed on the derived one. ProducerFencingRecoveryIT sets its
own transactional.id so the rogue can fence under it, and drops the derived-shape assertion, which
returns with the derivation. ProducerManagerRecoveryTest's factory-contract case leaves with the
contract. The two option/module test classes are rung 1's.

Records: the plan carries a dated re-cut note at its head rather than a rewrite of its units; the
in-flight note lists the three-PR stack; the roadmap entry, the issue-response draft, the refactoring
queue and the README template (a producer-recovery section only, no ownership section) say what this
PR now says.

Mechanically this commit is the inverse of what #420 becomes: reverting it on top of this
branch is how #420 is re-based, so its history is kept too.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VrpH51xNDodaajE4P2nhFg

Author: astubbs

README.adoc

@@ -523,30 +523,30 @@ Where `${project.version}` is the version to be used:
 [source,java,indent=0]
 ----
         Consumer<String, String> kafkaConsumer = getKafkaConsumer(); // <1>
+        Producer<String, String> kafkaProducer = getKafkaProducer();
 
         var options = ParallelConsumerOptions.<String, String>builder()
                 .ordering(KEY) // <2>
                 .maxConcurrency(1000) // <3>
                 .consumer(kafkaConsumer)
-                .producerConfig(getProducerConfig()) // <4>
-                .producerFactory(getProducerFactory())
+                .producer(kafkaProducer)
                 .build();
 
         ParallelStreamProcessor<String, String> eosStreamProcessor =
                 ParallelStreamProcessor.createEosStreamProcessor(options);
 
-        eosStreamProcessor.subscribe(of(inputTopic)); // <5>
+        eosStreamProcessor.subscribe(of(inputTopic)); // <4>
 
         return eosStreamProcessor;
 ----
 
 <1> Setup your clients as per normal.
+A Producer is only required if using the `produce` flows: hand PC either a finished `Producer` instance, or the configuration to build one from as `producerConfig` - any `ProducerConfig` key, serializers included, and PC builds it with `new KafkaProducer<>(config)`, and builds it again if the broker invalidates it (see <<producer-recovery>>).
+In the transactional commit mode, set `transactional.id` in that configuration as you would when building the producer yourself.
 <2> Choose your ordering type, `KEY` in this case.
 This ensures maximum concurrency, while ensuring messages are processed and committed in `KEY` order, making sure no offset is committed unless all offsets before it in it's partition, are completed also.
 <3> The maximum number of concurrent processing operations to be performing at any given time.
 Also, because the library coordinates offsets, `enable.auto.commit` must be disabled in your consumer.
-<4> Producer configuration, only required if using the `produce` flows: PC builds the producer from it through the factory on the next line (the default is `new KafkaProducer<>(config)`; override it to wrap or instrument the producer, keeping every key it is given), and rebuilds it if the broker invalidates it - see <<producer-ownership>> and <<producer-recovery>>.
-Handing PC a finished `Producer` instance still works, deprecated.
 <5> Subscribe to your topics
 
 NOTE: Because the library coordinates offsets, `enable.auto.commit` must be disabled.
@@ -635,7 +635,7 @@ There is no guarantee that the messages will be retried again in the same batch.
 ----
         ParallelStreamProcessor.createEosStreamProcessor(ParallelConsumerOptions.<String, String>builder()
                 .consumer(getKafkaConsumer())
-                .producerConfig(getProducerConfig())
+                .producer(getKafkaProducer())
                 .maxConcurrency(100)
                 .batchSize(5) // <1>
                 .build());
@@ -794,18 +794,6 @@ The API is source-compatible, so beyond the import lines no source edit is neede
 Committed offset metadata is unchanged too, so an existing consumer group can be upgraded in place without resetting or migrating offsets.
 The library's shared internal utilities moved with the rest, into `bz.stub.parallelconsumer.internal.utils` -- they were always plumbing, they are not part of the published API, and they are now named as such.
 
-==== Move from a producer instance to producer configuration
-
-The `producer` option - a finished `Producer` instance - is deprecated in favour of `producerConfig` (plus, optionally, `producerFactory`), and its removal is queued for `0.7.0.0`.
-Nothing forces the move in this release, but only the configuration path recovers when the broker invalidates the producer; see <<producer-recovery>>.
-To move:
-
-* Replace `.producer(kafkaProducer)` with `.producerConfig(producerProperties)`, the same map you built the producer from, serializers included.
-* Drop any `transactional.id` from that map. PC derives one per running instance, and a value you set is replaced with a WARN.
-* If your cluster grants TransactionalId ACLs, grant a prefixed one on `pc-<L>-<group.id>-` (`<L>` being the length of your `group.id`) - the literal id you used to grant on will no longer be used.
-* If operational tooling keyed on your `transactional.id` - dashboards, alerts, `kafka-transactions.sh` scripts - it now needs the derived prefix instead; the exact id is logged at start-up.
-* If you wrapped or instrumented the producer, do that in a `producerFactory` instead; it receives the resolved configuration and must build from it unchanged.
-
 === From 0.4 to 0.5
 
 This version has a breaking change in the API - instead of passing in `ConsumerRecord` instances, it passes in a `PollContext` object which has extra information and utility methods.
@@ -1101,24 +1089,11 @@ It is covered by an existing test, but no negative control was observed for it,
 The produce-lock timeout returning the record for retry started out the same way and no longer is: the eager-processing test holds the commit lock shut until a worker's acquisition is observed to time out, then watches that record's retry succeed, and removing the hold turns it red.
 It is recorded `PROVED`.
 
-[[producer-ownership]]
-=== Producer ownership
-
-Supply the producer as configuration - `producerConfig`, the map you would hand `new KafkaProducer<>(config)` - and PC builds the producer itself, through `producerFactory` (default: exactly that constructor; override it to wrap or instrument the producer, keeping every key it is given).
-PC sets the `transactional.id` where it builds the producer: `pc-<L>-<group.id>-<uuid>`, where `<L>` is the decimal length of the consumer's `group.id`.
-The prefix `pc-<L>-<group.id>-` is stable for the group, so one prefixed TransactionalId ACL authorises every id PC derives for it, and the length field keeps one group's prefix from being a prefix of another's (`app` versus `app-x`).
-A `transactional.id` set in `producerConfig` does not take effect there, and PC says so at WARN naming the value it derived instead.
-Configuration values are never rendered into logs: PC's own log lines and `toString()` show an allow-list of non-secret keys and redact the rest.
-A factory must return a new producer on every call - PC may discard one and build another from the same configuration - and a factory that returns one it already returned, or drops the `transactional.id` it was given, is rejected at construction naming that contract.
-
-Supplying a finished `Producer` instance still works for every flow it worked for, and is deprecated: PC cannot read a finished producer's configuration back out, so it can never build another from it.
-PC logs one WARN at start-up naming the remedy.
-The instance option's removal is queued for `0.7.0.0`.
-
 [[producer-recovery]]
 === Producer recovery
 
 Where PC built the producer, a producer the broker reports invalid is replaced and the work its aborted transaction discarded is processed again, so processing continues instead of the instance stopping.
+The replacement is built from the same configuration, `transactional.id` included, and initialising it is what fences the producer it replaces; a finished `Producer` instance carries no configuration to rebuild from, so on that path these conditions keep their earlier behaviour.
 Recovery is a transactional-mode mechanism: it runs from the transactional commit loop, so in the consumer-commit modes a PC-built producer is not recovered and these conditions surface as the produce failures they were before.
 To make the replay possible, transactional mode keeps every completed record until the commit that carries it succeeds - on both producer paths - so memory held per partition grows with `commitInterval`; the default interval keeps it to a fraction of a second of throughput.
 The conditions PC treats this way are `ProducerFencedException`, `InvalidProducerEpochException`, `InvalidPidMappingException`, `OutOfOrderSequenceException` (and so `UnknownProducerIdException`) and `CommitFailedException`, on both the commit path and the produce path, unwrapped from the `ExecutionException` a send future raises and the `KafkaException` the client wraps a stored error in.

docs/data/roadmap.yaml

@@ -190,10 +190,10 @@ entries:
     stage_detail: >-
       Implemented on astubbs#410 to the plan's Definition of Done: PC builds its own transactional producer
       from configuration, detects the broker-reported conditions on both paths, aborts, replays the
-      completed-but-uncommitted ledger, and rebuilds under the same derived id; proven at the unit level
+      completed-but-uncommitted ledger, and rebuilds under the same transactional.id; proven at the unit level
       (fenced-during-commit was red before the change) and on a real coordinator by ProducerFencingRecoveryIT,
       with the recovery guarantee registered as a transactional claim. Not yet run in anger, so not
-      "implemented"; the producer-instance path keeps its old behaviour and is deprecated.
+      "implemented"; the producer-instance path keeps its old behaviour.
     why_now: >-
       Fencing is an ordinary consequence of a rebalance under transactions, and a broker can expire an
       idle producer id outright; either way PC dies on the commit path and spins on the produce path.
@@ -202,7 +202,7 @@ entries:
     done_when: >-
       A producer the broker has invalidated is replaced and processing continues, with no offset from
       the aborted transaction committed and its work processed again. Applies where PC builds the
-      producer; supplying a Producer instance is deprecated and keeps today's behaviour.
+      producer; supplying a Producer instance keeps today's behaviour.
     tracking: astubbs#225
     related: astubbs#411
     pull_request: astubbs#410

docs/inflight/core-recoverable-producer-fencing.md

@@ -4,6 +4,19 @@
 <!-- inflight-impact: crash -->
 
 
+## In flight as a stack of three PRs (2026-09-03)
+
+<!-- post-merge: checked-begin -->
+- astubbs/parallel-consumer#426 - `producerConfig`: PC builds its producer from configuration with the
+  default constructor, the caller's `transactional.id` included. The only rung recovery needs, because
+  re-initialising a replacement under the same id is what fences the producer it replaces.
+- astubbs/parallel-consumer#410 - recovery: detection, the aborted-transaction ledger and replay, the
+  replacement with backoff, observability, the broker IT. Stacked on astubbs#426.
+- astubbs/parallel-consumer#420 - the rest of producer ownership: a derived prefix-free
+  `transactional.id`, a `ProducerFactory` with an enforced contract, configuration redaction, deprecating
+  the instance option, migrating the examples. Stacked on astubbs#410.
+<!-- post-merge: checked-end -->
+
 <!-- post-merge: checked -->
 The feature was implemented by astubbs/parallel-consumer#410 against the plan
 `docs/plans/2026-09-02-001-feat-recoverable-producer-fencing-plan.md`, which owns the requirements, the

docs/inflight/issue-response-225.md

@@ -23,9 +23,8 @@ the shape to copy, but it is not the "abort, then rejoin" this issue described.
 **"Rejoin" was not the hard part; the producer was.** Recovery needs a *new* producer, and PC cannot
 build one: `ParallelConsumerOptions` holds a finished `Producer` instance, and a `KafkaProducer`'s
 configuration cannot be read back out of it. So the change is an ownership change — PC takes producer
-configuration and builds the producer through an overridable factory, Kafka Streams' `KafkaClientSupplier`
-shape — not an exception swap. The producer-instance option stays, deprecated, without recovery, with
-its removal queued for the major after the one that deprecates it.
+configuration and builds the producer itself, the shape of Kafka Streams' `KafkaClientSupplier` — not an
+exception swap. The producer-instance option stays, without recovery.
 
 **"Whether rejoin is expressible in PC's lifecycle" — the part this issue said to investigate first —
 is answered, and no state-machine addition is needed.** The produce/commit lock pair already gives the

docs/inflight/issue-response-411.md

@@ -49,10 +49,10 @@ Two honest caveats:
 
 - **It will not reach your configuration unchanged.** Recovery requires Parallel Consumer to build the
   producer itself, from configuration, so it can build a replacement. Supplying a `Producer` instance —
-  the only option today, and what you were doing — stays supported and deprecated, but cannot recover,
+  the only option today, and what you were doing — stays supported, but cannot recover,
   because a `KafkaProducer`'s configuration cannot be read back out of it.
 - **The loop itself is a separate defect**, tracked in
-  `docs/inflight/bug-411-wrapped-send-failure-spins-forever.md`. Whether to close it on the deprecated
+  `docs/inflight/bug-411-wrapped-send-failure-spins-forever.md`. Whether to close it on the instance
   path as well — so the condition at least terminates rather than spinning — is an open decision.
 
 Thanks for the original report and for the fix attempt. The diagnosis in it was right; it was the

docs/plans/2026-09-02-001-feat-recoverable-producer-fencing-plan.md

@@ -11,6 +11,8 @@ execution: code
 
 # Recoverable Producer Fencing - Plan
 
+> **Re-cut into a stack, 2026-09-03.** The producer-ownership half of this plan ships in stages rather than in one PR: `producerConfig` with the default constructor is the base rung (astubbs/parallel-consumer#426); recovery (astubbs/parallel-consumer#410) stacks on that rung alone and rebuilds under the `transactional.id` the caller sets; the derived prefix-free id, the `ProducerFactory` and its contract, configuration redaction, and the deprecation of the instance option with the example migration are astubbs/parallel-consumer#420, stacked above recovery. Where the text below says the factory or the derived id is what recovery rebuilds through, read: the same configuration, id included - the mechanisms and decisions are unchanged, only the order they land in.
+
 ## Goal Capsule
 
 **Objective.** Give Parallel Consumer a transactional producer it owns, so that when the broker tells it the producer is no longer usable, PC closes that producer, builds a replacement, and carries on — instead of spinning forever on the produce path or shutting the instance down on the commit path. Tracked as astubbs#225.
@@ -331,6 +333,7 @@ U1 → U2 → U3 → U5 → U6 → U7; U4 has no dependencies and may be built a
 - **Requirements:** R1, R2, R3, R7, R16, R17, R19, R21; KTD1, KTD9.
 - **Dependencies:** none.
 - **Files:** `parallel-consumer-core/src/main/java/bz/stub/parallelconsumer/ParallelConsumerOptions.java`, new `parallel-consumer-core/src/main/java/bz/stub/parallelconsumer/ProducerFactory.java`, new `parallel-consumer-core/src/main/java/bz/stub/parallelconsumer/internal/ProducerConfigRedaction.java`, `docs/refactoring.md` (the entry exists; only re-check the release it names), tests in new `parallel-consumer-core/src/test/java/bz/stub/parallelconsumer/ParallelConsumerOptionsProducerConfigTest.java` and new `parallel-consumer-core/src/test/java/bz/stub/parallelconsumer/internal/ProducerConfigRedactionTest.java`.
+  <!-- file-refs: N/A - the factory, derivation and redaction files are astubbs#420's, stacked above this PR; see the re-cut note at the head of this plan -->
 - **Approach:**
   1. Add `producerConfig` (`Map<String, Object>`) and `producerFactory` (`ProducerFactory<K, V>`, `@Builder.Default` to a `KafkaProducer` constructor) fields; `isProducerSupplied()` becomes "instance or configuration present".
   2. `validate()`: both `producer` and `producerConfig` set fails with a message naming `Fields.producer` and `Fields.producerConfig` (R17); transactional mode with neither fails as today; `producer` set alone logs the single R19 WARN naming `producerConfig` plus `producerFactory`, the absence of recovery, and the removal release.
@@ -353,6 +356,7 @@ U1 → U2 → U3 → U5 → U6 → U7; U4 has no dependencies and may be built a
 - **Requirements:** R2, R4, R5, R6; KTD2, KTD8.
 - **Dependencies:** U1.
 - **Files:** `parallel-consumer-core/src/main/java/bz/stub/parallelconsumer/internal/PCModule.java`, `parallel-consumer-core/src/main/java/bz/stub/parallelconsumer/internal/ProducerWrapper.java`, new `parallel-consumer-core/src/main/java/bz/stub/parallelconsumer/internal/TransactionalIdDerivation.java`, `parallel-consumer-core/src/test/java/bz/stub/parallelconsumer/internal/PCModuleTestEnv.java` (constructor change follow-through), tests in new `parallel-consumer-core/src/test/java/bz/stub/parallelconsumer/internal/TransactionalIdDerivationTest.java` and new `parallel-consumer-core/src/test/java/bz/stub/parallelconsumer/internal/PcBuiltProducerTest.java`.
+  <!-- file-refs: N/A - the factory, derivation and redaction files are astubbs#420's, stacked above this PR; see the re-cut note at the head of this plan -->
 - **Approach:**
   1. `TransactionalIdDerivation`: `prefixFor(groupId)` returns `pc-<L>-<groupId>-`; `derive(groupId, instanceUuid)` appends the UUID; `resolve(producerConfig, commitMode, groupId, uuid)` returns a copy of the config with the id set (transactional mode) or removed (otherwise), emitting the R5 WARN when the caller had set one.
   2. `PCModule`: one `UUID` per module instance; `producerWrap()` builds from the instance when present, otherwise resolves the config and calls the factory; new `replacementProducerWrap()` returns `Optional<Supplier<ProducerWrapper<K, V>>>`, present only on the configuration path, each call resolving the same config and calling the factory again.

docs/refactoring.md

@@ -118,13 +118,6 @@ refactors below, which are non-breaking and can land at any point in any line.
   (`public void setCommitInterval`, `private final Duration defaultMessageRetryDelay`,
   `isUsingTransactionalProducer`) **and retire the temporary Kafka-compat work-around flag**
   (`ignoreReflectiveAccessExceptionsForAutoCommitDisabledCheck`) - `ParallelConsumerOptions.java`.
-- **Remove the `producer` instance option**, once the configuration-plus-factory path deprecating it
-  has landed - `ParallelConsumerOptions.java`. Queued rather than left open-ended because the two
-  paths are not equivalent: only the PC-built one can rebuild a producer the broker has invalidated,
-  so keeping both means every future change to producer handling is written twice, once on a path
-  that recovers and once on a path that cannot. Deprecated by
-  [`docs/plans/2026-09-02-001-feat-recoverable-producer-fencing-plan.md`](plans/2026-09-02-001-feat-recoverable-producer-fencing-plan.md)
-  (astubbs#225); the deprecation javadoc names the major AFTER the release that ships the deprecation, so the option is not removed in the same version that deprecates it.
 - **Remove the JStream API** (deprecate first) - design ref
   `origin/refactor/deprecate-jstream` @8a8f6508.
 - **Rename the enum to the standard pattern** (public enum rename) -