Document the unanswered-commit case in transactional persistence

The transactional mode's one failure that is not a rejection: an
unanswered commit. If `commitTransaction` times out client-side, the
broker may have committed anyway.

Nothing local resolves it - retrying blocks for up to `max.block.ms`
inside the poll cycle, and `abortTransaction` throws rather than aborting
while a commit is unresolved, a throw `commitBatch` already discards.

Nothing is lost either way: the input offset rides the same transaction,
so snapshot and offset commit together or not at all. Under the default
`ignorePersistErrors = false` the failure tears the module down and the
next producer's `initTransactions` settles the transaction broker-side.
Under `ignorePersistErrors = true` it is swallowed and the producer stays
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.

GroupCommitSpec gains a case pinning that the commit is never retried in
place. No production code changes.

The contract and a measurement of it are pinned as ext(K16) in the
verification research corpus; that harness is not checked in, since it
pins a kafka-clients contract rather than kafka-flow behavior.

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],