shibuya-kafka-adapter 0.9.0.1 → 0.9.1.0

raw patch · 8 files changed

+776/−112 lines, 8 filesdep ~effectful-coredep ~shibuya-corePVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: effectful-core, shibuya-core

API changes (from Hackage documentation)

Files

• CHANGELOG.md +37/−0
• shibuya-kafka-adapter.cabal +5/−5
• src/Shibuya/Adapter/Kafka.hs +19/−23
• src/Shibuya/Adapter/Kafka/Internal.hs +239/−53
• test/Kafka/TestEnv.hs +20/−0
• test/Shibuya/Adapter/Kafka/AckHandleTest.hs +243/−25
• test/Shibuya/Adapter/Kafka/AdapterTest.hs +2/−2
• test/Shibuya/Adapter/Kafka/IntegrationTest.hs +211/−4

@@ -1,5 +1,42 @@ # Changelog +## 0.9.1.0 — 2026-09-21++### Bug Fixes++- Preserve the earliest unresolved delivery when multiple buffered Kafka+  callbacks request retry. Delivery tokens distinguish an original callback+  from the replay that may resolve its recovery boundary, so later buffered or+  duplicate callbacks cannot advance the stored offset past unresolved work.+- Throw exhausted acknowledgement operations synchronously from the finalizer+  boundary. Core can now retain them as processor failures even after ingestion+  has stopped instead of depending on a future source poll of the fatal slot.+  The public `KafkaAcknowledgementException` carries the underlying+  `KafkaError` for callers that need to classify it.+- Fence callbacks retained across a revoke/assign cycle when callers install+  `kafkaRebalanceHandler`. Assignment generations prevent an old owner from+  storing, seeking, or pausing a partition after reassignment.+- Serialize each delivery finalizer with exception-safe ownership. Successful+  duplicates are no-ops, while failed or cancelled attempts remain retryable.++### Other Changes++- Require `shibuya-core ^>=0.10.0.0` across the library, tests, benchmark,+  and example so the adapter is certified against the lifecycle-remediation+  release candidate. Keep all three repository packages on version `0.9.1.0`.+- Add deterministic reference-model, cancellation, timeout, terminal-failure,+  repeated-shutdown, buffered-retry, restart, and actual-reassignment coverage.+- Make the live restart fixture stream delivery identities to external files,+  emit a reconciled per-delivery ledger, and fail on duplicates, missing+  deliveries, unexpected deliveries, or malformed payloads without retaining+  the identity set in the measured process heap.+- Keep the documented absence of a DLQ producer: `AckDeadLetter` still warns+  and stores the offset deliberately.+- Exclude `effectful-core` 2.7.0.0 through 2.7.1.0 from the library, tests,+  benchmark, and examples because upstream records a per-operation performance+  regression for dynamically dispatched effects. The 2.6 family and 2.7.1.1 or+  later remain accepted.+ ## 0.9.0.1 — 2026-09-15  ### Other Changes

@@ -1,6 +1,6 @@ cabal-version: 3.12 name: shibuya-kafka-adapter-version: 0.9.0.1+version: 0.9.1.0 synopsis: Kafka adapter for the Shibuya queue processing framework description:   A Shibuya adapter that integrates with Apache Kafka via kafka-effectful@@ -61,13 +61,13 @@     base ^>=4.21.0.0,     bytestring ^>=0.12,     containers ^>=0.7,-    effectful-core >=2.6.1 && <2.8,+    effectful-core (>=2.6.1 && <2.7) || (>=2.7.1.1 && <2.8),     hs-opentelemetry-api ^>=1.0,     hs-opentelemetry-semantic-conventions ^>=1.40,     hw-kafka-client >=5.3 && <6,     hw-kafka-streamly ^>=0.2,     kafka-effectful ^>=0.3.1.0,-    shibuya-core ^>=0.9.0.1,+    shibuya-core ^>=0.10.0.0,     stm ^>=2.5,     streamly ^>=0.11,     streamly-core ^>=0.3,@@ -112,14 +112,14 @@     base ^>=4.21.0.0,     bytestring,     containers,-    effectful-core,+    effectful-core (>=2.6.1 && <2.7) || (>=2.7.1.1 && <2.8),     hs-opentelemetry-api ^>=1.0,     hs-opentelemetry-semantic-conventions ^>=1.40,     hw-kafka-client,     kafka-effectful,     process,     random,-    shibuya-core ^>=0.9.0.1,+    shibuya-core ^>=0.10.0.0,     shibuya-kafka-adapter,     stm,     streamly,

@@ -32,7 +32,8 @@ -- 3. On @AckOk@, the offset is stored locally; auto-commit or consumer close --    later flushes stored offsets to the broker. -- 4. On @AckRetry@, the offset is not stored. The adapter seeks the partition---    back to the failed message so Kafka can redeliver it.+--    back to the earliest unresolved delivery so Kafka can redeliver it. Later+--    buffered callbacks cannot replace or commit past that recovery boundary. -- 5. On @AckDeadLetter@, the offset is stored after a loud stderr warning. -- 6. On @AckHalt@, the partition is paused and offset is not stored. --@@ -65,9 +66,10 @@ -- are filtered out of the poll stream. Any error that survives that filter is -- fatal by construction (for example, an SSL handshake failure, an authentication -- failure, or an invalid broker configuration) and terminates the stream by--- throwing through the 'Effectful.Error.Static.Error' @KafkaError@ effect. The--- caller observes the failure by receiving a @Left err@ from the--- @runError \@KafkaError@ scope around 'Shibuya.App.runApp'.+-- throwing through the 'Effectful.Error.Static.Error' @KafkaError@ effect.+-- Exhausted acknowledgement operations instead throw a synchronous typed+-- exception at the finalizer boundary; Shibuya records that as a processor+-- failure even if ingestion has already ended. -- -- == AckHalt Partition Pause Semantics --@@ -84,15 +86,15 @@ -- 'kafkaRebalanceHandler' is optional. Install it with -- @Kafka.Consumer.setCallback (Kafka.Consumer.rebalanceCallback (kafkaRebalanceHandler state))@ -- before creating the consumer when you want stderr visibility into assignment--- changes and eager cleanup of retry barriers for revoked partitions. Without it,--- the seek barrier still self-heals when messages are finalized at or below the--- barrier offset. Cooperative rebalance fencing of in-flight work is outside this--- adapter's scope.+-- changes and assignment-generation fencing of callbacks retained by a revoked+-- owner. Without it, retry recovery remains safe within one assignment, but stale+-- callbacks are not fenced across ownership changes. module Shibuya.Adapter.Kafka   ( -- * Adapter     kafkaAdapter,     kafkaAdapterWith,     KafkaAdapterState,+    KafkaAcknowledgementException (..),     newKafkaAdapterState,     kafkaRebalanceHandler, @@ -116,8 +118,6 @@ import Control.Concurrent.STM (atomically, writeTVar) import Control.Monad.IO.Class (liftIO) import Data.ByteString (ByteString)-import Data.IORef (atomicModifyIORef')-import Data.Map.Strict qualified as Map import Data.Set qualified as Set import Data.Text qualified as Text import Effectful (Eff, IOE, (:>))@@ -126,10 +126,10 @@ import Kafka.Consumer.Types (ConsumerGroupId (..), OffsetCommit (..), RebalanceEvent (..)) import Kafka.Consumer.Types qualified as KC import Kafka.Effectful.Consumer.Effect (KafkaConsumer, commitAllOffsets, subscription)-import Kafka.Types (BatchSize (..), BrokerAddress (..), KafkaError (..), PartitionId, Timeout (..), TopicName (..))+import Kafka.Types (BatchSize (..), BrokerAddress (..), KafkaError (..), Timeout (..), TopicName (..)) import Shibuya.Adapter (Adapter (..)) import Shibuya.Adapter.Kafka.Config (KafkaAdapterConfig (..), defaultConfig)-import Shibuya.Adapter.Kafka.Internal (KafkaAdapterState (..), dropStaleRecords, ingestedStream, kafkaSource, mkIngested, newKafkaAdapterState, withConsumerLock)+import Shibuya.Adapter.Kafka.Internal (KafkaAcknowledgementException (..), KafkaAdapterState (..), dropStaleRecords, ingestedStream, kafkaSource, markPartitionsAssigned, markPartitionsRevoked, mkIngested, newKafkaAdapterState, withConsumerLock) import System.IO (hPutStrLn, stderr)  -- | Create a Kafka adapter with the given configuration.@@ -208,8 +208,9 @@ -- -- Install with 'Kafka.Consumer.setCallback' and -- 'Kafka.Consumer.rebalanceCallback' before creating the consumer. The callback--- logs every rebalance event to stderr and clears pending retry barriers for--- revoked partitions. It does not fence in-flight work.+-- logs every rebalance event to stderr, clears local retry state on revocation,+-- and advances an assignment generation so late callbacks from a revoked owner+-- cannot store, seek, or pause a later assignment. kafkaRebalanceHandler ::   KafkaAdapterState ->   KC.KafkaConsumer ->@@ -218,12 +219,7 @@ kafkaRebalanceHandler state _consumer event = do   hPutStrLn stderr $ "[shibuya-kafka-adapter] rebalance: " <> show event   case event of-    RebalanceRevoke revoked ->-      clearRevokedBarriers revoked-    _ ->-      pure ()-  where-    clearRevokedBarriers :: [(TopicName, PartitionId)] -> IO ()-    clearRevokedBarriers revoked =-      atomicModifyIORef' state.seekBarrier $ \barriers ->-        (foldr Map.delete barriers revoked, ())+    RebalanceBeforeAssign assigned -> markPartitionsAssigned state assigned+    RebalanceAssign assigned -> markPartitionsAssigned state assigned+    RebalanceBeforeRevoke revoked -> markPartitionsRevoked state revoked+    RebalanceRevoke revoked -> markPartitionsRevoked state revoked

