Document the unanswered-commit case in transactional persistence

`KafkaSnapshotWriteDatabase.GroupCommit.commitBatch` aborts on every
failure. An adversarial review of the transactional mode reported that
this is a defect on one path - an ambiguous `commitTransaction` timeout,
where the broker may have committed anyway - and that the abort itself
leaves the producer unusable.

The first half is right; the second is not. kafka-clients throws
IllegalStateException from `abortTransaction` before any state transition
while a commit is unresolved, and `commitBatch` already discards that
throw via `voidError`. The abort is inert: what spends the producer is the
unacked commit itself, which the abort neither causes nor worsens. So
narrowing the abort to definitive failures would change no outcome.

Retrying the commit in place is the only change that would, and it is not
worth it - each attempt blocks up to a user-configured `max.block.ms`
inside the poll cycle, trading a loud failure for a possible silent
`max.poll.interval.ms` eviction. With the default
`ignorePersistErrors = false` the surfaced failure tears the module down,
and the next producer's `initTransactions` settles the transaction
broker-side.

No production change follows. What was missing is the documentation of the
case, and one pin:

- docs/persistence.md gains the failure that is not a rejection but an
  unanswered commit: nothing local resolves it, nothing is lost (the input
  offset rides the same transaction, so snapshot and offset commit
  together or not at all), and `ignorePersistErrors = true` swallows it
  too, leaving the partition frozen and self-consistent - not corrupted -
  until the next rebalance.
- GroupCommitSpec pins that the commit surfaces as itself and is never
  retried in place. That is the half a future edit could plausibly break,
  and it needs no broker.

The client contract behind this is pinned as ext(K16) in the verification
research corpus, together with a broker-paused measurement: the abort
aborts nothing, the transaction still lands, and the producer refuses
every later transaction until rebuilt. That harness is deliberately not
checked in - it pins a kafka-clients contract rather than kafka-flow
behavior, and pauses the broker container to get there.

Author: tobiajo

docs/persistence.md

@@ -63,6 +63,16 @@ You do not catch the rejection yourself; it is handled for you:
 Either way the rejected write does not land and no offset is committed for it, so the new owner
 replays the affected events.
 
+One **Kafka** transactional-mode failure is not a rejection but an *unanswered* commit: if
+`commitTransaction` times out client-side, the broker may have committed anyway. The write is reported
+as failed and nothing local can resolve it — retrying blocks for up to `max.block.ms` inside the poll
+cycle, and aborting is inert while a commit is unresolved. Nothing is lost: the input offset rides the
+same transaction, so snapshot and offset commit together or not at all, and the failure tears the flow
+down so the rebuilt module's `initTransactions` settles it. Note that `ignorePersistErrors = true`
+swallows this too and leaves the producer unusable for the rest of the assignment, so the partition
+stops persisting *and* stops advancing offsets until the next rebalance — frozen and self-consistent,
+not corrupted.
+
 ### Transactional snapshot writes (Kafka)
 
 **EXPERIMENTAL** — use at your own risk: the mechanism is design-verified but not yet proven in

persistence-kafka/src/test/scala/com/evolutiongaming/kafka/flow/kafkapersistence/GroupCommitSpec.scala

@@ -12,6 +12,7 @@ import com.evolutiongaming.skafka.consumer.ConsumerGroupMetadata
 import com.evolutiongaming.skafka.producer.{Producer, ProducerRecord, RecordMetadata}
 import com.evolutiongaming.skafka.{Offset, OffsetAndMetadata, Partition, ToBytes, Topic, TopicPartition}
 import munit.FunSuite
+import org.apache.kafka.common.errors.TimeoutException
 
 import scala.concurrent.duration.*
 
@@ -235,6 +236,31 @@ class GroupCommitSpec extends FunSuite {
     }
     test.unsafeRunSync()
   }
+
+  // a client-side commit timeout is ambiguous (the broker may have committed) and surfaces as itself; retrying in
+  // place is the tempting "fix" this guards against, since each attempt blocks up to max.block.ms inside a poll
+  // cycle. See docs/persistence.md. The abort's inertness needs a real client - this fake's abort succeeds.
+  test("an ambiguously timed-out commit surfaces as itself, unretried") {
+    val test = for {
+      events   <- Ref.of[IO, Vector[Event]](Vector.empty)
+      attempts <- Ref.of[IO, Int](0)
+      boom      = new TimeoutException("commit timed out")
+      commit    = attempts.update(_ + 1) *> IO.raiseError[Unit](boom)
+      tx <- buildTransactional(
+        recordingProducer(events, commitTransactionOf = commit.some),
+        ConsumerGroupMetadata.Empty.some,
+        maxWritesPerTransaction = 256,
+      )
+      result <- tx.writeDatabase.persist(kafkaKey("key1"), "state-1").attempt
+      tries  <- attempts.get
+    } yield {
+      // told it failed, though the broker may have committed - only a rebuilt producer resolves that
+      assertEquals(result.left.toOption, boom.some)
+      // exactly once: never retried in place
+      assertEquals(tries, 1)
+    }
+    test.unsafeRunSync()
+  }
 }
 
 object GroupCommitSpec {
@@ -251,19 +277,24 @@ object GroupCommitSpec {
   private val CommitBoom = new RuntimeException("commit boom")
 
   /** A `Producer` that records the transactional calls into `events` and delegates everything else to a no-op producer
-    * (which also fabricates the `RecordMetadata` for `send`). `failCommit` makes `commitTransaction` raise.
+    * (which also fabricates the `RecordMetadata` for `send`). `failCommit` makes `commitTransaction` raise;
+    * `commitTransactionOf` replaces the whole `commitTransaction` behavior (the override must record `Event.Commit`
+    * itself when it succeeds).
     */
   def recordingProducer(
     events: Ref[IO, Vector[Event]],
-    failCommit: Boolean          = false,
-    onBeginTransaction: IO[Unit] = IO.unit,
+    failCommit: Boolean                   = false,
+    onBeginTransaction: IO[Unit]          = IO.unit,
+    commitTransactionOf: Option[IO[Unit]] = None,
   ): Producer[IO] = {
     val base = Producer.empty[IO]
     new Producer[IO] {
-      def initTransactions: IO[Unit]  = base.initTransactions
-      def beginTransaction: IO[Unit]  = events.update(_ :+ Event.Begin) *> onBeginTransaction
-      def commitTransaction: IO[Unit] = if (failCommit) IO.raiseError(CommitBoom) else events.update(_ :+ Event.Commit)
-      def abortTransaction: IO[Unit]  = events.update(_ :+ Event.Abort)
+      def initTransactions: IO[Unit] = base.initTransactions
+      def beginTransaction: IO[Unit] = events.update(_ :+ Event.Begin) *> onBeginTransaction
+      def commitTransaction: IO[Unit] = commitTransactionOf.getOrElse {
+        if (failCommit) IO.raiseError(CommitBoom) else events.update(_ :+ Event.Commit)
+      }
+      def abortTransaction: IO[Unit] = events.update(_ :+ Event.Abort)
 
       def sendOffsetsToTransaction(
         offsets: NonEmptyMap[TopicPartition, OffsetAndMetadata],