@@ -3,7 +3,10 @@ module Shibuya.Adapter.Kafka.Internal   ( -- * Adapter State     KafkaAdapterState (..),+    KafkaAcknowledgementException (..),     newKafkaAdapterState,+    markPartitionsAssigned,+    markPartitionsRevoked,     withConsumerLock,      -- * Stream Construction@@ -22,13 +25,15 @@ import Control.Concurrent (threadDelay) import Control.Concurrent.MVar (MVar, newMVar, putMVar, takeMVar) import Control.Concurrent.STM (TVar, newTVarIO, readTVarIO)+import Control.Monad (unless, when) import Data.ByteString (ByteString) import Data.Function ((&))-import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef)+import Data.IORef (IORef, atomicModifyIORef', atomicWriteIORef, newIORef, readIORef) import Data.Map.Strict (Map) import Data.Map.Strict qualified as Map import Data.Text qualified as Text import Data.Time.Clock (NominalDiffTime)+import Data.Word (Word64) import Effectful (Eff, IOE, (:>)) import Effectful qualified import Effectful.Error.Static (Error, catchError, throwError)@@ -55,10 +60,60 @@  type PartitionKey = (TopicName, PartitionId) +newtype DeliveryToken = DeliveryToken Word64+  deriving stock (Eq, Ord, Show)++newtype AssignmentGeneration = AssignmentGeneration Word64+  deriving stock (Eq, Show)++data RetryBarrier = RetryBarrier+  { offset :: !Offset,+    retryToken :: !DeliveryToken+  }++data PartitionAckState = PartitionAckState+  { barrier :: !(Maybe RetryBarrier),+    validFromToken :: !DeliveryToken,+    assignmentGeneration :: !AssignmentGeneration,+    assigned :: !Bool+  }++data AckState = AckState+  { nextDeliveryToken :: !Word64,+    partitions :: !(Map PartitionKey PartitionAckState)+  }++data DeliveryAttempt = DeliveryAttempt+  { partition :: !PartitionKey,+    recordOffset :: !Offset,+    token :: !DeliveryToken,+    generation :: !AssignmentGeneration+  }++-- | A Kafka operation failed after the adapter's bounded retry budget. This is+-- a synchronous exception so Shibuya's finalizer boundary can retry it and,+-- when exhausted, retain a processor failure with the delivery identity.+newtype KafkaAcknowledgementException = KafkaAcknowledgementException KafkaError+  deriving stock (Show)++instance Exception.Exception KafkaAcknowledgementException++initialPartitionAckState :: PartitionAckState+initialPartitionAckState =+  PartitionAckState+    { barrier = Nothing,+      validFromToken = DeliveryToken 0,+      assignmentGeneration = AssignmentGeneration 0,+      assigned = True+    }++initialAckState :: AckState+initialAckState = AckState {nextDeliveryToken = 1, partitions = Map.empty}+ -- | Mutable state shared by the source stream and ack handles. data KafkaAdapterState = KafkaAdapterState   { shutdownVar :: !(TVar Bool),-    seekBarrier :: !(IORef (Map PartitionKey Offset)),+    ackState :: !(IORef AckState),     fatalError :: !(IORef (Maybe KafkaError)),     -- | Serializes every librdkafka consumer operation. Under     --     'Shibuya.App.runApp' the consumer handle is shared between the ingester@@ -78,10 +133,43 @@ newKafkaAdapterState =   KafkaAdapterState     <$> newTVarIO False-    <*> newIORef Map.empty+    <*> newIORef initialAckState     <*> newIORef Nothing     <*> newMVar () +-- | Mark partitions assigned to this adapter generation. A revoke/assign cycle+-- advances the generation, so ack handles retained from the previous owner can+-- no longer store, seek, or pause the newly assigned partition.+markPartitionsAssigned :: KafkaAdapterState -> [PartitionKey] -> IO ()+markPartitionsAssigned state = setPartitionsAssigned state True++-- | Mark partitions revoked and discard only the local retry ledger. Kafka's+-- committed offset remains the durable recovery boundary; callbacks from the+-- old generation are fenced by the incremented assignment generation.+markPartitionsRevoked :: KafkaAdapterState -> [PartitionKey] -> IO ()+markPartitionsRevoked state = setPartitionsAssigned state False++setPartitionsAssigned :: KafkaAdapterState -> Bool -> [PartitionKey] -> IO ()+setPartitionsAssigned state isAssigned keys =+  atomicModifyIORef' state.ackState $ \ackState' ->+    let partitions' = foldr (Map.alter (Just . transition . maybe initialPartitionAckState id)) ackState'.partitions keys+     in (ackState' {partitions = partitions'}, ())+  where+    transition partitionState+      | partitionState.assigned == isAssigned =+          if isAssigned+            then partitionState+            else partitionState {barrier = Nothing}+      | otherwise =+          partitionState+            { barrier = Nothing,+              validFromToken = DeliveryToken 0,+              assignmentGeneration = nextGeneration partitionState.assignmentGeneration,+              assigned = isAssigned+            }++    nextGeneration (AssignmentGeneration generation') = AssignmentGeneration (generation' + 1)+ -- | Run a librdkafka consumer operation while holding the shared consumer -- lock, guaranteeing no other consumer call runs concurrently on the same -- handle. See 'consumerLock' for why this is mandatory. The lock is always@@ -157,10 +245,12 @@   Stream.filterM $ \case     Left _ -> pure True     Right cr -> do-      barriers <- Effectful.liftIO $ readIORef state.seekBarrier-      pure $ case Map.lookup (partitionKey cr) barriers of-        Nothing -> True-        Just barrierOff -> cr.crOffset <= barrierOff+      ackState' <- Effectful.liftIO $ readIORef state.ackState+      let partitionState = Map.findWithDefault initialPartitionAckState (partitionKey cr) ackState'.partitions+      pure $+        partitionState.assigned && case partitionState.barrier of+          Nothing -> True+          Just retryBarrier -> cr.crOffset <= retryBarrier.offset  -- | Create an 'AckHandle' for a single 'ConsumerRecord'. --@@ -175,35 +265,79 @@   KafkaAdapterState ->   KafkaAdapterConfig ->   ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->-  AckHandle es-mkAckHandle state config cr = AckHandle $ \case-  AckOk ->-    ackAttempt state (storeGuarded state cr)+  Eff es (AckHandle es)+mkAckHandle state config cr = do+  attempt <- Effectful.liftIO $ registerDelivery state cr+  finalizerLock <- Effectful.liftIO $ newMVar ()+  completed <- Effectful.liftIO $ newIORef False+  pure $ AckHandle $ \decision ->+    Exception.bracket_+      (Effectful.liftIO (takeMVar finalizerLock))+      (Effectful.liftIO (putMVar finalizerLock ()))+      $ do+        isCompleted <- Effectful.liftIO $ readIORef completed+        unless isCompleted $ do+          finalizeAttempt state config attempt cr decision+          Effectful.liftIO $ atomicWriteIORef completed True++registerDelivery :: KafkaAdapterState -> ConsumerRecord k v -> IO DeliveryAttempt+registerDelivery state cr =+  atomicModifyIORef' state.ackState $ \ackState' ->+    let partition = partitionKey cr+        partitionState = Map.findWithDefault initialPartitionAckState partition ackState'.partitions+        token = DeliveryToken ackState'.nextDeliveryToken+        attempt =+          DeliveryAttempt+            { partition,+              recordOffset = cr.crOffset,+              token,+              generation = partitionState.assignmentGeneration+            }+        ackState'' =+          ackState'+            { nextDeliveryToken = ackState'.nextDeliveryToken + 1,+              partitions = Map.insert partition partitionState ackState'.partitions+            }+     in (ackState'', attempt)++finalizeAttempt ::+  (KafkaConsumer :> es, Error KafkaError :> es, IOE :> es) =>+  KafkaAdapterState ->+  KafkaAdapterConfig ->+  DeliveryAttempt ->+  ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->+  AckDecision ->+  Eff es ()+finalizeAttempt state config attempt cr = \case+  AckOk -> storeGuarded state attempt cr (pure ())   AckRetry (RetryDelay delay) -> do     Effectful.liftIO $ delayRetry delay-    Effectful.liftIO $-      atomicModifyIORef' state.seekBarrier $ \barriers ->-        (Map.insert (partitionKey cr) cr.crOffset barriers, ())-    ackAttempt state $-      withConsumerLock state $-        seekPartitions-          [ TopicPartition-              { tpTopicName = cr.crTopic,-                tpPartition = cr.crPartition,-                tpOffset = PartitionOffset (unOffset cr.crOffset)-              }-          ]-          (boundedLockTimeout config.pollTimeout)-  AckDeadLetter reason -> do-    Effectful.liftIO $-      hPutStrLn stderr $-        "[shibuya-kafka-adapter] WARNING: dead-lettered message DROPPED (no DLQ producer): "-          <> show (cr.crTopic, cr.crPartition, cr.crOffset)-          <> " reason="-          <> Text.unpack (renderDeadLetterReason reason)-    ackAttempt state (storeGuarded state cr)-  AckHalt _ ->-    ackAttempt state (withConsumerLock state (pausePartitions [(cr.crTopic, cr.crPartition)]))+    mbTarget <- Effectful.liftIO $ recordRetry state attempt+    case mbTarget of+      Nothing -> pure ()+      Just target ->+        ackAttempt state $+          withConsumerLock state $+            seekPartitions+              [ TopicPartition+                  { tpTopicName = cr.crTopic,+                    tpPartition = cr.crPartition,+                    tpOffset = PartitionOffset (unOffset target)+                  }+              ]+              (boundedLockTimeout config.pollTimeout)+  AckDeadLetter reason ->+    storeGuarded state attempt cr $+      Effectful.liftIO $+        hPutStrLn stderr $+          "[shibuya-kafka-adapter] WARNING: dead-lettered message DROPPED (no DLQ producer): "+            <> show (cr.crTopic, cr.crPartition, cr.crOffset)+            <> " reason="+            <> Text.unpack (renderDeadLetterReason reason)+  AckHalt _ -> do+    isAccepted <- Effectful.liftIO $ deliveryIsAccepted state attempt+    when isAccepted $+      ackAttempt state (withConsumerLock state (pausePartitions [(cr.crTopic, cr.crPartition)]))  ackAttempt ::   (Error KafkaError :> es, IOE :> es) =>@@ -218,7 +352,9 @@     go attempt =       action `catchError` \_ err ->         if isFatal err || attempt >= maxAttempts-          then Effectful.liftIO $ recordFatalError state err+          then do+            Effectful.liftIO $ recordFatalError state err+            Exception.throwIO (KafkaAcknowledgementException err)           else do             Effectful.liftIO $ threadDelay retryDelayMicros             go (attempt + 1)@@ -230,21 +366,69 @@     Nothing -> (Just err, ())  storeGuarded ::-  (KafkaConsumer :> es, IOE :> es) =>+  (KafkaConsumer :> es, Error KafkaError :> es, IOE :> es) =>   KafkaAdapterState ->+  DeliveryAttempt ->   ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->+  Eff es () ->   Eff es ()-storeGuarded state cr = do-  shouldStore <--    Effectful.liftIO $-      atomicModifyIORef' state.seekBarrier $ \barriers ->-        case Map.lookup (partitionKey cr) barriers of-          Nothing -> (barriers, True)-          Just barrierOff-            | cr.crOffset <= barrierOff -> (Map.delete (partitionKey cr) barriers, True)-            | otherwise -> (barriers, False)-  if shouldStore then withConsumerLock state (storeOffsetMessage cr) else pure ()+storeGuarded state attempt cr beforeStore = do+  shouldStore <- Effectful.liftIO $ claimStore state attempt+  when shouldStore $ do+    beforeStore+    ackAttempt state (withConsumerLock state (storeOffsetMessage cr)) +recordRetry :: KafkaAdapterState -> DeliveryAttempt -> IO (Maybe Offset)+recordRetry state attempt =+  atomicModifyIORef' state.ackState $ \ackState' ->+    let partitionState = Map.findWithDefault initialPartitionAckState attempt.partition ackState'.partitions+     in if not (attemptIsAccepted partitionState attempt)+          then (ackState', Nothing)+          else+            let retryBarrier = case partitionState.barrier of+                  Nothing -> RetryBarrier attempt.recordOffset attempt.token+                  Just existing+                    | attempt.recordOffset < existing.offset -> RetryBarrier attempt.recordOffset attempt.token+                    | attempt.recordOffset == existing.offset && attempt.token > existing.retryToken ->+                        existing {retryToken = attempt.token}+                    | otherwise -> existing+                partitionState' = partitionState {barrier = Just retryBarrier}+                ackState'' = ackState' {partitions = Map.insert attempt.partition partitionState' ackState'.partitions}+             in (ackState'', Just retryBarrier.offset)++claimStore :: KafkaAdapterState -> DeliveryAttempt -> IO Bool+claimStore state attempt =+  atomicModifyIORef' state.ackState $ \ackState' ->+    let partitionState = Map.findWithDefault initialPartitionAckState attempt.partition ackState'.partitions+     in if not (attemptIsAccepted partitionState attempt)+          then (ackState', False)+          else case partitionState.barrier of+            Nothing -> (ackState', True)+            Just retryBarrier+              | attempt.recordOffset > retryBarrier.offset -> (ackState', False)+              | attempt.recordOffset < retryBarrier.offset -> (ackState', True)+              | attempt.token <= retryBarrier.retryToken -> (ackState', False)+              | otherwise ->+                  let partitionState' =+                        partitionState+                          { barrier = Nothing,+                            validFromToken = max partitionState.validFromToken attempt.token+                          }+                      ackState'' = ackState' {partitions = Map.insert attempt.partition partitionState' ackState'.partitions}+                   in (ackState'', True)++deliveryIsAccepted :: KafkaAdapterState -> DeliveryAttempt -> IO Bool+deliveryIsAccepted state attempt = do+  ackState' <- readIORef state.ackState+  let partitionState = Map.findWithDefault initialPartitionAckState attempt.partition ackState'.partitions+  pure $ attemptIsAccepted partitionState attempt++attemptIsAccepted :: PartitionAckState -> DeliveryAttempt -> Bool+attemptIsAccepted partitionState attempt =+  partitionState.assigned+    && partitionState.assignmentGeneration == attempt.generation+    && attempt.token >= partitionState.validFromToken+ partitionKey :: ConsumerRecord k v -> PartitionKey partitionKey cr = (cr.crTopic, cr.crPartition) @@ -261,11 +445,13 @@   KafkaAdapterState ->   KafkaAdapterConfig ->   ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->-  Ingested es (Maybe ByteString)-mkIngested state config cr =-  Core.mkIngested-    (consumerRecordToEnvelope cr)-    (mkAckHandle state config cr)+  Eff es (Ingested es (Maybe ByteString))+mkIngested state config cr = do+  ackHandle <- mkAckHandle state config cr+  pure $+    Core.mkIngested+      (consumerRecordToEnvelope cr)+      ackHandle  -- | Transform a poll stream of @Either KafkaError ConsumerRecord@ into a -- stream of 'Ingested'.@@ -280,10 +466,10 @@ -- injects a synthetic @Left@ without standing up a real consumer. ingestedStream ::   (Error KafkaError :> es) =>-  (ConsumerRecord (Maybe ByteString) (Maybe ByteString) -> Ingested es (Maybe ByteString)) ->+  (ConsumerRecord (Maybe ByteString) (Maybe ByteString) -> Eff es (Ingested es (Maybe ByteString))) ->   Stream (Eff es) (Either KafkaError (ConsumerRecord (Maybe ByteString) (Maybe ByteString))) ->   Stream (Eff es) (Ingested es (Maybe ByteString)) ingestedStream mkI =   Stream.mapMaybeM $ \case-    Right cr -> pure (Just (mkI cr))+    Right cr -> Just <$> mkI cr     Left err -> throwError err

@@ -7,6 +7,7 @@     -- * Producing     produceMessages,     produceKeyedMessages,+    producePartitionMessages,      -- * Consuming via Adapter     consumeN,@@ -157,6 +158,25 @@       flushProducer   case result of     Left err -> error $ "Failed to produce: " <> show err+    Right () -> pure ()++-- | Produce payloads to exact partitions for deterministic rebalance tests.+producePartitionMessages :: TestEnv -> [(Int, ByteString)] -> IO ()+producePartitionMessages env records = do+  result <- runEff . runError @KafkaError $ do+    runKafkaProducer (mkProducerProps env) $ do+      forM_ records $ \(partition, payload) ->+        produceMessage+          ProducerRecord+            { prTopic = env.testTopic,+              prPartition = SpecifiedPartition partition,+              prKey = Nothing,+              prValue = Just payload,+              prHeaders = mempty+            }+      flushProducer+  case result of+    Left err -> error $ "Failed to produce to partitions: " <> show err     Right () -> pure ()  -- | Consume N messages from the test topic via the adapter, applying the given ack decision.

@@ -1,29 +1,43 @@ module Shibuya.Adapter.Kafka.AckHandleTest (tests) where +import Control.Concurrent.Async qualified as Async+import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar)+import Control.Exception (try) import Data.ByteString (ByteString) import Data.IORef (IORef, atomicModifyIORef', atomicWriteIORef, newIORef, readIORef) import Data.Int (Int64)-import Effectful (Eff, IOE, liftIO, runEff, (:>))+import Data.Map.Strict qualified as Map+import Effectful (Eff, IOE, Limit (..), Persistence (..), UnliftStrategy (..), liftIO, runEff, withEffToIO, (:>)) import Effectful.Dispatch.Dynamic (interpret) import Effectful.Error.Static (Error, runErrorNoCallStack, throwError) import Kafka.Consumer (RdKafkaRespErrT (..))-import Kafka.Consumer.Types (ConsumerRecord (..), Offset (..), PartitionOffset (..), Timestamp (..), TopicPartition (..))+import Kafka.Consumer.Types (ConsumerRecord (..), Offset (..), PartitionOffset (..), RebalanceEvent (..), Timestamp (..), TopicPartition (..)) import Kafka.Effectful.Consumer.Effect (KafkaConsumer (..)) import Kafka.Types (BatchSize (..), KafkaError (..), PartitionId (..), Timeout (..), TopicName (..))+import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter.Kafka (kafkaRebalanceHandler) import Shibuya.Adapter.Kafka.Config (KafkaAdapterConfig (..))-import Shibuya.Adapter.Kafka.Internal (KafkaAdapterState (..), ingestedStream, kafkaSource, mkAckHandle, newKafkaAdapterState)+import Shibuya.Adapter.Kafka.Internal (KafkaAcknowledgementException (..), KafkaAdapterState (..), ingestedStream, kafkaSource, mkAckHandle, mkIngested, newKafkaAdapterState)+import Shibuya.App (ProcessorId (..), defaultAppConfig, getAppMaster, mkProcessor, runApp, waitApp) import Shibuya.Core.Ack (AckDecision (..), HaltReason (..), RetryDelay (..)) import Shibuya.Core.AckHandle (AckHandle (..)) import Shibuya.Core.Ingested (Ingested)+import Shibuya.Internal.Runner.Master (ProcessorLifecycle (..), getLifecycleSnapshot)+import Shibuya.Telemetry.Effect (runTracingNoop) import Streamly.Data.Fold qualified as Fold import Streamly.Data.Stream qualified as Stream+import System.Random (mkStdGen, randomR)+import System.Timeout (timeout) import Test.Tasty (TestTree, testGroup) import Test.Tasty.HUnit (assertEqual, assertFailure, testCase)  data MockState = MockState   { storeAttempts :: !Int,+    storedOffsets :: ![Offset],     pauseAttempts :: !Int,     seekCalls :: ![TopicPartition],+    seekTimeouts :: ![Timeout],+    storeBlock :: !(Maybe (MVar (), MVar ())),     storeFailuresRemaining :: !Int,     pauseFailuresRemaining :: !Int,     seekFailuresRemaining :: !Int,@@ -37,11 +51,20 @@   testGroup     "AckHandle"     [ testCase "transient store failures retry and then succeed" testTransientStoreRetry,-      testCase "persistent transient store failure records fatal slot without throwing" testPersistentStoreFailure,-      testCase "fatal store failure records fatal slot after one attempt" testFatalStoreFailure,-      testCase "AckHalt pause failure does not throw and records fatal slot" testAckHaltPauseFailure,+      testCase "persistent transient store failure records fatal slot and throws" testPersistentStoreFailure,+      testCase "fatal store failure records fatal slot and throws after one attempt" testFatalStoreFailure,+      testCase "AckHalt pause failure records fatal slot and throws" testAckHaltPauseFailure,       testCase "AckRetry seeks exact failed offset and does not store" testAckRetrySeeks,+      testCase "AckRetry caps the consumer-lock seek timeout" testAckRetryBoundsSeek,       testCase "seek barrier prevents stale successor store" testBarrierSkipsSuccessorStore,+      testCase "earliest retry survives later retry and acknowledgement" testEarliestRetrySurvives,+      testCase "one delivery cannot resolve its own retry" testRetryRequiresRedelivery,+      testCase "repeated retry on one delivery is idempotent" testRepeatedRetry,+      testCase "fixed-seed sequences match the earliest-unresolved reference model" testReferenceModelSeeds,+      testCase "exhausted acknowledgement throws immediately" testPersistentStoreFailureThrows,+      testCase "revocation fences an old delivery callback" testRevocationFencesCallback,+      testCase "cancellation releases finalizer and consumer ownership" testCancellationReleasesOwnership,+      testCase "terminal acknowledgement failure reaches the core lifecycle" testTerminalFailureReachesCore,       testCase "source observes fatal slot before polling" testSourceObservesFatalSlot     ] @@ -61,8 +84,7 @@   let err = KafkaResponseError RdKafkaRespErrTransport   mock <- newIORef defaultMockState {storeFailuresRemaining = 99, storeError = err}   state <- newKafkaAdapterState-  result <- runFinalizer mock $ finalizeRecord state (recordAt 42) AckOk-  assertRight result+  assertAckFailure err $ runFinalizer mock $ finalizeRecord state (recordAt 42) AckOk   final <- readIORef mock   fatal <- readIORef state.fatalError   assertEqual "store attempts" 3 final.storeAttempts@@ -73,8 +95,7 @@   let err = KafkaBadConfiguration   mock <- newIORef defaultMockState {storeFailuresRemaining = 99, storeError = err}   state <- newKafkaAdapterState-  result <- runFinalizer mock $ finalizeRecord state (recordAt 42) AckOk-  assertRight result+  assertAckFailure err $ runFinalizer mock $ finalizeRecord state (recordAt 42) AckOk   final <- readIORef mock   fatal <- readIORef state.fatalError   assertEqual "store attempts" 1 final.storeAttempts@@ -85,8 +106,7 @@   let err = KafkaResponseError RdKafkaRespErrTransport   mock <- newIORef defaultMockState {pauseFailuresRemaining = 99, pauseError = err}   state <- newKafkaAdapterState-  result <- runFinalizer mock $ finalizeRecord state (recordAt 42) (AckHalt (HaltFatal "stop"))-  assertRight result+  assertAckFailure err $ runFinalizer mock $ finalizeRecord state (recordAt 42) (AckHalt (HaltFatal "stop"))   final <- readIORef mock   fatal <- readIORef state.fatalError   assertEqual "pause attempts" 3 final.pauseAttempts@@ -105,6 +125,18 @@     [TopicPartition (TopicName "orders") (PartitionId 0) (PartitionOffset 42)]     final.seekCalls +testAckRetryBoundsSeek :: IO ()+testAckRetryBoundsSeek = do+  mock <- newIORef defaultMockState+  state <- newKafkaAdapterState+  let slowConfig = testConfig {pollTimeout = Timeout 5000}+  result <- runFinalizer mock $ do+    AckHandle finalize <- mkAckHandle state slowConfig (recordAt 42)+    finalize (AckRetry (RetryDelay 0))+  assertRight result+  final <- readIORef mock+  assertEqual "seek timeout" [Timeout 100] final.seekTimeouts+ testBarrierSkipsSuccessorStore :: IO () testBarrierSkipsSuccessorStore = do   mock <- newIORef defaultMockState@@ -117,6 +149,164 @@   final <- readIORef mock   assertEqual "only retried message stored" 1 final.storeAttempts +testEarliestRetrySurvives :: IO ()+testEarliestRetrySurvives = do+  mock <- newIORef defaultMockState+  state <- newKafkaAdapterState+  result <- runFinalizer mock $ do+    finalizeRecord state (recordAt 42) (AckRetry (RetryDelay 0))+    finalizeRecord state (recordAt 43) (AckRetry (RetryDelay 0))+    finalizeRecord state (recordAt 43) AckOk+  assertRight result+  final <- readIORef mock+  assertEqual+    "both retries seek the earliest unresolved offset"+    [ TopicPartition (TopicName "orders") (PartitionId 0) (PartitionOffset 42),+      TopicPartition (TopicName "orders") (PartitionId 0) (PartitionOffset 42)+    ]+    final.seekCalls+  assertEqual "later acknowledgement remains fenced" 0 final.storeAttempts++testRetryRequiresRedelivery :: IO ()+testRetryRequiresRedelivery = do+  mock <- newIORef defaultMockState+  state <- newKafkaAdapterState+  result <- runFinalizer mock $ do+    AckHandle finalize <- mkAckHandle state testConfig (recordAt 42)+    finalize (AckRetry (RetryDelay 0))+    finalize AckOk+  assertRight result+  final <- readIORef mock+  assertEqual "same delivery cannot store after requesting retry" 0 final.storeAttempts++testRepeatedRetry :: IO ()+testRepeatedRetry = do+  mock <- newIORef defaultMockState+  state <- newKafkaAdapterState+  result <- runFinalizer mock $ do+    AckHandle finalize <- mkAckHandle state testConfig (recordAt 42)+    finalize (AckRetry (RetryDelay 0))+    finalize (AckRetry (RetryDelay 0))+  assertRight result+  final <- readIORef mock+  assertEqual "one successful retry performs one seek" 1 (length final.seekCalls)++testReferenceModelSeeds :: IO ()+testReferenceModelSeeds =+  -- EP-44's integrated release gate requires at least 1,000 recorded model+  -- cases. Keep the range deterministic so a failure names its replayable+  -- seed and the ordinary package suite exercises the full gate.+  mapM_ runSeed [400040 .. 401039]+  where+    runSeed seed = do+      let generator = mkStdGen seed+          (baseDelta, generator') = randomR (0, 5 :: Int64) generator+          (gap, generator'') = randomR (1, 5 :: Int64) generator'+          (laterFirst, _) = randomR (False, True) generator''+          base = 40 + baseDelta+          later = base + gap+          (firstOffset, secondOffset) = if laterFirst then (later, base) else (base, later)+          retryOrder = [firstOffset, secondOffset]+          expectedSeeks = map toTopicPartition (runningMinimum retryOrder)+      mock <- newIORef defaultMockState+      state <- newKafkaAdapterState+      result <- runFinalizer mock $ do+        first <- mkAckHandle state testConfig (recordAt firstOffset)+        second <- mkAckHandle state testConfig (recordAt secondOffset)+        finalizeHandle first (AckRetry (RetryDelay 0))+        finalizeHandle second (AckRetry (RetryDelay 0))+        prematureLater <- mkAckHandle state testConfig (recordAt later)+        finalizeHandle prematureLater AckOk+        replayBase <- mkAckHandle state testConfig (recordAt base)+        finalizeHandle replayBase AckOk+        replayLater <- mkAckHandle state testConfig (recordAt later)+        finalizeHandle replayLater AckOk+      assertRight result+      final <- readIORef mock+      assertEqual ("seed " <> show seed <> " seek boundary") expectedSeeks final.seekCalls+      assertEqual ("seed " <> show seed <> " stored offsets") [Offset base, Offset later] final.storedOffsets++    runningMinimum = \case+      [] -> []+      first : rest -> scanl min first rest++    toTopicPartition offset =+      TopicPartition (TopicName "orders") (PartitionId 0) (PartitionOffset offset)++    finalizeHandle (AckHandle finalize) = finalize++testPersistentStoreFailureThrows :: IO ()+testPersistentStoreFailureThrows = do+  let err = KafkaResponseError RdKafkaRespErrTransport+  mock <- newIORef defaultMockState {storeFailuresRemaining = 99, storeError = err}+  state <- newKafkaAdapterState+  assertAckFailure err $ runFinalizer mock $ finalizeRecord state (recordAt 42) AckOk++testRevocationFencesCallback :: IO ()+testRevocationFencesCallback = do+  mock <- newIORef defaultMockState+  state <- newKafkaAdapterState+  result <- runFinalizer mock $ do+    AckHandle finalize <- mkAckHandle state testConfig (recordAt 42)+    liftIO $+      kafkaRebalanceHandler+        state+        (error "consumer handle is not inspected")+        (RebalanceRevoke [(TopicName "orders", PartitionId 0)])+    finalize AckOk+  assertRight result+  final <- readIORef mock+  assertEqual "revoked callback cannot store" 0 final.storeAttempts++testCancellationReleasesOwnership :: IO ()+testCancellationReleasesOwnership = do+  started <- newEmptyMVar+  release <- newEmptyMVar+  mock <- newIORef defaultMockState {storeBlock = Just (started, release)}+  state <- newKafkaAdapterState+  result <- runFinalizer mock $ do+    AckHandle finalize <- mkAckHandle state testConfig (recordAt 42)+    withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO -> do+      worker <- Async.async (runInIO (finalize AckOk))+      takeMVar started+      Async.cancel worker+      atomicModifyIORef' mock (\mockState -> (mockState {storeBlock = Nothing}, ()))+      mbCompleted <- timeout 1000000 (runInIO (finalize AckOk))+      case mbCompleted of+        Nothing -> assertFailure "finalizer lock or consumer lock remained held after cancellation"+        Just () -> pure ()+  assertRight result+  final <- readIORef mock+  assertEqual "cancelled attempt plus successful retry" 2 final.storeAttempts++testTerminalFailureReachesCore :: IO ()+testTerminalFailureReachesCore = do+  let err = KafkaBadConfiguration+      processorId = ProcessorId "kafka-terminal-ack-failure"+  mock <- newIORef defaultMockState {storeFailuresRemaining = 99, storeError = err}+  state <- newKafkaAdapterState+  result <-+    timeout 5000000 $+      runEff . runErrorNoCallStack @KafkaError . runMockConsumer mock . runTracingNoop $ do+        ingested <- mkIngested state testConfig (recordAt 42)+        let adapter =+              Adapter+                { adapterName = "kafka:test-terminal-ack-failure",+                  source = Stream.fromList [ingested],+                  shutdown = pure ()+                }+            processor = mkProcessor adapter (\_ -> pure AckOk)+        appResult <- runApp defaultAppConfig [(processorId, processor)]+        case appResult of+          Left appError -> error $ "runApp failed: " <> show appError+          Right appHandle -> do+            waitApp appHandle+            lifecycle <- getLifecycleSnapshot (getAppMaster appHandle)+            pure (Map.lookup processorId lifecycle)+  case result of+    Just (Right (Just (LifecycleFailed _ _))) -> pure ()+    other -> assertFailure $ "expected retained terminal acknowledgement failure, got: " <> show other+ testSourceObservesFatalSlot :: IO () testSourceObservesFatalSlot = do   let err = KafkaBadConfiguration@@ -144,9 +334,9 @@   Eff es a runMockConsumer mock =   interpret $ \_env -> \case-    StoreOffsetMessage _ -> attemptStore mock+    StoreOffsetMessage cr -> attemptStore mock cr.crOffset     PausePartitions _ -> attemptPause mock-    SeekPartitions tps _ -> recordSeek mock tps+    SeekPartitions tps seekTimeout -> recordSeek mock tps seekTimeout     PollMessage _ -> error "AckHandleTest: PollMessage not exercised"     PollMessageBatch _ _ -> error "AckHandleTest: PollMessageBatch not exercised"     PollMessageEither _ -> error "AckHandleTest: PollMessageEither not exercised"@@ -164,17 +354,29 @@  unreachableBuilder ::   ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->-  Ingested es (Maybe ByteString)+  Eff es (Ingested es (Maybe ByteString)) unreachableBuilder _ = error "AckHandleTest: source should not yield records" -attemptStore :: (IOE :> es, Error KafkaError :> es) => IORef MockState -> Eff es ()-attemptStore mock = do-  mbErr <-+attemptStore ::+  (IOE :> es, Error KafkaError :> es) =>+  IORef MockState ->+  Offset ->+  Eff es ()+attemptStore mock offset = do+  (mbBlock, mbErr) <-     liftIO $       atomicModifyIORef' mock $ \s ->         let remaining = s.storeFailuresRemaining-            s' = s {storeAttempts = s.storeAttempts + 1, storeFailuresRemaining = max 0 (remaining - 1)}-         in (s', if remaining > 0 then Just s.storeError else Nothing)+            s' =+              s+                { storeAttempts = s.storeAttempts + 1,+                  storedOffsets = s.storedOffsets <> [offset],+                  storeFailuresRemaining = max 0 (remaining - 1)+                }+         in (s', (s.storeBlock, if remaining > 0 then Just s.storeError else Nothing))+  liftIO $ case mbBlock of+    Nothing -> pure ()+    Just (started, release) -> putMVar started () >> takeMVar release   maybe (pure ()) throwError mbErr  attemptPause :: (IOE :> es, Error KafkaError :> es) => IORef MockState -> Eff es ()@@ -187,13 +389,18 @@          in (s', if remaining > 0 then Just s.pauseError else Nothing)   maybe (pure ()) throwError mbErr -recordSeek :: (IOE :> es, Error KafkaError :> es) => IORef MockState -> [TopicPartition] -> Eff es ()-recordSeek mock tps = do+recordSeek :: (IOE :> es, Error KafkaError :> es) => IORef MockState -> [TopicPartition] -> Timeout -> Eff es ()+recordSeek mock tps seekTimeout = do   mbErr <-     liftIO $       atomicModifyIORef' mock $ \s ->         let remaining = s.seekFailuresRemaining-            s' = s {seekCalls = s.seekCalls <> tps, seekFailuresRemaining = max 0 (remaining - 1)}+            s' =+              s+                { seekCalls = s.seekCalls <> tps,+                  seekTimeouts = s.seekTimeouts <> [seekTimeout],+                  seekFailuresRemaining = max 0 (remaining - 1)+                }          in (s', if remaining > 0 then Just s.seekError else Nothing)   maybe (pure ()) throwError mbErr @@ -204,8 +411,9 @@   AckDecision ->   Eff es () finalizeRecord state cr decision =-  let AckHandle finalize = mkAckHandle state testConfig cr-   in finalize decision+  do+    AckHandle finalize <- mkAckHandle state testConfig cr+    finalize decision  recordAt :: Int64 -> ConsumerRecord (Maybe ByteString) (Maybe ByteString) recordAt offset =@@ -231,8 +439,11 @@ defaultMockState =   MockState     { storeAttempts = 0,+      storedOffsets = [],       pauseAttempts = 0,       seekCalls = [],+      seekTimeouts = [],+      storeBlock = Nothing,       storeFailuresRemaining = 0,       pauseFailuresRemaining = 0,       seekFailuresRemaining = 0,@@ -245,3 +456,10 @@ assertRight = \case   Left err -> assertFailure $ "expected Right, got Left: " <> show err   Right _ -> pure ()++assertAckFailure :: KafkaError -> IO (Either KafkaError ()) -> IO ()+assertAckFailure expected action = do+  result <- try @KafkaAcknowledgementException action+  case result of+    Left (KafkaAcknowledgementException actual) -> assertEqual "acknowledgement error" expected actual+    Right value -> assertFailure $ "expected KafkaAcknowledgementException, got: " <> show value

@@ -4,7 +4,7 @@ module Shibuya.Adapter.Kafka.AdapterTest (tests) where  import Data.ByteString (ByteString)-import Effectful (runEff)+import Effectful (Eff, runEff) import Effectful.Error.Static (runError) import Kafka.Consumer.Types (ConsumerRecord) import Kafka.Types (KafkaError (..))@@ -28,7 +28,7 @@ -- never fires. unreachableBuilder ::   ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->-  Ingested es (Maybe ByteString)+  Eff es (Ingested es (Maybe ByteString)) unreachableBuilder _ = error "AdapterTest: Right branch should not be reached"  testFatalPropagation :: IO ()

@@ -1,23 +1,28 @@ module Shibuya.Adapter.Kafka.IntegrationTest (tests) where +import Control.Concurrent.Async qualified as Async+import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)+import Control.Concurrent.STM (TChan, atomically, newTChanIO, readTChan, writeTChan) import Control.Exception (throwIO) import Control.Monad (forM) import Control.Monad.IO.Class (liftIO) import Data.ByteString (ByteString) import Data.ByteString.Char8 qualified as BS8-import Data.IORef (modifyIORef', newIORef, readIORef, writeIORef)+import Data.IORef (atomicModifyIORef', modifyIORef', newIORef, readIORef, writeIORef) import Data.List (nub, sort) import Data.Maybe (mapMaybe) import Data.Text qualified as Text-import Effectful (runEff)+import Effectful (Eff, Limit (..), Persistence (..), UnliftStrategy (..), runEff, withEffToIO) import Effectful.Error.Static (runError)-import Kafka.Consumer.Types (OffsetCommit (..), OffsetReset (..))+import Kafka.Consumer.Types (ConsumerRecord (..), OffsetCommit (..), OffsetReset (..), RebalanceEvent (..)) import Kafka.Effectful.Consumer   ( brokersList,     groupId,     noAutoOffsetStore,     offsetReset,+    rebalanceCallback,     runKafkaConsumer,+    setCallback,     topics,   ) import Kafka.Effectful.Consumer.Effect (commitAllOffsets, pollMessageBatch)@@ -28,16 +33,18 @@     createTopicWithPartitions,     produceKeyedMessages,     produceMessages,+    producePartitionMessages,     withTestEnv,   ) import Kafka.Types   ( BatchSize (..),     KafkaError,+    PartitionId (..),     Timeout (..),     TopicName (..),   ) import Shibuya.Adapter (Adapter (..))-import Shibuya.Adapter.Kafka (KafkaAdapterConfig (..), kafkaAdapter)+import Shibuya.Adapter.Kafka (KafkaAdapterConfig (..), kafkaAdapter, kafkaAdapterWith, kafkaRebalanceHandler, newKafkaAdapterState) import Shibuya.App (ProcessorId (..), defaultAppConfig, mkProcessor, runApp, waitApp) import Shibuya.Core.Ack (AckDecision (..), RetryDelay (..)) import Shibuya.Core.AckHandle (AckHandle (..))@@ -60,8 +67,12 @@       testCase "Batch polling" testBatchPolling,       testCase "Graceful shutdown" testGracefulShutdown,       testCase "Idle graceful shutdown completes promptly" testIdleGracefulShutdown,+      testCase "Repeated shutdown is idempotent" testRepeatedShutdown,       testCase "AckRetry redelivers within the same session" testAckRetryRedelivery,+      testCase "later buffered retry cannot replace the earliest recovery boundary" testBufferedRetryBoundary,       testCase "AckRetry is not committed past when session exits" testAckRetryAbandonedSession,+      testCase "late callback after revocation cannot advance the broker offset" testRevokedCallbackDoesNotCommit,+      testCase "actual reassignment fences a late callback from the old owner" testActualReassignmentFence,       testCase "Handler exception redelivers instead of skipping" testHandlerExceptionRedelivery     ] @@ -205,6 +216,22 @@     Just (Left (_cs, err)) -> assertFailure $ "idle shutdown failed: " <> show err     Just (Right ()) -> pure () +testRepeatedShutdown :: IO ()+testRepeatedShutdown = withTestEnv $ \env -> do+  createTopic env++  timedResult <- timeout 3000000 $ runEff . runError @KafkaError $ do+    let props = brokersList [env.testBroker] <> groupId env.testGroupId <> noAutoOffsetStore+        sub = topics [env.testTopic] <> offsetReset Earliest+    runKafkaConsumer props sub $ do+      Adapter {shutdown} <- kafkaAdapter (testConfig env)+      shutdown+      shutdown+  case timedResult of+    Nothing -> assertFailure "repeated shutdown did not terminate promptly"+    Just (Left (_cs, err)) -> assertFailure $ "repeated shutdown failed: " <> show err+    Just (Right ()) -> pure ()+ testAckRetryRedelivery :: IO () testAckRetryRedelivery = withTestEnv $ \env -> do   createTopic env@@ -251,6 +278,55 @@     Left (_cs, err) -> assertFailure $ "post-commit verification failed: " <> show err     Right () -> pure () +testBufferedRetryBoundary :: IO ()+testBufferedRetryBoundary = withTestEnv $ \env -> do+  createTopic env+  let payloads = ["boundary-42", "boundary-43"]+  produceMessages env payloads++  result <- runEff . runError @KafkaError $ do+    let props = brokersList [env.testBroker] <> groupId env.testGroupId <> noAutoOffsetStore+        sub = topics [env.testTopic] <> offsetReset Earliest+    runKafkaConsumer props sub $ do+      Adapter {source} <- kafkaAdapter (testConfig env)+      firstBuffered <- liftIO $ newIORef Nothing+      deliveryIndex <- liftIO $ newIORef (0 :: Int)+      replayedPayloads <- liftIO $ newIORef ([] :: [ByteString])+      Stream.fold Fold.drain+        $ Stream.mapM+          ( \ingested -> do+              index <- liftIO $ atomicModifyIORef' deliveryIndex (\n -> (n + 1, n))+              case index of+                0 -> liftIO $ writeIORef firstBuffered (Just ingested)+                1 -> do+                  first <-+                    liftIO (readIORef firstBuffered) >>= \case+                      Just value -> pure value+                      Nothing -> error "missing first buffered delivery"+                  finalizeIngested first (AckRetry (RetryDelay 0))+                  finalizeIngested ingested (AckRetry (RetryDelay 0))+                _ -> do+                  case ingested of+                    Ingested {envelope = Envelope {payload = Just payload}} ->+                      liftIO $ modifyIORef' replayedPayloads (<> [payload])+                    _ -> pure ()+                  finalizeIngested ingested AckOk+          )+        $ Stream.take 4 source+      replayed <- liftIO $ readIORef replayedPayloads+      liftIO $+        assertEqual+          "retry seeks the earliest delivery before its successor"+          payloads+          replayed+      commitAllOffsets OffsetCommit+  case result of+    Left (_cs, err) -> assertFailure $ "buffered retry boundary failed: " <> show err+    Right () -> pure ()++  noRedelivery <- pollPayloads env 3+  assertEqual "both replayed deliveries committed" [] noRedelivery+ testAckRetryAbandonedSession :: IO () testAckRetryAbandonedSession = withTestEnv $ \env -> do   createTopic env@@ -295,6 +371,100 @@   delivered <- readIORef redelivered   assertBool ("expected ab-2 redelivery, saw " <> show delivered) ("ab-2" `elem` delivered) +testRevokedCallbackDoesNotCommit :: IO ()+testRevokedCallbackDoesNotCommit = withTestEnv $ \env -> do+  createTopic env+  produceMessages env ["revoked-late-callback"]++  firstSession <- runEff . runError @KafkaError $ do+    let props = brokersList [env.testBroker] <> groupId env.testGroupId <> noAutoOffsetStore+        sub = topics [env.testTopic] <> offsetReset Earliest+    runKafkaConsumer props sub $ do+      state <- liftIO newKafkaAdapterState+      Adapter {source} <- kafkaAdapterWith state (testConfig env)+      delivered <- Stream.fold Fold.toList $ Stream.take 1 source+      case delivered of+        [ingested] -> do+          liftIO $+            kafkaRebalanceHandler+              state+              (error "consumer handle is not inspected")+              (RebalanceRevoke [(env.testTopic, PartitionId 0)])+          finalizeIngested ingested AckOk+        other -> liftIO $ assertFailure $ "expected one delivery before revoke, got " <> show (length other)+  case firstSession of+    Left (_cs, err) -> assertFailure $ "revoked session failed: " <> show err+    Right () -> pure ()++  replayed <- consumeN env 1 AckOk+  assertEqual+    "revoked delivery remains recoverable"+    [Just "revoked-late-callback"]+    [envelope.payload | envelope <- replayed]++testActualReassignmentFence :: IO ()+testActualReassignmentFence = withTestEnv $ \env -> do+  createTopicWithPartitions env 2+  producePartitionMessages env [(0, "rebalance-partition-0"), (1, "rebalance-partition-1")]+  state <- newKafkaAdapterState+  events <- newTChanIO+  handlesReady <- newEmptyMVar+  releaseFirstConsumer <- newEmptyMVar++  let callback consumer event = do+        kafkaRebalanceHandler state consumer event+        atomically $ writeTChan events event+      firstProps =+        brokersList [env.testBroker]+          <> groupId env.testGroupId+          <> noAutoOffsetStore+          <> setCallback (rebalanceCallback callback)+      sub = topics [env.testTopic] <> offsetReset Earliest+      firstConsumer =+        runEff . runError @KafkaError $+          runKafkaConsumer firstProps sub $ do+            Adapter {source} <- kafkaAdapterWith state (testConfig env)+            delivered <- Stream.fold Fold.toList $ Stream.take 2 source+            withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO -> do+              putMVar handlesReady (map (callbackHandle runInIO) delivered)+              takeMVar releaseFirstConsumer+      secondConsumer =+        runEff . runError @KafkaError $+          runKafkaConsumer+            (brokersList [env.testBroker] <> groupId env.testGroupId <> noAutoOffsetStore)+            sub+            ( do+                _ <- forM [1 .. 20 :: Int] $ \_ -> pollMessageBatch (Timeout 250) (BatchSize 100)+                pure ()+            )++  Async.withAsync firstConsumer $ \firstAsync -> do+    mbHandles <- timeout 10000000 (takeMVar handlesReady)+    handles <- case mbHandles of+      Nothing -> assertFailure "first consumer did not receive both partitions" >> pure []+      Just value -> pure value+    revoked <- Async.withAsync secondConsumer $ \secondAsync -> do+      mbRevoked <- timeout 10000000 (awaitRevokedPartition env.testTopic events)+      revoked <- case mbRevoked of+        Nothing -> assertFailure "second consumer did not trigger a revocation" >> pure (PartitionId (-1))+        Just value -> pure value+      case lookup revoked handles of+        Nothing -> assertFailure $ "no retained handle for revoked partition " <> show revoked+        Just lateFinalize -> lateFinalize AckOk+      putMVar releaseFirstConsumer ()+      firstResult <- Async.wait firstAsync+      case firstResult of+        Left (_cs, err) -> assertFailure $ "first reassignment consumer failed: " <> show err+        Right () -> pure ()+      Async.cancel secondAsync+      pure revoked++    replayed <- pollPayloads env 5+    let expected = if revoked == PartitionId 0 then "rebalance-partition-0" else "rebalance-partition-1"+    assertBool+      ("expected revoked partition payload to remain recoverable, saw " <> show replayed)+      (expected `elem` replayed)+ testHandlerExceptionRedelivery :: IO () testHandlerExceptionRedelivery = withTestEnv $ \env -> do   createTopic env@@ -341,3 +511,40 @@  countPayload :: ByteString -> [ByteString] -> Int countPayload target = length . filter (== target)++finalizeIngested :: Ingested es payload -> AckDecision -> Eff es ()+finalizeIngested Ingested {ack = AckHandle finalize} = finalize++pollPayloads :: TestEnv -> Int -> IO [ByteString]+pollPayloads env pollCount = do+  result <- runEff . runError @KafkaError $ do+    let props = brokersList [env.testBroker] <> groupId env.testGroupId <> noAutoOffsetStore+        sub = topics [env.testTopic] <> offsetReset Earliest+    runKafkaConsumer props sub $ do+      batches <- forM [1 .. pollCount] $ \_ ->+        pollMessageBatch (Timeout 500) (BatchSize 100)+      pure [payload | Right record <- concat batches, Just payload <- [record.crValue]]+  case result of+    Left (_cs, err) -> assertFailure ("poll verification failed: " <> show err) >> pure []+    Right payloads -> pure payloads++callbackHandle ::+  (forall a. Eff es a -> IO a) ->+  Ingested es payload ->+  (PartitionId, AckDecision -> IO ())+callbackHandle runInIO Ingested {envelope = Envelope {partition}, ack = AckHandle finalize} =+  case partition of+    Just partitionText -> (PartitionId (read (Text.unpack partitionText)), runInIO . finalize)+    Nothing -> error "Kafka delivery did not carry a partition"++awaitRevokedPartition :: TopicName -> TChan RebalanceEvent -> IO PartitionId+awaitRevokedPartition topic events = atomically loop+  where+    loop = do+      event <- readTChan events+      case event of+        RebalanceRevoke revoked ->+          case [partition | (topic', partition) <- revoked, topic' == topic] of+            partition : _ -> pure partition+            [] -> loop+        _ -> loop