shibuya-kafka-adapter 0.9.0.0 → 0.9.0.1
raw patch · 12 files changed
+1230/−1225 lines, 12 filesdep ~effectful-coredep ~kafka-effectfuldep ~shibuya-corePVP ok
version bump matches the API change (PVP)
Dependency ranges changed: effectful-core, kafka-effectful, shibuya-core
API changes (from Hackage documentation)
Files
- CHANGELOG.md +19/−0
- shibuya-kafka-adapter.cabal +74/−64
- src/Shibuya/Adapter/Kafka.hs +174/−178
- src/Shibuya/Adapter/Kafka/Config.hs +30/−33
- src/Shibuya/Adapter/Kafka/Convert.hs +73/−79
- src/Shibuya/Adapter/Kafka/Internal.hs +187/−197
- test/Kafka/TestEnv.hs +96/−96
- test/Main.hs +8/−8
- test/Shibuya/Adapter/Kafka/AckHandleTest.hs +167/−166
- test/Shibuya/Adapter/Kafka/AdapterTest.hs +33/−35
- test/Shibuya/Adapter/Kafka/ConvertTest.hs +126/−126
- test/Shibuya/Adapter/Kafka/IntegrationTest.hs +243/−243
CHANGELOG.md view
@@ -1,5 +1,24 @@ # Changelog +## 0.9.0.1 — 2026-09-15++### Other Changes++- Support `effectful-core` 2.7: raise the upper bound from `<2.7` to `<2.8`.+ Built and tested against effectful-core 2.7.1.2 on GHC 9.12.4; no source+ changes were needed. Nothing here uses the APIs 2.7 deprecated, renamed, or+ removed: `withLiftMap`, `stateM`/`modifyM`, the `*StateMVar` functions,+ `Effectful.Internal.MTL`, `SharedSuffix`, `KnownEffects`, or the `LocalEnv`+ `handlerEs` parameter. The lower bound stays at 2.6.1 so consumers are not+ forced to upgrade.+- Require `shibuya-core ^>=0.9.0.1` and `kafka-effectful ^>=0.3.1.0`. Both are+ bounds-only upstream releases that widen `effectful` to `<2.8`; the previous+ versions cap it at `<2.7` and would hold the whole build on effectful 2.6.+ `kafka-effectful` 0.3.1.0 also fixes consumer crash-loops on partition EOF+ and idle-commit conditions.+- Keep the example and benchmark packages on the shared `0.9.0.1` repo version+ line.+ ## 0.9.0.0 — 2026-08-10 ### Breaking Changes
shibuya-kafka-adapter.cabal view
@@ -1,39 +1,45 @@-cabal-version: 3.12-name: shibuya-kafka-adapter-version: 0.9.0.0-synopsis: Kafka adapter for the Shibuya queue processing framework+cabal-version: 3.12+name: shibuya-kafka-adapter+version: 0.9.0.1+synopsis: Kafka adapter for the Shibuya queue processing framework description: A Shibuya adapter that integrates with Apache Kafka via kafka-effectful and hw-kafka-client. Provides polling, offset commit semantics, partition awareness, and graceful shutdown. -author: Nadeem Bitar-copyright: 2026 Nadeem Bitar-maintainer: nadeem@gmail.com-homepage: https://github.com/shinzui/shibuya-kafka-adapter-bug-reports: https://github.com/shinzui/shibuya-kafka-adapter/issues-license: MIT-license-file: LICENSE-build-type: Simple-tested-with: GHC ==9.12.2-category: Concurrency, Streaming+author: Nadeem Bitar+copyright: 2026 Nadeem Bitar+maintainer: nadeem@gmail.com+homepage: https://github.com/shinzui/shibuya-kafka-adapter+bug-reports: https://github.com/shinzui/shibuya-kafka-adapter/issues+license: MIT+license-file: LICENSE+build-type: Simple+tested-with: ghc ==9.12.2+category: Concurrency, Streaming extra-doc-files: CHANGELOG.md README.md source-repository head- type: git+ type: git location: https://github.com/shinzui/shibuya-kafka-adapter.git common warnings ghc-options:- -Wall -Wcompat -Widentities -Wincomplete-uni-patterns- -Wincomplete-record-updates -Wredundant-constraints- -fhide-source-paths -Wmissing-export-lists -Wpartial-fields+ -Wall+ -Wcompat+ -Widentities+ -Wincomplete-uni-patterns+ -Wincomplete-record-updates+ -Wredundant-constraints+ -fhide-source-paths+ -Wmissing-export-lists+ -Wpartial-fields -Wmissing-deriving-strategies library- import: warnings+ import: warnings exposed-modules: Shibuya.Adapter.Kafka Shibuya.Adapter.Kafka.Config@@ -52,33 +58,37 @@ QuasiQuotes build-depends:- , base ^>=4.21.0.0- , bytestring ^>=0.12- , containers ^>=0.7- , effectful-core ^>=2.6.1.0- , 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.0.0- , shibuya-core ^>=0.9.0.0- , stm ^>=2.5- , streamly ^>=0.11- , streamly-core ^>=0.3- , text ^>=2.1- , time ^>=1.14- , unordered-containers ^>=0.2+ base ^>=4.21.0.0,+ bytestring ^>=0.12,+ containers ^>=0.7,+ effectful-core >=2.6.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,+ stm ^>=2.5,+ streamly ^>=0.11,+ streamly-core ^>=0.3,+ text ^>=2.1,+ time ^>=1.14,+ unordered-containers ^>=0.2, - hs-source-dirs: src- default-language: GHC2024+ hs-source-dirs: src+ default-language: GHC2024 test-suite shibuya-kafka-adapter-test- import: warnings- default-language: GHC2024- type: exitcode-stdio-1.0- ghc-options: -threaded -rtsopts -with-rtsopts=-N- hs-source-dirs: test- main-is: Main.hs+ import: warnings+ default-language: GHC2024+ type: exitcode-stdio-1.0+ ghc-options:+ -threaded+ -rtsopts+ -with-rtsopts=-N++ hs-source-dirs: test+ main-is: Main.hs default-extensions: DeriveAnyClass DerivingStrategies@@ -98,24 +108,24 @@ Shibuya.Adapter.Kafka.IntegrationTest build-depends:- , async ^>=2.2- , base ^>=4.21.0.0- , bytestring- , containers- , effectful-core- , hs-opentelemetry-api ^>=1.0- , hs-opentelemetry-semantic-conventions ^>=1.40- , hw-kafka-client- , kafka-effectful- , process- , random- , shibuya-core ^>=0.9.0.0- , shibuya-kafka-adapter- , stm- , streamly- , streamly-core- , tasty ^>=1.5- , tasty-hunit ^>=0.10- , text- , time- , unordered-containers ^>=0.2+ async ^>=2.2,+ base ^>=4.21.0.0,+ bytestring,+ containers,+ effectful-core,+ 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-kafka-adapter,+ stm,+ streamly,+ streamly-core,+ tasty ^>=1.5,+ tasty-hunit ^>=0.10,+ text,+ time,+ unordered-containers ^>=0.2,
src/Shibuya/Adapter/Kafka.hs view
@@ -1,96 +1,95 @@-{- | Kafka adapter for the Shibuya queue processing framework.--This adapter integrates with Apache Kafka via-[kafka-effectful](https://github.com/shinzui/kafka-effectful) and-[hw-kafka-client](https://github.com/haskell-works/hw-kafka-client).--== Example Usage--@-import Shibuya.App (defaultAppConfig, runApp, mkProcessor)-import Shibuya.Adapter.Kafka (kafkaAdapter, defaultConfig)-import Kafka.Effectful.Consumer (runKafkaConsumer)-import Kafka.Consumer (brokersList, groupId, noAutoOffsetStore)--main :: IO ()-main = runEff- . runError \@KafkaError- . runKafkaConsumer props sub- $ do- adapter <- kafkaAdapter (defaultConfig [TopicName \"orders\"])- result <- runApp defaultAppConfig- [ (ProcessorId \"orders\", mkProcessor adapter myHandler)- ]- ...-@--== Message Lifecycle--1. Messages are polled from Kafka in batches.-2. Each message is wrapped as an @Ingested@ value with an @AckHandle@ via- Shibuya's adapter-facing smart constructors.-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.-5. On @AckDeadLetter@, the offset is stored after a loud stderr warning.-6. On @AckHalt@, the partition is paused and offset is not stored.--== Serial Operation Required--This adapter must be run with serial message processing. librdkafka stores the-highest offset per partition without gap tracking, so concurrent finalization-can commit past an earlier message that failed, halted, or requested retry.-The 'Adapter' value does not contain the processor concurrency policy, so this-is a caller contract rather than a runtime guard: do not use @Async@ or @Ahead@-processing with this adapter until a gap-tracking commit layer exists.--== Dead Letters Are Dropped--This adapter does not include a DLQ producer. @AckDeadLetter@ stores the-message offset so the consumer group moves on, emits a warning to stderr, and-makes the message unrecoverable from that consumer group's committed position.-Core tracing still records the dead-letter decision and reason on the-per-message span.--Kafka does not expose a per-message delivery counter through this consumer-API, so 'Shibuya.Core.Types.Envelope.attempt' is always 'Nothing'. Handlers-cannot safely cap retries by counting attempts from the envelope; use an-external store, or return @AckHalt@ to stop the stream.--== Fatal Error Propagation--Non-fatal Kafka errors (poll timeouts, partition EOFs, and the rest of the-non-fatal set defined by @hw-kafka-streamly@'s 'Kafka.Streamly.Stream.isFatal')-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'.--== AckHalt Partition Pause Semantics--@AckHalt@ pauses the originating partition by calling @pausePartitions@ from-@kafka-effectful@ and the processor stops. Polling therefore stops. After-@max.poll.interval.ms@ (librdkafka default: 300000 ms, or 5 minutes) the broker-may evict this consumer from its group and rebalance the partition to another-member, which resumes from the last committed offset. A single-member group-simply stalls until restart. Paused state is session-local and does not outlive-the current consumer.--== Rebalance Callback Helper--'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.--}-module Shibuya.Adapter.Kafka (- -- * Adapter+-- | Kafka adapter for the Shibuya queue processing framework.+--+-- This adapter integrates with Apache Kafka via+-- [kafka-effectful](https://github.com/shinzui/kafka-effectful) and+-- [hw-kafka-client](https://github.com/haskell-works/hw-kafka-client).+--+-- == Example Usage+--+-- @+-- import Shibuya.App (defaultAppConfig, runApp, mkProcessor)+-- import Shibuya.Adapter.Kafka (kafkaAdapter, defaultConfig)+-- import Kafka.Effectful.Consumer (runKafkaConsumer)+-- import Kafka.Consumer (brokersList, groupId, noAutoOffsetStore)+--+-- main :: IO ()+-- main = runEff+-- . runError \@KafkaError+-- . runKafkaConsumer props sub+-- $ do+-- adapter <- kafkaAdapter (defaultConfig [TopicName \"orders\"])+-- result <- runApp defaultAppConfig+-- [ (ProcessorId \"orders\", mkProcessor adapter myHandler)+-- ]+-- ...+-- @+--+-- == Message Lifecycle+--+-- 1. Messages are polled from Kafka in batches.+-- 2. Each message is wrapped as an @Ingested@ value with an @AckHandle@ via+-- Shibuya's adapter-facing smart constructors.+-- 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.+-- 5. On @AckDeadLetter@, the offset is stored after a loud stderr warning.+-- 6. On @AckHalt@, the partition is paused and offset is not stored.+--+-- == Serial Operation Required+--+-- This adapter must be run with serial message processing. librdkafka stores the+-- highest offset per partition without gap tracking, so concurrent finalization+-- can commit past an earlier message that failed, halted, or requested retry.+-- The 'Adapter' value does not contain the processor concurrency policy, so this+-- is a caller contract rather than a runtime guard: do not use @Async@ or @Ahead@+-- processing with this adapter until a gap-tracking commit layer exists.+--+-- == Dead Letters Are Dropped+--+-- This adapter does not include a DLQ producer. @AckDeadLetter@ stores the+-- message offset so the consumer group moves on, emits a warning to stderr, and+-- makes the message unrecoverable from that consumer group's committed position.+-- Core tracing still records the dead-letter decision and reason on the+-- per-message span.+--+-- Kafka does not expose a per-message delivery counter through this consumer+-- API, so 'Shibuya.Core.Types.Envelope.attempt' is always 'Nothing'. Handlers+-- cannot safely cap retries by counting attempts from the envelope; use an+-- external store, or return @AckHalt@ to stop the stream.+--+-- == Fatal Error Propagation+--+-- Non-fatal Kafka errors (poll timeouts, partition EOFs, and the rest of the+-- non-fatal set defined by @hw-kafka-streamly@'s 'Kafka.Streamly.Stream.isFatal')+-- 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'.+--+-- == AckHalt Partition Pause Semantics+--+-- @AckHalt@ pauses the originating partition by calling @pausePartitions@ from+-- @kafka-effectful@ and the processor stops. Polling therefore stops. After+-- @max.poll.interval.ms@ (librdkafka default: 300000 ms, or 5 minutes) the broker+-- may evict this consumer from its group and rebalance the partition to another+-- member, which resumes from the last committed offset. A single-member group+-- simply stalls until restart. Paused state is session-local and does not outlive+-- the current consumer.+--+-- == Rebalance Callback Helper+--+-- '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.+module Shibuya.Adapter.Kafka+ ( -- * Adapter kafkaAdapter, kafkaAdapterWith, KafkaAdapterState,@@ -111,7 +110,7 @@ Timeout (..), BatchSize (..), KafkaError,-)+ ) where import Control.Concurrent.STM (atomically, writeTVar)@@ -133,101 +132,98 @@ import Shibuya.Adapter.Kafka.Internal (KafkaAdapterState (..), dropStaleRecords, ingestedStream, kafkaSource, mkIngested, newKafkaAdapterState, withConsumerLock) import System.IO (hPutStrLn, stderr) -{- | Create a Kafka adapter with the given configuration.--The adapter operates within an existing 'KafkaConsumer' effect scope.-Consumer lifecycle (connection, group membership) is managed by-@runKafkaConsumer@ from kafka-effectful.--The adapter uses @noAutoOffsetStore@ with manual @storeOffsetMessage@ +-auto-commit for offset management. On shutdown, @commitAllOffsets@ flushes-offsets stored so far. Messages finalized during the drain window store offsets-after that explicit commit; let the surrounding @runKafkaConsumer@ scope end-normally after 'Shibuya.App.stopApp' returns so the consumer close path can-flush the final stored offsets under the same auto-commit mode.--The returned 'Shibuya.Adapter.Adapter.shutdown' action must be invoked while-the 'KafkaConsumer' effect is still in scope. Invoking it after-@runKafkaConsumer@ has returned will throw a 'KafkaError' from-@commitAllOffsets@ against a consumer that is no longer valid. This is a-caller-side invariant; the adapter does not catch the error.--}+-- | Create a Kafka adapter with the given configuration.+--+-- The adapter operates within an existing 'KafkaConsumer' effect scope.+-- Consumer lifecycle (connection, group membership) is managed by+-- @runKafkaConsumer@ from kafka-effectful.+--+-- The adapter uses @noAutoOffsetStore@ with manual @storeOffsetMessage@ ++-- auto-commit for offset management. On shutdown, @commitAllOffsets@ flushes+-- offsets stored so far. Messages finalized during the drain window store offsets+-- after that explicit commit; let the surrounding @runKafkaConsumer@ scope end+-- normally after 'Shibuya.App.stopApp' returns so the consumer close path can+-- flush the final stored offsets under the same auto-commit mode.+--+-- The returned 'Shibuya.Adapter.Adapter.shutdown' action must be invoked while+-- the 'KafkaConsumer' effect is still in scope. Invoking it after+-- @runKafkaConsumer@ has returned will throw a 'KafkaError' from+-- @commitAllOffsets@ against a consumer that is no longer valid. This is a+-- caller-side invariant; the adapter does not catch the error. kafkaAdapter ::- (KafkaConsumer :> es, Error KafkaError :> es, IOE :> es) =>- KafkaAdapterConfig ->- Eff es (Adapter es (Maybe ByteString))+ (KafkaConsumer :> es, Error KafkaError :> es, IOE :> es) =>+ KafkaAdapterConfig ->+ Eff es (Adapter es (Maybe ByteString)) kafkaAdapter config = do- state <- liftIO newKafkaAdapterState- kafkaAdapterWith state config--{- | Create a Kafka adapter using caller-owned adapter state.+ state <- liftIO newKafkaAdapterState+ kafkaAdapterWith state config -Use this when the same @KafkaAdapterState@ must also be referenced by-'kafkaRebalanceHandler', which is installed in consumer properties before the-consumer is created.--}+-- | Create a Kafka adapter using caller-owned adapter state.+--+-- Use this when the same @KafkaAdapterState@ must also be referenced by+-- 'kafkaRebalanceHandler', which is installed in consumer properties before the+-- consumer is created. kafkaAdapterWith ::- (KafkaConsumer :> es, Error KafkaError :> es, IOE :> es) =>- KafkaAdapterState ->- KafkaAdapterConfig ->- Eff es (Adapter es (Maybe ByteString))+ (KafkaConsumer :> es, Error KafkaError :> es, IOE :> es) =>+ KafkaAdapterState ->+ KafkaAdapterConfig ->+ Eff es (Adapter es (Maybe ByteString)) kafkaAdapterWith state config = do- warnOnSubscriptionMismatch config- let messageSource =- ingestedStream (mkIngested state config) $- dropStaleRecords state $- kafkaSource state config- pure- Adapter- { adapterName = "kafka:" <> Text.intercalate "," (map unTopicName config.topics)- , source = messageSource- , shutdown = do- liftIO $ atomically $ writeTVar state.shutdownVar True- withConsumerLock state (commitAllOffsets OffsetCommit)- `catchError` \_ err -> case err of- KafkaResponseError RdKafkaRespErrNoOffset -> pure ()- _ -> throwError err- }+ warnOnSubscriptionMismatch config+ let messageSource =+ ingestedStream (mkIngested state config) $+ dropStaleRecords state $+ kafkaSource state config+ pure+ Adapter+ { adapterName = "kafka:" <> Text.intercalate "," (map unTopicName config.topics),+ source = messageSource,+ shutdown = do+ liftIO $ atomically $ writeTVar state.shutdownVar True+ withConsumerLock state (commitAllOffsets OffsetCommit)+ `catchError` \_ err -> case err of+ KafkaResponseError RdKafkaRespErrNoOffset -> pure ()+ _ -> throwError err+ } warnOnSubscriptionMismatch ::- (KafkaConsumer :> es, IOE :> es) =>- KafkaAdapterConfig ->- Eff es ()+ (KafkaConsumer :> es, IOE :> es) =>+ KafkaAdapterConfig ->+ Eff es () warnOnSubscriptionMismatch config = do- liveSubscription <- subscription- let configured = Set.fromList config.topics- subscribed = Set.fromList (map fst liveSubscription)- if configured == subscribed- then pure ()- else- liftIO $- hPutStrLn stderr $- "[shibuya-kafka-adapter] WARNING: config topics differ from live Kafka subscription; configured="- <> show (Set.toList configured)- <> " subscribed="- <> show (Set.toList subscribed)--{- | Rebalance callback helper for caller-installed Kafka callbacks.+ liveSubscription <- subscription+ let configured = Set.fromList config.topics+ subscribed = Set.fromList (map fst liveSubscription)+ if configured == subscribed+ then pure ()+ else+ liftIO $+ hPutStrLn stderr $+ "[shibuya-kafka-adapter] WARNING: config topics differ from live Kafka subscription; configured="+ <> show (Set.toList configured)+ <> " subscribed="+ <> show (Set.toList subscribed) -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.--}+-- | Rebalance callback helper for caller-installed Kafka callbacks.+--+-- 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. kafkaRebalanceHandler ::- KafkaAdapterState ->- KC.KafkaConsumer ->- RebalanceEvent ->- IO ()+ KafkaAdapterState ->+ KC.KafkaConsumer ->+ RebalanceEvent ->+ IO () kafkaRebalanceHandler state _consumer event = do- hPutStrLn stderr $ "[shibuya-kafka-adapter] rebalance: " <> show event- case event of- RebalanceRevoke revoked ->- clearRevokedBarriers revoked- _ ->- pure ()+ 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, ())+ atomicModifyIORef' state.seekBarrier $ \barriers ->+ (foldr Map.delete barriers revoked, ())
src/Shibuya/Adapter/Kafka/Config.hs view
@@ -1,47 +1,44 @@ -- | Configuration types for the Kafka adapter.-module Shibuya.Adapter.Kafka.Config (- -- * Main Configuration+module Shibuya.Adapter.Kafka.Config+ ( -- * Main Configuration KafkaAdapterConfig (..), -- * Defaults defaultConfig,-)+ ) where import GHC.Generics (Generic) import Kafka.Types (BatchSize (..), Timeout (..), TopicName) -{- | Configuration for the Kafka adapter.--Consumer properties (brokers, group ID, etc.) are provided when running-@runKafkaConsumer@ — the adapter operates /within/ the @KafkaConsumer@-effect scope, not outside it.--}+-- | Configuration for the Kafka adapter.+--+-- Consumer properties (brokers, group ID, etc.) are provided when running+-- @runKafkaConsumer@ — the adapter operates /within/ the @KafkaConsumer@+-- effect scope, not outside it. data KafkaAdapterConfig = KafkaAdapterConfig- { topics :: ![TopicName]- {- ^ Topics expected by the adapter, used for observability metadata and- checked against the live consumer subscription at construction. The- actual subscription, including offset-reset policy, is supplied to- @runKafkaConsumer@ by the caller.- -}- , pollTimeout :: !Timeout- -- ^ Timeout for each poll call (default: 1000ms)- , batchSize :: !BatchSize- -- ^ Maximum messages per poll batch (default: 100)- }- deriving stock (Show, Eq, Generic)--{- | Default adapter configuration for the given topics.--Defaults:+ { -- | Topics expected by the adapter, used for observability metadata and+ -- checked against the live consumer subscription at construction. The+ -- actual subscription, including offset-reset policy, is supplied to+ -- @runKafkaConsumer@ by the caller.+ topics :: ![TopicName],+ -- | Timeout for each poll call (default: 1000ms)+ pollTimeout :: !Timeout,+ -- | Maximum messages per poll batch (default: 100)+ batchSize :: !BatchSize+ }+ deriving stock (Show, Eq, Generic) -* @pollTimeout@: 1000ms-* @batchSize@: 100--}+-- | Default adapter configuration for the given topics.+--+-- Defaults:+--+-- * @pollTimeout@: 1000ms+-- * @batchSize@: 100 defaultConfig :: [TopicName] -> KafkaAdapterConfig defaultConfig ts =- KafkaAdapterConfig- { topics = ts- , pollTimeout = Timeout 1000- , batchSize = BatchSize 100- }+ KafkaAdapterConfig+ { topics = ts,+ pollTimeout = Timeout 1000,+ batchSize = BatchSize 100+ }
src/Shibuya/Adapter/Kafka/Convert.hs view
@@ -1,6 +1,6 @@ -- | Type conversions between Kafka and Shibuya types.-module Shibuya.Adapter.Kafka.Convert (- -- * Message Conversion+module Shibuya.Adapter.Kafka.Convert+ ( -- * Message Conversion consumerRecordToEnvelope, -- * Trace Context@@ -9,7 +9,7 @@ -- * Timestamp Conversion timestampToUTCTime,-)+ ) where import Data.ByteString (ByteString)@@ -21,109 +21,103 @@ import Data.Time (UTCTime) import Data.Time.Clock.POSIX (posixSecondsToUTCTime) import Kafka.Consumer.Types (ConsumerRecord (..), Offset (..), Timestamp (..))-import Kafka.Types (- Headers,+import Kafka.Types+ ( Headers, Millis (..), PartitionId (..), TopicName (..), headersToList,- )+ ) import OpenTelemetry.Attributes (Attribute, toAttribute, unkey) import OpenTelemetry.SemanticConventions qualified as Sem import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..), TraceHeaders, mkEnvelope) -{- | Convert a Kafka 'ConsumerRecord' to a Shibuya 'Envelope'.--Field mapping:--* @messageId@: @\"{topic}-{partition}-{offset}\"@ (globally unique within a cluster)-* @cursor@: @CursorInt offset@-* @partition@: @Just (show partitionId)@-* @enqueuedAt@: converted from Kafka timestamp if available-* @traceContext@: extracted from @traceparent@/@tracestate@ headers-* @headers@: every Kafka header verbatim (ordered, duplicates- preserved) via @headersToList@; @Just []@ when the record carried- no headers. The W3C trace headers appear here in addition to their- parsed form in @traceContext@.-* @attempt@: 'Nothing' (Kafka does not expose a redelivery counter)-* @attributes@: kafka-typed OTel attributes (system, partition,- offset). The framework's @processOne@ adds these to its- Consumer-kind span and the @messaging.system@ key overrides- the framework default of @"shibuya"@. Other @messaging.*@- attributes (destination.name, operation, message.id) are- populated by the framework from the @ProcessorId@/@MessageId@- pair and are not duplicated here.-* @payload@: the @crValue@ field (@Maybe ByteString@)--}+-- | Convert a Kafka 'ConsumerRecord' to a Shibuya 'Envelope'.+--+-- Field mapping:+--+-- * @messageId@: @\"{topic}-{partition}-{offset}\"@ (globally unique within a cluster)+-- * @cursor@: @CursorInt offset@+-- * @partition@: @Just (show partitionId)@+-- * @enqueuedAt@: converted from Kafka timestamp if available+-- * @traceContext@: extracted from @traceparent@/@tracestate@ headers+-- * @headers@: every Kafka header verbatim (ordered, duplicates+-- preserved) via @headersToList@; @Just []@ when the record carried+-- no headers. The W3C trace headers appear here in addition to their+-- parsed form in @traceContext@.+-- * @attempt@: 'Nothing' (Kafka does not expose a redelivery counter)+-- * @attributes@: kafka-typed OTel attributes (system, partition,+-- offset). The framework's @processOne@ adds these to its+-- Consumer-kind span and the @messaging.system@ key overrides+-- the framework default of @"shibuya"@. Other @messaging.*@+-- attributes (destination.name, operation, message.id) are+-- populated by the framework from the @ProcessorId@/@MessageId@+-- pair and are not duplicated here.+-- * @payload@: the @crValue@ field (@Maybe ByteString@) consumerRecordToEnvelope ::- ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->- Envelope (Maybe ByteString)+ ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->+ Envelope (Maybe ByteString) consumerRecordToEnvelope cr =- let headerList = headersToList cr.crHeaders- in (mkEnvelope (mkMessageId cr.crTopic cr.crPartition cr.crOffset) cr.crValue)- { cursor = Just (CursorInt (fromIntegral (unOffset cr.crOffset)))- , partition = Just (Text.pack (show (unPartitionId cr.crPartition)))- , enqueuedAt = timestampToUTCTime cr.crTimestamp- , traceContext = extractTraceHeadersFromList headerList- , headers = Just headerList- , attempt = Nothing- , attributes = kafkaSpanAttributes cr.crPartition cr.crOffset- }--{- | OpenTelemetry attributes that the framework's per-message span-should carry for a Kafka-sourced envelope.+ let headerList = headersToList cr.crHeaders+ in (mkEnvelope (mkMessageId cr.crTopic cr.crPartition cr.crOffset) cr.crValue)+ { cursor = Just (CursorInt (fromIntegral (unOffset cr.crOffset))),+ partition = Just (Text.pack (show (unPartitionId cr.crPartition))),+ enqueuedAt = timestampToUTCTime cr.crTimestamp,+ traceContext = extractTraceHeadersFromList headerList,+ headers = Just headerList,+ attempt = Nothing,+ attributes = kafkaSpanAttributes cr.crPartition cr.crOffset+ } -* @messaging.system@ — overrides the framework default- @"shibuya"@ to @"kafka"@.-* @messaging.kafka.destination.partition@ — Int64; from- @PartitionId@.-* @messaging.kafka.message.offset@ — Int64; from @Offset@.--}+-- | OpenTelemetry attributes that the framework's per-message span+-- should carry for a Kafka-sourced envelope.+--+-- * @messaging.system@ — overrides the framework default+-- @"shibuya"@ to @"kafka"@.+-- * @messaging.kafka.destination.partition@ — Int64; from+-- @PartitionId@.+-- * @messaging.kafka.message.offset@ — Int64; from @Offset@. kafkaSpanAttributes :: PartitionId -> Offset -> HashMap Text Attribute kafkaSpanAttributes (PartitionId pid) (Offset off) =- HashMap.fromList- [ ("messaging.system", toAttribute ("kafka" :: Text))- ,- ( unkey Sem.messaging_kafka_destination_partition- , toAttribute (fromIntegral pid :: Int64)- )- ,- ( unkey Sem.messaging_kafka_message_offset- , toAttribute (off :: Int64)- )- ]+ HashMap.fromList+ [ ("messaging.system", toAttribute ("kafka" :: Text)),+ ( unkey Sem.messaging_kafka_destination_partition,+ toAttribute (fromIntegral pid :: Int64)+ ),+ ( unkey Sem.messaging_kafka_message_offset,+ toAttribute (off :: Int64)+ )+ ] -- | Build a globally unique message ID from topic, partition, and offset. mkMessageId :: TopicName -> PartitionId -> Offset -> MessageId mkMessageId (TopicName topic) (PartitionId pid) (Offset off) =- MessageId (topic <> "-" <> Text.pack (show pid) <> "-" <> Text.pack (show off))--{- | Extract W3C trace headers from Kafka message headers.+ MessageId (topic <> "-" <> Text.pack (show pid) <> "-" <> Text.pack (show off)) -Looks for @traceparent@ and @tracestate@ header keys.-Returns 'Nothing' if @traceparent@ is not present (it's required for valid context).--}+-- | Extract W3C trace headers from Kafka message headers.+--+-- Looks for @traceparent@ and @tracestate@ header keys.+-- Returns 'Nothing' if @traceparent@ is not present (it's required for valid context). extractTraceHeaders :: Headers -> Maybe TraceHeaders extractTraceHeaders = extractTraceHeadersFromList . headersToList -- | Extract W3C trace headers from an already materialized Kafka header list. extractTraceHeadersFromList :: [(ByteString, ByteString)] -> Maybe TraceHeaders extractTraceHeadersFromList headerList =- case (lookup "traceparent" headerList, lookup "tracestate" headerList) of- (Nothing, _) -> Nothing- (Just tp, Nothing) -> Just [("traceparent", tp)]- (Just tp, Just ts) -> Just [("traceparent", tp), ("tracestate", ts)]--{- | Convert a Kafka 'Timestamp' to 'UTCTime'.+ case (lookup "traceparent" headerList, lookup "tracestate" headerList) of+ (Nothing, _) -> Nothing+ (Just tp, Nothing) -> Just [("traceparent", tp)]+ (Just tp, Just ts) -> Just [("traceparent", tp), ("tracestate", ts)] -Returns 'Nothing' for 'NoTimestamp'.-Both 'CreateTime' and 'LogAppendTime' carry milliseconds since Unix epoch.--}+-- | Convert a Kafka 'Timestamp' to 'UTCTime'.+--+-- Returns 'Nothing' for 'NoTimestamp'.+-- Both 'CreateTime' and 'LogAppendTime' carry milliseconds since Unix epoch. timestampToUTCTime :: Timestamp -> Maybe UTCTime timestampToUTCTime = \case- CreateTime (Millis ms) -> Just (millisToUTCTime ms)- LogAppendTime (Millis ms) -> Just (millisToUTCTime ms)- NoTimestamp -> Nothing+ CreateTime (Millis ms) -> Just (millisToUTCTime ms)+ LogAppendTime (Millis ms) -> Just (millisToUTCTime ms)+ NoTimestamp -> Nothing -- | Convert milliseconds since Unix epoch to UTCTime. millisToUTCTime :: Int64 -> UTCTime
src/Shibuya/Adapter/Kafka/Internal.hs view
@@ -1,8 +1,7 @@-{- | Internal implementation details for the Kafka adapter.-This module is not part of the public API and may change without notice.--}-module Shibuya.Adapter.Kafka.Internal (- -- * Adapter State+-- | Internal implementation details for the Kafka adapter.+-- This module is not part of the public API and may change without notice.+module Shibuya.Adapter.Kafka.Internal+ ( -- * Adapter State KafkaAdapterState (..), newKafkaAdapterState, withConsumerLock,@@ -17,7 +16,7 @@ -- * AckHandle Construction mkAckHandle,-)+ ) where import Control.Concurrent (threadDelay)@@ -35,13 +34,13 @@ import Effectful.Error.Static (Error, catchError, throwError) import Effectful.Exception qualified as Exception import Kafka.Consumer.Types (ConsumerRecord (..), Offset (..), PartitionOffset (..), TopicPartition (..))-import Kafka.Effectful.Consumer.Effect (- KafkaConsumer,+import Kafka.Effectful.Consumer.Effect+ ( KafkaConsumer, pausePartitions, pollMessageBatch, seekPartitions, storeOffsetMessage,- )+ ) import Kafka.Streamly.Stream (isFatal, skipNonFatal) import Kafka.Types (KafkaError, PartitionId, Timeout (..), TopicName) import Shibuya.Adapter.Kafka.Config (KafkaAdapterConfig (..))@@ -58,242 +57,233 @@ -- | Mutable state shared by the source stream and ack handles. data KafkaAdapterState = KafkaAdapterState- { shutdownVar :: !(TVar Bool)- , seekBarrier :: !(IORef (Map PartitionKey Offset))- , fatalError :: !(IORef (Maybe KafkaError))- , consumerLock :: !(MVar ())- {- ^ Serializes every librdkafka consumer operation. Under- 'Shibuya.App.runApp' the consumer handle is shared between the ingester- thread (which polls) and the processor thread (which seeks, stores,- pauses, and commits during finalize). Running @rd_kafka_consume_batch_queue@- concurrently with a seek/store on the same handle corrupts librdkafka's- internal fetch queue and crashes with a native SIGSEGV, so every consumer- call is wrapped in this mutex. Each poll is additionally bounded (see- @maxPollHoldMillis@) so the lock is released frequently and finalize is- never starved.- -}- }+ { shutdownVar :: !(TVar Bool),+ seekBarrier :: !(IORef (Map PartitionKey Offset)),+ fatalError :: !(IORef (Maybe KafkaError)),+ -- | Serializes every librdkafka consumer operation. Under+ -- 'Shibuya.App.runApp' the consumer handle is shared between the ingester+ -- thread (which polls) and the processor thread (which seeks, stores,+ -- pauses, and commits during finalize). Running @rd_kafka_consume_batch_queue@+ -- concurrently with a seek/store on the same handle corrupts librdkafka's+ -- internal fetch queue and crashes with a native SIGSEGV, so every consumer+ -- call is wrapped in this mutex. Each poll is additionally bounded (see+ -- @maxPollHoldMillis@) so the lock is released frequently and finalize is+ -- never starved.+ consumerLock :: !(MVar ())+ } -{- | Allocate mutable state shared by the Kafka source, ack handles, and-optional rebalance callback.--}+-- | Allocate mutable state shared by the Kafka source, ack handles, and+-- optional rebalance callback. newKafkaAdapterState :: IO KafkaAdapterState newKafkaAdapterState =- KafkaAdapterState- <$> newTVarIO False- <*> newIORef Map.empty- <*> newIORef Nothing- <*> newMVar ()+ KafkaAdapterState+ <$> newTVarIO False+ <*> newIORef Map.empty+ <*> newIORef Nothing+ <*> newMVar () -{- | 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-released, including when the action throws through the 'Error' @KafkaError@-effect.--}+-- | 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+-- released, including when the action throws through the 'Error' @KafkaError@+-- effect. withConsumerLock :: (IOE :> es) => KafkaAdapterState -> Eff es a -> Eff es a withConsumerLock state =- Exception.bracket_- (Effectful.liftIO (takeMVar state.consumerLock))- (Effectful.liftIO (putMVar state.consumerLock ()))+ Exception.bracket_+ (Effectful.liftIO (takeMVar state.consumerLock))+ (Effectful.liftIO (putMVar state.consumerLock ())) -{- | Upper bound (milliseconds) on how long any single blocking consumer call-may hold the consumer lock. Every consumer call runs under 'consumerLock' so-the poll never races a concurrent seek/store on the shared handle (see-'consumerLock'). A call that blocks for its whole timeout — an empty poll, or a-seek that cannot complete — would then hold the lock for that long and starve-(and, under GHC's deadlock detector, wedge) the finalize path that must seek to-redeliver a retried message. Capping the poll and seek timeouts at this value-keeps the lock available roughly every @maxPollHoldMillis@ so finalize can-interleave and the ingester promptly re-polls to observe redelivered records.-It also bounds shutdown latency. The cap leaves ample headroom over a healthy-seek (which acknowledges in single-digit milliseconds), so it only bites when a-call is genuinely stuck — in which case surfacing it fast (via the ack path's-bounded retry and fatal slot) is the desired behavior.--}+-- | Upper bound (milliseconds) on how long any single blocking consumer call+-- may hold the consumer lock. Every consumer call runs under 'consumerLock' so+-- the poll never races a concurrent seek/store on the shared handle (see+-- 'consumerLock'). A call that blocks for its whole timeout — an empty poll, or a+-- seek that cannot complete — would then hold the lock for that long and starve+-- (and, under GHC's deadlock detector, wedge) the finalize path that must seek to+-- redeliver a retried message. Capping the poll and seek timeouts at this value+-- keeps the lock available roughly every @maxPollHoldMillis@ so finalize can+-- interleave and the ingester promptly re-polls to observe redelivered records.+-- It also bounds shutdown latency. The cap leaves ample headroom over a healthy+-- seek (which acknowledges in single-digit milliseconds), so it only bites when a+-- call is genuinely stuck — in which case surfacing it fast (via the ack path's+-- bounded retry and fatal slot) is the desired behavior. maxPollHoldMillis :: Int maxPollHoldMillis = 100 -{- | Cap a caller-supplied timeout at 'maxPollHoldMillis' so a single blocking-consumer call cannot hold the consumer lock longer than that. Applied to both-the poll and the retry seek.--}+-- | Cap a caller-supplied timeout at 'maxPollHoldMillis' so a single blocking+-- consumer call cannot hold the consumer lock longer than that. Applied to both+-- the poll and the retry seek. boundedLockTimeout :: Timeout -> Timeout boundedLockTimeout t = Timeout (min (unTimeout t) maxPollHoldMillis) -{- | Create a stream of 'ConsumerRecord's by repeatedly polling the broker.--Calls 'pollMessageBatch' in a loop under 'consumerLock', preserving errors as-@Left@ values. Each poll uses a timeout capped at @maxPollHoldMillis@ so the-lock stays available to the finalize path (which seeks, stores, pauses, and-commits). Non-fatal errors (timeouts, partition EOF, etc.) are filtered out-via 'skipNonFatal' from hw-kafka-streamly. Fatal errors are preserved for-upstream handling; a fatal error recorded by the ack path (see 'fatalError')-terminates the stream.--}+-- | Create a stream of 'ConsumerRecord's by repeatedly polling the broker.+--+-- Calls 'pollMessageBatch' in a loop under 'consumerLock', preserving errors as+-- @Left@ values. Each poll uses a timeout capped at @maxPollHoldMillis@ so the+-- lock stays available to the finalize path (which seeks, stores, pauses, and+-- commits). Non-fatal errors (timeouts, partition EOF, etc.) are filtered out+-- via 'skipNonFatal' from hw-kafka-streamly. Fatal errors are preserved for+-- upstream handling; a fatal error recorded by the ack path (see 'fatalError')+-- terminates the stream. kafkaSource ::- (KafkaConsumer :> es, Error KafkaError :> es, IOE :> es) =>- KafkaAdapterState ->- KafkaAdapterConfig ->- Stream (Eff es) (Either KafkaError (ConsumerRecord (Maybe ByteString) (Maybe ByteString)))+ (KafkaConsumer :> es, Error KafkaError :> es, IOE :> es) =>+ KafkaAdapterState ->+ KafkaAdapterConfig ->+ Stream (Eff es) (Either KafkaError (ConsumerRecord (Maybe ByteString) (Maybe ByteString))) kafkaSource state config =- skipNonFatal $- Stream.unfoldrM step ()- & Stream.concatMap Stream.fromList+ skipNonFatal $+ Stream.unfoldrM step ()+ & Stream.concatMap Stream.fromList where pollT = boundedLockTimeout config.pollTimeout step () = do- mbFatal <- Effectful.liftIO $ readIORef state.fatalError- case mbFatal of- Just err -> throwError err- Nothing -> pure ()- isShutdown <- Effectful.liftIO $ readTVarIO state.shutdownVar- if isShutdown- then pure Nothing- else do- batch <- withConsumerLock state (pollMessageBatch pollT config.batchSize)- pure (Just (batch, ()))+ mbFatal <- Effectful.liftIO $ readIORef state.fatalError+ case mbFatal of+ Just err -> throwError err+ Nothing -> pure ()+ isShutdown <- Effectful.liftIO $ readTVarIO state.shutdownVar+ if isShutdown+ then pure Nothing+ else do+ batch <- withConsumerLock state (pollMessageBatch pollT config.batchSize)+ pure (Just (batch, ())) -- | Drop records already buffered above a pending retry barrier. dropStaleRecords ::- (IOE :> es) =>- KafkaAdapterState ->- Stream (Eff es) (Either KafkaError (ConsumerRecord (Maybe ByteString) (Maybe ByteString))) ->- Stream (Eff es) (Either KafkaError (ConsumerRecord (Maybe ByteString) (Maybe ByteString)))+ (IOE :> es) =>+ KafkaAdapterState ->+ Stream (Eff es) (Either KafkaError (ConsumerRecord (Maybe ByteString) (Maybe ByteString))) ->+ Stream (Eff es) (Either KafkaError (ConsumerRecord (Maybe ByteString) (Maybe ByteString))) dropStaleRecords state =- 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--{- | Create an 'AckHandle' for a single 'ConsumerRecord'.--Maps 'AckDecision' to Kafka operations:+ 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 -* 'AckOk' -> 'storeOffsetMessage' (mark offset ready for commit)-* 'AckRetry' -> record seek barrier and seek partition back to the failed offset-* 'AckDeadLetter' -> warn to stderr, then 'storeOffsetMessage' (DLQ deferred)-* 'AckHalt' -> 'pausePartitions' (do NOT store offset; message will be re-consumed)--}+-- | Create an 'AckHandle' for a single 'ConsumerRecord'.+--+-- Maps 'AckDecision' to Kafka operations:+--+-- * 'AckOk' -> 'storeOffsetMessage' (mark offset ready for commit)+-- * 'AckRetry' -> record seek barrier and seek partition back to the failed offset+-- * 'AckDeadLetter' -> warn to stderr, then 'storeOffsetMessage' (DLQ deferred)+-- * 'AckHalt' -> 'pausePartitions' (do NOT store offset; message will be re-consumed) mkAckHandle ::- (KafkaConsumer :> es, Error KafkaError :> es, IOE :> es) =>- KafkaAdapterState ->- KafkaAdapterConfig ->- ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->- AckHandle es+ (KafkaConsumer :> es, Error KafkaError :> es, IOE :> es) =>+ KafkaAdapterState ->+ KafkaAdapterConfig ->+ ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->+ AckHandle es mkAckHandle state config cr = AckHandle $ \case- AckOk ->- ackAttempt state (storeGuarded state cr)- 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)]))+ AckOk ->+ ackAttempt state (storeGuarded state cr)+ 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)])) ackAttempt ::- (Error KafkaError :> es, IOE :> es) =>- KafkaAdapterState ->- Eff es () ->- Eff es ()+ (Error KafkaError :> es, IOE :> es) =>+ KafkaAdapterState ->+ Eff es () ->+ Eff es () ackAttempt state action = go (1 :: Int) where maxAttempts = 3 retryDelayMicros = 50000 go attempt =- action `catchError` \_ err ->- if isFatal err || attempt >= maxAttempts- then Effectful.liftIO $ recordFatalError state err- else do- Effectful.liftIO $ threadDelay retryDelayMicros- go (attempt + 1)+ action `catchError` \_ err ->+ if isFatal err || attempt >= maxAttempts+ then Effectful.liftIO $ recordFatalError state err+ else do+ Effectful.liftIO $ threadDelay retryDelayMicros+ go (attempt + 1) recordFatalError :: KafkaAdapterState -> KafkaError -> IO () recordFatalError state err =- atomicModifyIORef' state.fatalError $ \case- Just existing -> (Just existing, ())- Nothing -> (Just err, ())+ atomicModifyIORef' state.fatalError $ \case+ Just existing -> (Just existing, ())+ Nothing -> (Just err, ()) storeGuarded ::- (KafkaConsumer :> es, IOE :> es) =>- KafkaAdapterState ->- ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->- Eff es ()+ (KafkaConsumer :> es, IOE :> es) =>+ KafkaAdapterState ->+ ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->+ 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 ()+ 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 () partitionKey :: ConsumerRecord k v -> PartitionKey partitionKey cr = (cr.crTopic, cr.crPartition) delayRetry :: NominalDiffTime -> IO () delayRetry delay- | delay <= 0 = pure ()- | otherwise = threadDelay (floor (realToFrac delay * (1000000 :: Double)))--{- | Combine conversion and ack handle to produce an 'Ingested'.+ | delay <= 0 = pure ()+ | otherwise = threadDelay (floor (realToFrac delay * (1000000 :: Double))) -Lease is always 'Nothing' for Kafka (no visibility timeout mechanism).--}+-- | Combine conversion and ack handle to produce an 'Ingested'.+--+-- Lease is always 'Nothing' for Kafka (no visibility timeout mechanism). mkIngested ::- (KafkaConsumer :> es, Error KafkaError :> es, IOE :> es) =>- KafkaAdapterState ->- KafkaAdapterConfig ->- ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->- Ingested es (Maybe ByteString)+ (KafkaConsumer :> es, Error KafkaError :> es, IOE :> es) =>+ KafkaAdapterState ->+ KafkaAdapterConfig ->+ ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->+ Ingested es (Maybe ByteString) mkIngested state config cr =- Core.mkIngested- (consumerRecordToEnvelope cr)- (mkAckHandle state config cr)--{- | Transform a poll stream of @Either KafkaError ConsumerRecord@ into a-stream of 'Ingested'.--A @Right cr@ is wrapped via the supplied builder (in production,-'mkIngested'). A @Left err@ that reaches this stage is fatal by construction-— 'Kafka.Streamly.Stream.skipNonFatal' has already dropped non-fatal errors-— and is thrown via the 'Error' @KafkaError@ effect, terminating the stream.+ Core.mkIngested+ (consumerRecordToEnvelope cr)+ (mkAckHandle state config cr) -Parameterizing over the builder function keeps this helper free of the-'KafkaConsumer' constraint, so it can be exercised in a unit test that-injects a synthetic @Left@ without standing up a real consumer.--}+-- | Transform a poll stream of @Either KafkaError ConsumerRecord@ into a+-- stream of 'Ingested'.+--+-- A @Right cr@ is wrapped via the supplied builder (in production,+-- 'mkIngested'). A @Left err@ that reaches this stage is fatal by construction+-- — 'Kafka.Streamly.Stream.skipNonFatal' has already dropped non-fatal errors+-- — and is thrown via the 'Error' @KafkaError@ effect, terminating the stream.+--+-- Parameterizing over the builder function keeps this helper free of the+-- 'KafkaConsumer' constraint, so it can be exercised in a unit test that+-- injects a synthetic @Left@ without standing up a real consumer. ingestedStream ::- (Error KafkaError :> es) =>- (ConsumerRecord (Maybe ByteString) (Maybe ByteString) -> Ingested es (Maybe ByteString)) ->- Stream (Eff es) (Either KafkaError (ConsumerRecord (Maybe ByteString) (Maybe ByteString))) ->- Stream (Eff es) (Ingested es (Maybe ByteString))+ (Error KafkaError :> es) =>+ (ConsumerRecord (Maybe ByteString) (Maybe ByteString) -> 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))- Left err -> throwError err+ Stream.mapMaybeM $ \case+ Right cr -> pure (Just (mkI cr))+ Left err -> throwError err
test/Kafka/TestEnv.hs view
@@ -1,6 +1,6 @@ -- | Test environment helpers for Kafka integration tests.-module Kafka.TestEnv (- -- * Test Environment+module Kafka.TestEnv+ ( -- * Test Environment TestEnv (..), withTestEnv, @@ -14,7 +14,7 @@ -- * Topic Management createTopic, createTopicWithPartitions,-)+ ) where import Control.Monad (forM_)@@ -27,30 +27,30 @@ import Kafka.Consumer.ConsumerProperties (ConsumerProperties) import Kafka.Consumer.Subscription (Subscription) import Kafka.Consumer.Types (ConsumerGroupId (..), OffsetCommit (..), OffsetReset (..))-import Kafka.Effectful.Consumer (- brokersList,+import Kafka.Effectful.Consumer+ ( brokersList, groupId, noAutoOffsetStore, offsetReset, runKafkaConsumer, topics,- )+ ) import Kafka.Effectful.Consumer.Effect (commitAllOffsets)-import Kafka.Effectful.Producer (- flushProducer,+import Kafka.Effectful.Producer+ ( flushProducer, produceMessage, runKafkaProducer,- )+ ) import Kafka.Effectful.Producer qualified as P import Kafka.Producer.ProducerProperties (ProducerProperties) import Kafka.Producer.Types (ProducePartition (..), ProducerRecord (..))-import Kafka.Types (- BatchSize (..),+import Kafka.Types+ ( BatchSize (..), BrokerAddress (..), KafkaError, Timeout (..), TopicName (..),- )+ ) import Shibuya.Adapter (Adapter (..)) import Shibuya.Adapter.Kafka (KafkaAdapterConfig (..), kafkaAdapter) import Shibuya.Core.Ack (AckDecision (..))@@ -64,30 +64,30 @@ -- | Test environment with isolated topic and group. data TestEnv = TestEnv- { testBroker :: !BrokerAddress- , testTopic :: !TopicName- , testGroupId :: !ConsumerGroupId- , testPrefix :: !String- }+ { testBroker :: !BrokerAddress,+ testTopic :: !TopicName,+ testGroupId :: !ConsumerGroupId,+ testPrefix :: !String+ } -- | Create a test environment with a random prefix for isolation. withTestEnv :: (TestEnv -> IO a) -> IO a withTestEnv f = do- prefix <- randomPrefix- let env =- TestEnv- { testBroker = BrokerAddress "127.0.0.1:9092"- , testTopic = TopicName (Text.pack (prefix <> "-topic"))- , testGroupId = ConsumerGroupId (Text.pack (prefix <> "-group"))- , testPrefix = prefix- }- f env+ prefix <- randomPrefix+ let env =+ TestEnv+ { testBroker = BrokerAddress "127.0.0.1:9092",+ testTopic = TopicName (Text.pack (prefix <> "-topic")),+ testGroupId = ConsumerGroupId (Text.pack (prefix <> "-group")),+ testPrefix = prefix+ }+ f env -- | Generate a random 10-character alphanumeric prefix. randomPrefix :: IO String randomPrefix = do- let chars = ['a' .. 'z'] ++ ['0' .. '9']- mapM (\_ -> do i <- randomRIO (0, length chars - 1); pure (chars !! i)) [1 .. 10 :: Int]+ let chars = ['a' .. 'z'] ++ ['0' .. '9']+ mapM (\_ -> do i <- randomRIO (0, length chars - 1); pure (chars !! i)) [1 .. 10 :: Int] -- | Create a topic with 1 partition via rpk. createTopic :: TestEnv -> IO ()@@ -96,95 +96,95 @@ -- | Create a topic with the specified number of partitions via rpk. createTopicWithPartitions :: TestEnv -> Int -> IO () createTopicWithPartitions env n =- callCommand $- "rpk topic create "- <> Text.unpack (unTopicName env.testTopic)- <> " -p "- <> show n- <> " 2>/dev/null || true"+ callCommand $+ "rpk topic create "+ <> Text.unpack (unTopicName env.testTopic)+ <> " -p "+ <> show n+ <> " 2>/dev/null || true" -- | Build consumer properties for testing. mkConsumerProps :: TestEnv -> ConsumerProperties mkConsumerProps env =- brokersList [env.testBroker]- <> groupId env.testGroupId- <> noAutoOffsetStore+ brokersList [env.testBroker]+ <> groupId env.testGroupId+ <> noAutoOffsetStore -- | Build producer properties for testing. mkProducerProps :: TestEnv -> ProducerProperties mkProducerProps env =- P.brokersList [env.testBroker]+ P.brokersList [env.testBroker] -- | Build subscription for testing. mkSubscription :: TestEnv -> Subscription mkSubscription env =- topics [env.testTopic]- <> offsetReset Earliest+ topics [env.testTopic]+ <> offsetReset Earliest -- | Produce messages with the given payloads to the test topic. produceMessages :: TestEnv -> [ByteString] -> IO () produceMessages env payloads = do- result <- runEff . runError @KafkaError $ do- runKafkaProducer (mkProducerProps env) $ do- forM_ payloads $ \payload ->- produceMessage- ProducerRecord- { prTopic = env.testTopic- , prPartition = UnassignedPartition- , prKey = Nothing- , prValue = Just payload- , prHeaders = mempty- }- flushProducer- case result of- Left err -> error $ "Failed to produce: " <> show err- Right () -> pure ()+ result <- runEff . runError @KafkaError $ do+ runKafkaProducer (mkProducerProps env) $ do+ forM_ payloads $ \payload ->+ produceMessage+ ProducerRecord+ { prTopic = env.testTopic,+ prPartition = UnassignedPartition,+ prKey = Nothing,+ prValue = Just payload,+ prHeaders = mempty+ }+ flushProducer+ case result of+ Left err -> error $ "Failed to produce: " <> show err+ Right () -> pure () -- | Produce keyed messages to the test topic. produceKeyedMessages :: TestEnv -> [(ByteString, ByteString)] -> IO () produceKeyedMessages env pairs = do- result <- runEff . runError @KafkaError $ do- runKafkaProducer (mkProducerProps env) $ do- forM_ pairs $ \(key, payload) ->- produceMessage- ProducerRecord- { prTopic = env.testTopic- , prPartition = UnassignedPartition- , prKey = Just key- , prValue = Just payload- , prHeaders = mempty- }- flushProducer- case result of- Left err -> error $ "Failed to produce: " <> show err- Right () -> pure ()+ result <- runEff . runError @KafkaError $ do+ runKafkaProducer (mkProducerProps env) $ do+ forM_ pairs $ \(key, payload) ->+ produceMessage+ ProducerRecord+ { prTopic = env.testTopic,+ prPartition = UnassignedPartition,+ prKey = Just key,+ prValue = Just payload,+ prHeaders = mempty+ }+ flushProducer+ case result of+ Left err -> error $ "Failed to produce: " <> show err+ Right () -> pure () -- | Consume N messages from the test topic via the adapter, applying the given ack decision. consumeN ::- TestEnv ->- Int ->- AckDecision ->- IO [Envelope (Maybe ByteString)]+ TestEnv ->+ Int ->+ AckDecision ->+ IO [Envelope (Maybe ByteString)] consumeN env n ackDecision = do- ref <- newIORef ([] :: [Envelope (Maybe ByteString)])- result <- runEff . runError @KafkaError $ do- runKafkaConsumer (mkConsumerProps env) (mkSubscription env) $ do- let config =- KafkaAdapterConfig- { topics = [env.testTopic]- , pollTimeout = Timeout 5000- , batchSize = BatchSize 100- }- Adapter{source} <- kafkaAdapter config- Stream.fold Fold.drain- $ Stream.mapM- ( \(Ingested{envelope, ack = AckHandle finalize}) -> do- liftIO $ modifyIORef' ref (envelope :)- finalize ackDecision- )- $ Stream.take n source- commitAllOffsets OffsetCommit- case result of- Left err -> error $ "Failed to consume: " <> show err- Right () -> pure ()- reverse <$> readIORef ref+ ref <- newIORef ([] :: [Envelope (Maybe ByteString)])+ result <- runEff . runError @KafkaError $ do+ runKafkaConsumer (mkConsumerProps env) (mkSubscription env) $ do+ let config =+ KafkaAdapterConfig+ { topics = [env.testTopic],+ pollTimeout = Timeout 5000,+ batchSize = BatchSize 100+ }+ Adapter {source} <- kafkaAdapter config+ Stream.fold Fold.drain+ $ Stream.mapM+ ( \(Ingested {envelope, ack = AckHandle finalize}) -> do+ liftIO $ modifyIORef' ref (envelope :)+ finalize ackDecision+ )+ $ Stream.take n source+ commitAllOffsets OffsetCommit+ case result of+ Left err -> error $ "Failed to consume: " <> show err+ Right () -> pure ()+ reverse <$> readIORef ref
test/Main.hs view
@@ -8,11 +8,11 @@ main :: IO () main =- defaultMain $- testGroup- "shibuya-kafka-adapter"- [ AckHandleTest.tests- , AdapterTest.tests- , ConvertTest.tests- , IntegrationTest.tests- ]+ defaultMain $+ testGroup+ "shibuya-kafka-adapter"+ [ AckHandleTest.tests,+ AdapterTest.tests,+ ConvertTest.tests,+ IntegrationTest.tests+ ]
test/Shibuya/Adapter/Kafka/AckHandleTest.hs view
@@ -21,226 +21,227 @@ import Test.Tasty.HUnit (assertEqual, assertFailure, testCase) data MockState = MockState- { storeAttempts :: !Int- , pauseAttempts :: !Int- , seekCalls :: ![TopicPartition]- , storeFailuresRemaining :: !Int- , pauseFailuresRemaining :: !Int- , seekFailuresRemaining :: !Int- , storeError :: !KafkaError- , pauseError :: !KafkaError- , seekError :: !KafkaError- }+ { storeAttempts :: !Int,+ pauseAttempts :: !Int,+ seekCalls :: ![TopicPartition],+ storeFailuresRemaining :: !Int,+ pauseFailuresRemaining :: !Int,+ seekFailuresRemaining :: !Int,+ storeError :: !KafkaError,+ pauseError :: !KafkaError,+ seekError :: !KafkaError+ } tests :: TestTree tests =- 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 "AckRetry seeks exact failed offset and does not store" testAckRetrySeeks- , testCase "seek barrier prevents stale successor store" testBarrierSkipsSuccessorStore- , testCase "source observes fatal slot before polling" testSourceObservesFatalSlot- ]+ 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 "AckRetry seeks exact failed offset and does not store" testAckRetrySeeks,+ testCase "seek barrier prevents stale successor store" testBarrierSkipsSuccessorStore,+ testCase "source observes fatal slot before polling" testSourceObservesFatalSlot+ ] testTransientStoreRetry :: IO () testTransientStoreRetry = do- mock <- newIORef defaultMockState{storeFailuresRemaining = 2}- state <- newKafkaAdapterState- result <- runFinalizer mock $ finalizeRecord state (recordAt 42) AckOk- assertRight result- final <- readIORef mock- fatal <- readIORef state.fatalError- assertEqual "store attempts" 3 final.storeAttempts- assertEqual "fatal slot" Nothing fatal+ mock <- newIORef defaultMockState {storeFailuresRemaining = 2}+ state <- newKafkaAdapterState+ result <- runFinalizer mock $ finalizeRecord state (recordAt 42) AckOk+ assertRight result+ final <- readIORef mock+ fatal <- readIORef state.fatalError+ assertEqual "store attempts" 3 final.storeAttempts+ assertEqual "fatal slot" Nothing fatal testPersistentStoreFailure :: IO () testPersistentStoreFailure = do- let err = KafkaResponseError RdKafkaRespErrTransport- mock <- newIORef defaultMockState{storeFailuresRemaining = 99, storeError = err}- state <- newKafkaAdapterState- result <- runFinalizer mock $ finalizeRecord state (recordAt 42) AckOk- assertRight result- final <- readIORef mock- fatal <- readIORef state.fatalError- assertEqual "store attempts" 3 final.storeAttempts- assertEqual "fatal slot" (Just err) fatal+ let err = KafkaResponseError RdKafkaRespErrTransport+ mock <- newIORef defaultMockState {storeFailuresRemaining = 99, storeError = err}+ state <- newKafkaAdapterState+ result <- runFinalizer mock $ finalizeRecord state (recordAt 42) AckOk+ assertRight result+ final <- readIORef mock+ fatal <- readIORef state.fatalError+ assertEqual "store attempts" 3 final.storeAttempts+ assertEqual "fatal slot" (Just err) fatal testFatalStoreFailure :: IO () testFatalStoreFailure = do- let err = KafkaBadConfiguration- mock <- newIORef defaultMockState{storeFailuresRemaining = 99, storeError = err}- state <- newKafkaAdapterState- result <- runFinalizer mock $ finalizeRecord state (recordAt 42) AckOk- assertRight result- final <- readIORef mock- fatal <- readIORef state.fatalError- assertEqual "store attempts" 1 final.storeAttempts- assertEqual "fatal slot" (Just err) fatal+ let err = KafkaBadConfiguration+ mock <- newIORef defaultMockState {storeFailuresRemaining = 99, storeError = err}+ state <- newKafkaAdapterState+ result <- runFinalizer mock $ finalizeRecord state (recordAt 42) AckOk+ assertRight result+ final <- readIORef mock+ fatal <- readIORef state.fatalError+ assertEqual "store attempts" 1 final.storeAttempts+ assertEqual "fatal slot" (Just err) fatal testAckHaltPauseFailure :: IO () testAckHaltPauseFailure = do- 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- final <- readIORef mock- fatal <- readIORef state.fatalError- assertEqual "pause attempts" 3 final.pauseAttempts- assertEqual "fatal slot" (Just err) fatal+ 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+ final <- readIORef mock+ fatal <- readIORef state.fatalError+ assertEqual "pause attempts" 3 final.pauseAttempts+ assertEqual "fatal slot" (Just err) fatal testAckRetrySeeks :: IO () testAckRetrySeeks = do- mock <- newIORef defaultMockState- state <- newKafkaAdapterState- result <- runFinalizer mock $ finalizeRecord state (recordAt 42) (AckRetry (RetryDelay 0))- assertRight result- final <- readIORef mock- assertEqual "store attempts" 0 final.storeAttempts- assertEqual- "seek call"- [TopicPartition (TopicName "orders") (PartitionId 0) (PartitionOffset 42)]- final.seekCalls+ mock <- newIORef defaultMockState+ state <- newKafkaAdapterState+ result <- runFinalizer mock $ finalizeRecord state (recordAt 42) (AckRetry (RetryDelay 0))+ assertRight result+ final <- readIORef mock+ assertEqual "store attempts" 0 final.storeAttempts+ assertEqual+ "seek call"+ [TopicPartition (TopicName "orders") (PartitionId 0) (PartitionOffset 42)]+ final.seekCalls testBarrierSkipsSuccessorStore :: IO () testBarrierSkipsSuccessorStore = do- mock <- newIORef defaultMockState- state <- newKafkaAdapterState- result <- runFinalizer mock $ do- finalizeRecord state (recordAt 42) (AckRetry (RetryDelay 0))- finalizeRecord state (recordAt 43) AckOk- finalizeRecord state (recordAt 42) AckOk- assertRight result- final <- readIORef mock- assertEqual "only retried message stored" 1 final.storeAttempts+ mock <- newIORef defaultMockState+ state <- newKafkaAdapterState+ result <- runFinalizer mock $ do+ finalizeRecord state (recordAt 42) (AckRetry (RetryDelay 0))+ finalizeRecord state (recordAt 43) AckOk+ finalizeRecord state (recordAt 42) AckOk+ assertRight result+ final <- readIORef mock+ assertEqual "only retried message stored" 1 final.storeAttempts testSourceObservesFatalSlot :: IO () testSourceObservesFatalSlot = do- let err = KafkaBadConfiguration- mock <- newIORef defaultMockState- state <- newKafkaAdapterState- atomicWriteIORef state.fatalError (Just err)- result <-- runEff . runErrorNoCallStack @KafkaError . runMockConsumer mock $- Stream.fold Fold.drain $- ingestedStream unreachableBuilder (kafkaSource state testConfig)- assertEqual "source error" (Left err) result+ let err = KafkaBadConfiguration+ mock <- newIORef defaultMockState+ state <- newKafkaAdapterState+ atomicWriteIORef state.fatalError (Just err)+ result <-+ runEff . runErrorNoCallStack @KafkaError . runMockConsumer mock $+ Stream.fold Fold.drain $+ ingestedStream unreachableBuilder (kafkaSource state testConfig)+ assertEqual "source error" (Left err) result runFinalizer ::- IORef MockState ->- Eff '[KafkaConsumer, Error KafkaError, IOE] a ->- IO (Either KafkaError a)+ IORef MockState ->+ Eff '[KafkaConsumer, Error KafkaError, IOE] a ->+ IO (Either KafkaError a) runFinalizer mock action =- runEff . runErrorNoCallStack @KafkaError . runMockConsumer mock $- action+ runEff . runErrorNoCallStack @KafkaError . runMockConsumer mock $+ action runMockConsumer ::- (IOE :> es, Error KafkaError :> es) =>- IORef MockState ->- Eff (KafkaConsumer : es) a ->- Eff es a+ (IOE :> es, Error KafkaError :> es) =>+ IORef MockState ->+ Eff (KafkaConsumer : es) a ->+ Eff es a runMockConsumer mock =- interpret $ \_env -> \case- StoreOffsetMessage _ -> attemptStore mock- PausePartitions _ -> attemptPause mock- SeekPartitions tps _ -> recordSeek mock tps- PollMessage _ -> error "AckHandleTest: PollMessage not exercised"- PollMessageBatch _ _ -> error "AckHandleTest: PollMessageBatch not exercised"- CommitOffsetMessage _ _ -> error "AckHandleTest: CommitOffsetMessage not exercised"- CommitAllOffsets _ -> error "AckHandleTest: CommitAllOffsets not exercised"- CommitPartitionsOffsets _ _ -> error "AckHandleTest: CommitPartitionsOffsets not exercised"- StoreOffsets _ -> error "AckHandleTest: StoreOffsets not exercised"- Assign _ -> error "AckHandleTest: Assign not exercised"- ResumePartitions _ -> error "AckHandleTest: ResumePartitions not exercised"- Committed _ _ -> error "AckHandleTest: Committed not exercised"- Position _ -> error "AckHandleTest: Position not exercised"- Assignment -> error "AckHandleTest: Assignment not exercised"- Subscription -> error "AckHandleTest: Subscription not exercised"- AskConsumerHandle -> error "AckHandleTest: AskConsumerHandle not exercised"+ interpret $ \_env -> \case+ StoreOffsetMessage _ -> attemptStore mock+ PausePartitions _ -> attemptPause mock+ SeekPartitions tps _ -> recordSeek mock tps+ PollMessage _ -> error "AckHandleTest: PollMessage not exercised"+ PollMessageBatch _ _ -> error "AckHandleTest: PollMessageBatch not exercised"+ PollMessageEither _ -> error "AckHandleTest: PollMessageEither not exercised"+ CommitOffsetMessage _ _ -> error "AckHandleTest: CommitOffsetMessage not exercised"+ CommitAllOffsets _ -> error "AckHandleTest: CommitAllOffsets not exercised"+ CommitPartitionsOffsets _ _ -> error "AckHandleTest: CommitPartitionsOffsets not exercised"+ StoreOffsets _ -> error "AckHandleTest: StoreOffsets not exercised"+ Assign _ -> error "AckHandleTest: Assign not exercised"+ ResumePartitions _ -> error "AckHandleTest: ResumePartitions not exercised"+ Committed _ _ -> error "AckHandleTest: Committed not exercised"+ Position _ -> error "AckHandleTest: Position not exercised"+ Assignment -> error "AckHandleTest: Assignment not exercised"+ Subscription -> error "AckHandleTest: Subscription not exercised"+ AskConsumerHandle -> error "AckHandleTest: AskConsumerHandle not exercised" unreachableBuilder ::- ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->- Ingested es (Maybe ByteString)+ ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->+ 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 <-- 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)- maybe (pure ()) throwError mbErr+ 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)+ maybe (pure ()) throwError mbErr attemptPause :: (IOE :> es, Error KafkaError :> es) => IORef MockState -> Eff es () attemptPause mock = do- mbErr <-- liftIO $- atomicModifyIORef' mock $ \s ->- let remaining = s.pauseFailuresRemaining- s' = s{pauseAttempts = s.pauseAttempts + 1, pauseFailuresRemaining = max 0 (remaining - 1)}- in (s', if remaining > 0 then Just s.pauseError else Nothing)- maybe (pure ()) throwError mbErr+ mbErr <-+ liftIO $+ atomicModifyIORef' mock $ \s ->+ let remaining = s.pauseFailuresRemaining+ s' = s {pauseAttempts = s.pauseAttempts + 1, pauseFailuresRemaining = max 0 (remaining - 1)}+ 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- mbErr <-- liftIO $- atomicModifyIORef' mock $ \s ->- let remaining = s.seekFailuresRemaining- s' = s{seekCalls = s.seekCalls <> tps, seekFailuresRemaining = max 0 (remaining - 1)}- in (s', if remaining > 0 then Just s.seekError else Nothing)- maybe (pure ()) throwError mbErr+ mbErr <-+ liftIO $+ atomicModifyIORef' mock $ \s ->+ let remaining = s.seekFailuresRemaining+ s' = s {seekCalls = s.seekCalls <> tps, seekFailuresRemaining = max 0 (remaining - 1)}+ in (s', if remaining > 0 then Just s.seekError else Nothing)+ maybe (pure ()) throwError mbErr finalizeRecord ::- (KafkaConsumer :> es, Error KafkaError :> es, IOE :> es) =>- KafkaAdapterState ->- ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->- AckDecision ->- Eff es ()+ (KafkaConsumer :> es, Error KafkaError :> es, IOE :> es) =>+ KafkaAdapterState ->+ ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->+ AckDecision ->+ Eff es () finalizeRecord state cr decision =- let AckHandle finalize = mkAckHandle state testConfig cr- in finalize decision+ let AckHandle finalize = mkAckHandle state testConfig cr+ in finalize decision recordAt :: Int64 -> ConsumerRecord (Maybe ByteString) (Maybe ByteString) recordAt offset =- ConsumerRecord- { crTopic = TopicName "orders"- , crPartition = PartitionId 0- , crOffset = Offset offset- , crTimestamp = NoTimestamp- , crHeaders = mempty- , crKey = Nothing- , crValue = Just "payload"- }+ ConsumerRecord+ { crTopic = TopicName "orders",+ crPartition = PartitionId 0,+ crOffset = Offset offset,+ crTimestamp = NoTimestamp,+ crHeaders = mempty,+ crKey = Nothing,+ crValue = Just "payload"+ } testConfig :: KafkaAdapterConfig testConfig =- KafkaAdapterConfig- { topics = [TopicName "orders"]- , pollTimeout = Timeout 100- , batchSize = BatchSize 100- }+ KafkaAdapterConfig+ { topics = [TopicName "orders"],+ pollTimeout = Timeout 100,+ batchSize = BatchSize 100+ } defaultMockState :: MockState defaultMockState =- MockState- { storeAttempts = 0- , pauseAttempts = 0- , seekCalls = []- , storeFailuresRemaining = 0- , pauseFailuresRemaining = 0- , seekFailuresRemaining = 0- , storeError = KafkaResponseError RdKafkaRespErrTransport- , pauseError = KafkaResponseError RdKafkaRespErrTransport- , seekError = KafkaResponseError RdKafkaRespErrTransport- }+ MockState+ { storeAttempts = 0,+ pauseAttempts = 0,+ seekCalls = [],+ storeFailuresRemaining = 0,+ pauseFailuresRemaining = 0,+ seekFailuresRemaining = 0,+ storeError = KafkaResponseError RdKafkaRespErrTransport,+ pauseError = KafkaResponseError RdKafkaRespErrTransport,+ seekError = KafkaResponseError RdKafkaRespErrTransport+ } assertRight :: (Show e) => Either e a -> IO () assertRight = \case- Left err -> assertFailure $ "expected Right, got Left: " <> show err- Right _ -> pure ()+ Left err -> assertFailure $ "expected Right, got Left: " <> show err+ Right _ -> pure ()
test/Shibuya/Adapter/Kafka/AdapterTest.hs view
@@ -1,7 +1,6 @@-{- | White-box tests for the Kafka adapter that do not require a running-broker. Exercises 'ingestedStream' with synthetic inputs to assert that-fatal 'KafkaError' values propagate through the 'Error' effect.--}+-- | White-box tests for the Kafka adapter that do not require a running+-- broker. Exercises 'ingestedStream' with synthetic inputs to assert that+-- fatal 'KafkaError' values propagate through the 'Error' effect. module Shibuya.Adapter.Kafka.AdapterTest (tests) where import Data.ByteString (ByteString)@@ -18,43 +17,42 @@ tests :: TestTree tests =- testGroup- "Adapter"- [ testCase "fatal KafkaError surfaces via Error effect" testFatalPropagation- , testCase "non-empty prefix of Rights then Left still surfaces fatal" testFatalAfterRights- ]+ testGroup+ "Adapter"+ [ testCase "fatal KafkaError surfaces via Error effect" testFatalPropagation,+ testCase "non-empty prefix of Rights then Left still surfaces fatal" testFatalAfterRights+ ] -{- | Builder that must never be evaluated in these tests. The input streams-contain only @Left@ values, so the @Right@ branch of 'ingestedStream'-never fires.--}+-- | Builder that must never be evaluated in these tests. The input streams+-- contain only @Left@ values, so the @Right@ branch of 'ingestedStream'+-- never fires. unreachableBuilder ::- ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->- Ingested es (Maybe ByteString)+ ConsumerRecord (Maybe ByteString) (Maybe ByteString) ->+ Ingested es (Maybe ByteString) unreachableBuilder _ = error "AdapterTest: Right branch should not be reached" testFatalPropagation :: IO () testFatalPropagation = do- let fatalErr = KafkaBadConfiguration- input :: [Either KafkaError (ConsumerRecord (Maybe ByteString) (Maybe ByteString))]- input = [Left fatalErr]- result <- runEff . runError @KafkaError $ do- Stream.fold Fold.drain $ ingestedStream unreachableBuilder (Stream.fromList input)- case result of- Left (_cs, err) -> assertEqual "propagated error" fatalErr err- Right () -> assertFailure "expected fatal error to propagate via Error effect"+ let fatalErr = KafkaBadConfiguration+ input :: [Either KafkaError (ConsumerRecord (Maybe ByteString) (Maybe ByteString))]+ input = [Left fatalErr]+ result <- runEff . runError @KafkaError $ do+ Stream.fold Fold.drain $ ingestedStream unreachableBuilder (Stream.fromList input)+ case result of+ Left (_cs, err) -> assertEqual "propagated error" fatalErr err+ Right () -> assertFailure "expected fatal error to propagate via Error effect" testFatalAfterRights :: IO () testFatalAfterRights = do- -- `skipNonFatal` has already filtered non-fatal errors upstream by the time- -- a stream reaches `ingestedStream`, so any Left here is fatal by- -- construction. This case confirms the stream aborts on the first fatal- -- Left — the second element must not be drawn.- let fatalErr = KafkaBadConfiguration- input :: [Either KafkaError (ConsumerRecord (Maybe ByteString) (Maybe ByteString))]- input = [Left fatalErr, Left (error "AdapterTest: second element must not be forced")]- result <- runEff . runError @KafkaError $ do- Stream.fold Fold.drain $ ingestedStream unreachableBuilder (Stream.fromList input)- case result of- Left (_cs, err) -> assertEqual "propagated error" fatalErr err- Right () -> assertFailure "expected fatal error to propagate via Error effect"+ -- `skipNonFatal` has already filtered non-fatal errors upstream by the time+ -- a stream reaches `ingestedStream`, so any Left here is fatal by+ -- construction. This case confirms the stream aborts on the first fatal+ -- Left — the second element must not be drawn.+ let fatalErr = KafkaBadConfiguration+ input :: [Either KafkaError (ConsumerRecord (Maybe ByteString) (Maybe ByteString))]+ input = [Left fatalErr, Left (error "AdapterTest: second element must not be forced")]+ result <- runEff . runError @KafkaError $ do+ Stream.fold Fold.drain $ ingestedStream unreachableBuilder (Stream.fromList input)+ case result of+ Left (_cs, err) -> assertEqual "propagated error" fatalErr err+ Right () -> assertFailure "expected fatal error to propagate via Error effect"
test/Shibuya/Adapter/Kafka/ConvertTest.hs view
@@ -4,157 +4,157 @@ import Data.HashMap.Strict qualified as HashMap import Data.Time.Clock.POSIX (posixSecondsToUTCTime) import Kafka.Consumer.Types (ConsumerRecord (..), Offset (..), Timestamp (..))-import Kafka.Types (- Headers,+import Kafka.Types+ ( Headers, Millis (..), PartitionId (..), TopicName (..), headersFromList,- )+ ) import OpenTelemetry.Attributes (Attribute (..), PrimitiveAttribute (..), unkey) import OpenTelemetry.SemanticConventions qualified as Sem-import Shibuya.Adapter.Kafka.Convert (- consumerRecordToEnvelope,+import Shibuya.Adapter.Kafka.Convert+ ( consumerRecordToEnvelope, extractTraceHeaders, extractTraceHeadersFromList, timestampToUTCTime,- )+ ) import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..)) import Test.Tasty (TestTree, testGroup) import Test.Tasty.HUnit (assertEqual, testCase) tests :: TestTree tests =- testGroup- "Convert"- [ testGroup "consumerRecordToEnvelope" envelopeTests- , testGroup "extractTraceHeaders" traceHeaderTests- , testGroup "timestampToUTCTime" timestampTests- ]+ testGroup+ "Convert"+ [ testGroup "consumerRecordToEnvelope" envelopeTests,+ testGroup "extractTraceHeaders" traceHeaderTests,+ testGroup "timestampToUTCTime" timestampTests+ ] -- | A minimal ConsumerRecord for testing. mkRecord ::- TopicName ->- PartitionId ->- Offset ->- Timestamp ->- Headers ->- Maybe ByteString ->- Maybe ByteString ->- ConsumerRecord (Maybe ByteString) (Maybe ByteString)+ TopicName ->+ PartitionId ->+ Offset ->+ Timestamp ->+ Headers ->+ Maybe ByteString ->+ Maybe ByteString ->+ ConsumerRecord (Maybe ByteString) (Maybe ByteString) mkRecord topic pid offset ts hdrs key value =- ConsumerRecord- { crTopic = topic- , crPartition = pid- , crOffset = offset- , crTimestamp = ts- , crHeaders = hdrs- , crKey = key- , crValue = value- }+ ConsumerRecord+ { crTopic = topic,+ crPartition = pid,+ crOffset = offset,+ crTimestamp = ts,+ crHeaders = hdrs,+ crKey = key,+ crValue = value+ } envelopeTests :: [TestTree] envelopeTests =- [ testCase "messageId is topic-partition-offset" $ do- let cr = mkRecord (TopicName "orders") (PartitionId 2) (Offset 42) NoTimestamp mempty Nothing (Just "hello")- env = consumerRecordToEnvelope cr- assertEqual "messageId" (MessageId "orders-2-42") env.messageId- , testCase "cursor is CursorInt of offset" $ do- let cr = mkRecord (TopicName "t") (PartitionId 0) (Offset 99) NoTimestamp mempty Nothing Nothing- env = consumerRecordToEnvelope cr- assertEqual "cursor" (Just (CursorInt 99)) env.cursor- , testCase "partition is show of partitionId" $ do- let cr = mkRecord (TopicName "t") (PartitionId 5) (Offset 0) NoTimestamp mempty Nothing Nothing- env = consumerRecordToEnvelope cr- assertEqual "partition" (Just "5") env.partition- , testCase "enqueuedAt from CreateTime" $ do- let cr = mkRecord (TopicName "t") (PartitionId 0) (Offset 0) (CreateTime (Millis 1700000000000)) mempty Nothing Nothing- env = consumerRecordToEnvelope cr- assertEqual "enqueuedAt" (Just (posixSecondsToUTCTime 1700000000)) env.enqueuedAt- , testCase "enqueuedAt Nothing for NoTimestamp" $ do- let cr = mkRecord (TopicName "t") (PartitionId 0) (Offset 0) NoTimestamp mempty Nothing Nothing- env = consumerRecordToEnvelope cr- assertEqual "enqueuedAt" Nothing env.enqueuedAt- , testCase "payload is crValue" $ do- let cr = mkRecord (TopicName "t") (PartitionId 0) (Offset 0) NoTimestamp mempty Nothing (Just "payload-data")- env = consumerRecordToEnvelope cr- assertEqual "payload" (Just "payload-data") env.payload- , testCase "payload Nothing when crValue is Nothing" $ do- let cr = mkRecord (TopicName "t") (PartitionId 0) (Offset 0) NoTimestamp mempty Nothing Nothing- env = consumerRecordToEnvelope cr- assertEqual "payload" Nothing env.payload- , testCase "traceContext extracted from headers" $ do- let hdrs = headersFromList [("traceparent", "00-abc-def-01")]- cr = mkRecord (TopicName "t") (PartitionId 0) (Offset 0) NoTimestamp hdrs Nothing Nothing- env = consumerRecordToEnvelope cr- assertEqual "traceContext" (Just [("traceparent", "00-abc-def-01")]) env.traceContext- , testCase "headers surfaced verbatim (order and duplicates preserved)" $ do- let raw = [("schema-id", "42"), ("x-tag", "a"), ("x-tag", "b")]- cr = mkRecord (TopicName "t") (PartitionId 0) (Offset 0) NoTimestamp (headersFromList raw) Nothing Nothing- env = consumerRecordToEnvelope cr- assertEqual "headers" (Just raw) env.headers- , testCase "empty headers surface as Just []" $ do- let cr = mkRecord (TopicName "t") (PartitionId 0) (Offset 0) NoTimestamp mempty Nothing Nothing- env = consumerRecordToEnvelope cr- assertEqual "headers" (Just []) env.headers- , testCase "attributes carry messaging.system=kafka" $ do- let cr = mkRecord (TopicName "orders") (PartitionId 2) (Offset 42) NoTimestamp mempty Nothing Nothing- env = consumerRecordToEnvelope cr- assertEqual- "messaging.system"- (Just (AttributeValue (TextAttribute "kafka")))- (HashMap.lookup "messaging.system" env.attributes)- , testCase "attributes carry typed messaging.kafka.destination.partition" $ do- let cr = mkRecord (TopicName "orders") (PartitionId 2) (Offset 42) NoTimestamp mempty Nothing Nothing- env = consumerRecordToEnvelope cr- assertEqual- "messaging.kafka.destination.partition"- (Just (AttributeValue (IntAttribute 2)))- (HashMap.lookup (unkey Sem.messaging_kafka_destination_partition) env.attributes)- , testCase "attributes carry typed messaging.kafka.message.offset" $ do- let cr = mkRecord (TopicName "orders") (PartitionId 2) (Offset 42) NoTimestamp mempty Nothing Nothing- env = consumerRecordToEnvelope cr- assertEqual- "messaging.kafka.message.offset"- (Just (AttributeValue (IntAttribute 42)))- (HashMap.lookup (unkey Sem.messaging_kafka_message_offset) env.attributes)- ]+ [ testCase "messageId is topic-partition-offset" $ do+ let cr = mkRecord (TopicName "orders") (PartitionId 2) (Offset 42) NoTimestamp mempty Nothing (Just "hello")+ env = consumerRecordToEnvelope cr+ assertEqual "messageId" (MessageId "orders-2-42") env.messageId,+ testCase "cursor is CursorInt of offset" $ do+ let cr = mkRecord (TopicName "t") (PartitionId 0) (Offset 99) NoTimestamp mempty Nothing Nothing+ env = consumerRecordToEnvelope cr+ assertEqual "cursor" (Just (CursorInt 99)) env.cursor,+ testCase "partition is show of partitionId" $ do+ let cr = mkRecord (TopicName "t") (PartitionId 5) (Offset 0) NoTimestamp mempty Nothing Nothing+ env = consumerRecordToEnvelope cr+ assertEqual "partition" (Just "5") env.partition,+ testCase "enqueuedAt from CreateTime" $ do+ let cr = mkRecord (TopicName "t") (PartitionId 0) (Offset 0) (CreateTime (Millis 1700000000000)) mempty Nothing Nothing+ env = consumerRecordToEnvelope cr+ assertEqual "enqueuedAt" (Just (posixSecondsToUTCTime 1700000000)) env.enqueuedAt,+ testCase "enqueuedAt Nothing for NoTimestamp" $ do+ let cr = mkRecord (TopicName "t") (PartitionId 0) (Offset 0) NoTimestamp mempty Nothing Nothing+ env = consumerRecordToEnvelope cr+ assertEqual "enqueuedAt" Nothing env.enqueuedAt,+ testCase "payload is crValue" $ do+ let cr = mkRecord (TopicName "t") (PartitionId 0) (Offset 0) NoTimestamp mempty Nothing (Just "payload-data")+ env = consumerRecordToEnvelope cr+ assertEqual "payload" (Just "payload-data") env.payload,+ testCase "payload Nothing when crValue is Nothing" $ do+ let cr = mkRecord (TopicName "t") (PartitionId 0) (Offset 0) NoTimestamp mempty Nothing Nothing+ env = consumerRecordToEnvelope cr+ assertEqual "payload" Nothing env.payload,+ testCase "traceContext extracted from headers" $ do+ let hdrs = headersFromList [("traceparent", "00-abc-def-01")]+ cr = mkRecord (TopicName "t") (PartitionId 0) (Offset 0) NoTimestamp hdrs Nothing Nothing+ env = consumerRecordToEnvelope cr+ assertEqual "traceContext" (Just [("traceparent", "00-abc-def-01")]) env.traceContext,+ testCase "headers surfaced verbatim (order and duplicates preserved)" $ do+ let raw = [("schema-id", "42"), ("x-tag", "a"), ("x-tag", "b")]+ cr = mkRecord (TopicName "t") (PartitionId 0) (Offset 0) NoTimestamp (headersFromList raw) Nothing Nothing+ env = consumerRecordToEnvelope cr+ assertEqual "headers" (Just raw) env.headers,+ testCase "empty headers surface as Just []" $ do+ let cr = mkRecord (TopicName "t") (PartitionId 0) (Offset 0) NoTimestamp mempty Nothing Nothing+ env = consumerRecordToEnvelope cr+ assertEqual "headers" (Just []) env.headers,+ testCase "attributes carry messaging.system=kafka" $ do+ let cr = mkRecord (TopicName "orders") (PartitionId 2) (Offset 42) NoTimestamp mempty Nothing Nothing+ env = consumerRecordToEnvelope cr+ assertEqual+ "messaging.system"+ (Just (AttributeValue (TextAttribute "kafka")))+ (HashMap.lookup "messaging.system" env.attributes),+ testCase "attributes carry typed messaging.kafka.destination.partition" $ do+ let cr = mkRecord (TopicName "orders") (PartitionId 2) (Offset 42) NoTimestamp mempty Nothing Nothing+ env = consumerRecordToEnvelope cr+ assertEqual+ "messaging.kafka.destination.partition"+ (Just (AttributeValue (IntAttribute 2)))+ (HashMap.lookup (unkey Sem.messaging_kafka_destination_partition) env.attributes),+ testCase "attributes carry typed messaging.kafka.message.offset" $ do+ let cr = mkRecord (TopicName "orders") (PartitionId 2) (Offset 42) NoTimestamp mempty Nothing Nothing+ env = consumerRecordToEnvelope cr+ assertEqual+ "messaging.kafka.message.offset"+ (Just (AttributeValue (IntAttribute 42)))+ (HashMap.lookup (unkey Sem.messaging_kafka_message_offset) env.attributes)+ ] traceHeaderTests :: [TestTree] traceHeaderTests =- [ testCase "extracts traceparent only" $ do- let hdrs = headersFromList [("traceparent", "00-abc-def-01")]- assertEqual "trace" (Just [("traceparent", "00-abc-def-01")]) (extractTraceHeaders hdrs)- , testCase "extracts traceparent and tracestate" $ do- let hdrs = headersFromList [("traceparent", "00-abc-def-01"), ("tracestate", "vendor=opaque")]- assertEqual- "trace"- (Just [("traceparent", "00-abc-def-01"), ("tracestate", "vendor=opaque")])- (extractTraceHeaders hdrs)- , testCase "returns Nothing when no traceparent" $ do- let hdrs = headersFromList [("other-header", "value")]- assertEqual "trace" Nothing (extractTraceHeaders hdrs)- , testCase "returns Nothing for empty headers" $ do- assertEqual "trace" Nothing (extractTraceHeaders mempty)- , testCase "extracts from materialized header list" $ do- let raw = [("traceparent", "00-list-parent-01"), ("tracestate", "vendor=list")]- assertEqual- "trace"- (Just [("traceparent", "00-list-parent-01"), ("tracestate", "vendor=list")])- (extractTraceHeadersFromList raw)- ]+ [ testCase "extracts traceparent only" $ do+ let hdrs = headersFromList [("traceparent", "00-abc-def-01")]+ assertEqual "trace" (Just [("traceparent", "00-abc-def-01")]) (extractTraceHeaders hdrs),+ testCase "extracts traceparent and tracestate" $ do+ let hdrs = headersFromList [("traceparent", "00-abc-def-01"), ("tracestate", "vendor=opaque")]+ assertEqual+ "trace"+ (Just [("traceparent", "00-abc-def-01"), ("tracestate", "vendor=opaque")])+ (extractTraceHeaders hdrs),+ testCase "returns Nothing when no traceparent" $ do+ let hdrs = headersFromList [("other-header", "value")]+ assertEqual "trace" Nothing (extractTraceHeaders hdrs),+ testCase "returns Nothing for empty headers" $ do+ assertEqual "trace" Nothing (extractTraceHeaders mempty),+ testCase "extracts from materialized header list" $ do+ let raw = [("traceparent", "00-list-parent-01"), ("tracestate", "vendor=list")]+ assertEqual+ "trace"+ (Just [("traceparent", "00-list-parent-01"), ("tracestate", "vendor=list")])+ (extractTraceHeadersFromList raw)+ ] timestampTests :: [TestTree] timestampTests =- [ testCase "CreateTime converts to UTCTime" $ do- let result = timestampToUTCTime (CreateTime (Millis 1700000000000))- assertEqual "time" (Just (posixSecondsToUTCTime 1700000000)) result- , testCase "LogAppendTime converts to UTCTime" $ do- let result = timestampToUTCTime (LogAppendTime (Millis 1700000000000))- assertEqual "time" (Just (posixSecondsToUTCTime 1700000000)) result- , testCase "NoTimestamp returns Nothing" $ do- assertEqual "time" Nothing (timestampToUTCTime NoTimestamp)- , testCase "zero millis converts to epoch" $ do- let result = timestampToUTCTime (CreateTime (Millis 0))- assertEqual "time" (Just (posixSecondsToUTCTime 0)) result- ]+ [ testCase "CreateTime converts to UTCTime" $ do+ let result = timestampToUTCTime (CreateTime (Millis 1700000000000))+ assertEqual "time" (Just (posixSecondsToUTCTime 1700000000)) result,+ testCase "LogAppendTime converts to UTCTime" $ do+ let result = timestampToUTCTime (LogAppendTime (Millis 1700000000000))+ assertEqual "time" (Just (posixSecondsToUTCTime 1700000000)) result,+ testCase "NoTimestamp returns Nothing" $ do+ assertEqual "time" Nothing (timestampToUTCTime NoTimestamp),+ testCase "zero millis converts to epoch" $ do+ let result = timestampToUTCTime (CreateTime (Millis 0))+ assertEqual "time" (Just (posixSecondsToUTCTime 0)) result+ ]
test/Shibuya/Adapter/Kafka/IntegrationTest.hs view
@@ -12,30 +12,30 @@ import Effectful (runEff) import Effectful.Error.Static (runError) import Kafka.Consumer.Types (OffsetCommit (..), OffsetReset (..))-import Kafka.Effectful.Consumer (- brokersList,+import Kafka.Effectful.Consumer+ ( brokersList, groupId, noAutoOffsetStore, offsetReset, runKafkaConsumer, topics,- )+ ) import Kafka.Effectful.Consumer.Effect (commitAllOffsets, pollMessageBatch)-import Kafka.TestEnv (- TestEnv (..),+import Kafka.TestEnv+ ( TestEnv (..), consumeN, createTopic, createTopicWithPartitions, produceKeyedMessages, produceMessages, withTestEnv,- )-import Kafka.Types (- BatchSize (..),+ )+import Kafka.Types+ ( BatchSize (..), KafkaError, Timeout (..), TopicName (..),- )+ ) import Shibuya.Adapter (Adapter (..)) import Shibuya.Adapter.Kafka (KafkaAdapterConfig (..), kafkaAdapter) import Shibuya.App (ProcessorId (..), defaultAppConfig, mkProcessor, runApp, waitApp)@@ -52,292 +52,292 @@ tests :: TestTree tests =- testGroup- "Integration"- [ testCase "Basic produce-consume" testBasicProduceConsume- , testCase "Offset commit verification" testOffsetCommit- , testCase "Multi-partition distribution" testMultiPartition- , testCase "Batch polling" testBatchPolling- , testCase "Graceful shutdown" testGracefulShutdown- , testCase "Idle graceful shutdown completes promptly" testIdleGracefulShutdown- , testCase "AckRetry redelivers within the same session" testAckRetryRedelivery- , testCase "AckRetry is not committed past when session exits" testAckRetryAbandonedSession- , testCase "Handler exception redelivers instead of skipping" testHandlerExceptionRedelivery- ]+ testGroup+ "Integration"+ [ testCase "Basic produce-consume" testBasicProduceConsume,+ testCase "Offset commit verification" testOffsetCommit,+ testCase "Multi-partition distribution" testMultiPartition,+ testCase "Batch polling" testBatchPolling,+ testCase "Graceful shutdown" testGracefulShutdown,+ testCase "Idle graceful shutdown completes promptly" testIdleGracefulShutdown,+ testCase "AckRetry redelivers within the same session" testAckRetryRedelivery,+ testCase "AckRetry is not committed past when session exits" testAckRetryAbandonedSession,+ testCase "Handler exception redelivers instead of skipping" testHandlerExceptionRedelivery+ ] testBasicProduceConsume :: IO () testBasicProduceConsume = withTestEnv $ \env -> do- createTopic env- let payloads = ["msg-1", "msg-2", "msg-3", "msg-4", "msg-5"]- produceMessages env payloads- envelopes <- consumeN env 5 AckOk+ createTopic env+ let payloads = ["msg-1", "msg-2", "msg-3", "msg-4", "msg-5"]+ produceMessages env payloads+ envelopes <- consumeN env 5 AckOk - -- Verify all 5 messages received- assertEqual "message count" 5 (length envelopes)+ -- Verify all 5 messages received+ assertEqual "message count" 5 (length envelopes) - -- Verify payloads- let receivedPayloads = map (\(Envelope{payload}) -> payload) envelopes- assertEqual "payloads" (map Just payloads) receivedPayloads+ -- Verify payloads+ let receivedPayloads = map (\(Envelope {payload}) -> payload) envelopes+ assertEqual "payloads" (map Just payloads) receivedPayloads - -- Verify messageId format: topic-partition-offset- case envelopes of- (Envelope{messageId = MessageId firstIdText} : _) ->- assertBool "messageId contains topic" (Text.isPrefixOf (unTopicName env.testTopic) firstIdText)- [] -> error "unreachable: already verified 5 envelopes"+ -- Verify messageId format: topic-partition-offset+ case envelopes of+ (Envelope {messageId = MessageId firstIdText} : _) ->+ assertBool "messageId contains topic" (Text.isPrefixOf (unTopicName env.testTopic) firstIdText)+ [] -> error "unreachable: already verified 5 envelopes" - -- Verify cursor is populated- assertBool "cursor is Just" (all (\(Envelope{cursor}) -> case cursor of Just (CursorInt _) -> True; _ -> False) envelopes)+ -- Verify cursor is populated+ assertBool "cursor is Just" (all (\(Envelope {cursor}) -> case cursor of Just (CursorInt _) -> True; _ -> False) envelopes) - -- Verify partition is populated- assertBool "partition is Just" (all (\(Envelope{partition}) -> case partition of Just _ -> True; _ -> False) envelopes)+ -- Verify partition is populated+ assertBool "partition is Just" (all (\(Envelope {partition}) -> case partition of Just _ -> True; _ -> False) envelopes) testOffsetCommit :: IO () testOffsetCommit = withTestEnv $ \env -> do- createTopic env- let payloads = ["oc-1", "oc-2", "oc-3"]- produceMessages env payloads+ createTopic env+ let payloads = ["oc-1", "oc-2", "oc-3"]+ produceMessages env payloads - -- Consume all 3, AckOk each (stores offsets), then commit- _ <- consumeN env 3 AckOk+ -- Consume all 3, AckOk each (stores offsets), then commit+ _ <- consumeN env 3 AckOk - -- Create new consumer in same group - should get no messages- result <- runEff . runError @KafkaError $ do- let props = brokersList [env.testBroker] <> groupId env.testGroupId <> noAutoOffsetStore- sub = topics [env.testTopic] <> offsetReset Earliest- runKafkaConsumer props sub $ do- -- Poll a few times to allow group join + rebalance, verify no re-delivery- allResults <- forM [1 .. 3 :: Int] $ \_ -> do- results <- pollMessageBatch (Timeout 3000) (BatchSize 100)- pure [cr | Right cr <- results]- let totalMessages = concat allResults- liftIO $ assertEqual "no re-delivery" 0 (length totalMessages)- case result of- Left err -> error $ "Failed: " <> show err- Right () -> pure ()+ -- Create new consumer in same group - should get no messages+ result <- runEff . runError @KafkaError $ do+ let props = brokersList [env.testBroker] <> groupId env.testGroupId <> noAutoOffsetStore+ sub = topics [env.testTopic] <> offsetReset Earliest+ runKafkaConsumer props sub $ do+ -- Poll a few times to allow group join + rebalance, verify no re-delivery+ allResults <- forM [1 .. 3 :: Int] $ \_ -> do+ results <- pollMessageBatch (Timeout 3000) (BatchSize 100)+ pure [cr | Right cr <- results]+ let totalMessages = concat allResults+ liftIO $ assertEqual "no re-delivery" 0 (length totalMessages)+ case result of+ Left err -> error $ "Failed: " <> show err+ Right () -> pure () testMultiPartition :: IO () testMultiPartition = withTestEnv $ \env -> do- createTopicWithPartitions env 3- let pairs =- [ ("key-a", "msg-a")- , ("key-b", "msg-b")- , ("key-c", "msg-c")- , ("key-d", "msg-d")- , ("key-e", "msg-e")- , ("key-f", "msg-f")- ]- produceKeyedMessages env pairs- envelopes <- consumeN env 6 AckOk+ createTopicWithPartitions env 3+ let pairs =+ [ ("key-a", "msg-a"),+ ("key-b", "msg-b"),+ ("key-c", "msg-c"),+ ("key-d", "msg-d"),+ ("key-e", "msg-e"),+ ("key-f", "msg-f")+ ]+ produceKeyedMessages env pairs+ envelopes <- consumeN env 6 AckOk - assertEqual "message count" 6 (length envelopes)+ assertEqual "message count" 6 (length envelopes) - let partitions = mapMaybe (\(Envelope{partition}) -> partition) envelopes- assertEqual "all have partition" 6 (length partitions)+ let partitions = mapMaybe (\(Envelope {partition}) -> partition) envelopes+ assertEqual "all have partition" 6 (length partitions) - let uniquePartitions = nub partitions- assertBool- ("expected multiple partitions, got: " <> show uniquePartitions)- (length uniquePartitions >= 2)+ let uniquePartitions = nub partitions+ assertBool+ ("expected multiple partitions, got: " <> show uniquePartitions)+ (length uniquePartitions >= 2) testBatchPolling :: IO () testBatchPolling = withTestEnv $ \env -> do- createTopic env- let payloads = map (\i -> BS8.pack ("batch-" <> show i)) [1 .. 20 :: Int]- produceMessages env payloads- envelopes <- consumeN env 20 AckOk+ createTopic env+ let payloads = map (\i -> BS8.pack ("batch-" <> show i)) [1 .. 20 :: Int]+ produceMessages env payloads+ envelopes <- consumeN env 20 AckOk - assertEqual "message count" 20 (length envelopes)+ assertEqual "message count" 20 (length envelopes) - let receivedPayloads = sort $ mapMaybe (\(Envelope{payload}) -> payload) envelopes- assertEqual "payloads" (sort payloads) receivedPayloads+ let receivedPayloads = sort $ mapMaybe (\(Envelope {payload}) -> payload) envelopes+ assertEqual "payloads" (sort payloads) receivedPayloads testGracefulShutdown :: IO () testGracefulShutdown = withTestEnv $ \env -> do- createTopic env- let payloads = ["sd-1", "sd-2", "sd-3"]- produceMessages env payloads+ createTopic env+ let payloads = ["sd-1", "sd-2", "sd-3"]+ produceMessages env payloads - ref <- newIORef ([] :: [Envelope (Maybe ByteString)])- result <- runEff . runError @KafkaError $ do- let props = brokersList [env.testBroker] <> groupId env.testGroupId <> noAutoOffsetStore- sub = topics [env.testTopic] <> offsetReset Earliest- runKafkaConsumer props sub $ do- let config =- KafkaAdapterConfig- { topics = [env.testTopic]- , pollTimeout = Timeout 5000- , batchSize = BatchSize 100- }- Adapter{source, shutdown} <- kafkaAdapter config- Stream.fold Fold.drain- $ Stream.mapM- ( \(Ingested{envelope, ack = AckHandle finalize}) -> do- liftIO $ modifyIORef' ref (envelope :)- finalize AckOk- )- $ Stream.take 3 source- shutdown- case result of- Left err -> error $ "Failed: " <> show err- Right () -> pure ()- envelopes <- reverse <$> readIORef ref- assertEqual "consumed 3 before shutdown" 3 (length envelopes)+ ref <- newIORef ([] :: [Envelope (Maybe ByteString)])+ result <- runEff . runError @KafkaError $ do+ let props = brokersList [env.testBroker] <> groupId env.testGroupId <> noAutoOffsetStore+ sub = topics [env.testTopic] <> offsetReset Earliest+ runKafkaConsumer props sub $ do+ let config =+ KafkaAdapterConfig+ { topics = [env.testTopic],+ pollTimeout = Timeout 5000,+ batchSize = BatchSize 100+ }+ Adapter {source, shutdown} <- kafkaAdapter config+ Stream.fold Fold.drain+ $ Stream.mapM+ ( \(Ingested {envelope, ack = AckHandle finalize}) -> do+ liftIO $ modifyIORef' ref (envelope :)+ finalize AckOk+ )+ $ Stream.take 3 source+ shutdown+ case result of+ Left err -> error $ "Failed: " <> show err+ Right () -> pure ()+ envelopes <- reverse <$> readIORef ref+ assertEqual "consumed 3 before shutdown" 3 (length envelopes) testIdleGracefulShutdown :: IO () testIdleGracefulShutdown = withTestEnv $ \env -> do- createTopic env+ 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- let config =- KafkaAdapterConfig- { topics = [env.testTopic]- , pollTimeout = Timeout 250- , batchSize = BatchSize 100- }- Adapter{source, shutdown} <- kafkaAdapter config- shutdown- Stream.fold Fold.drain source- case timedResult of- Nothing -> assertFailure "idle shutdown did not terminate promptly"- Just (Left (_cs, err)) -> assertFailure $ "idle shutdown failed: " <> show err- Just (Right ()) -> pure ()+ 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+ let config =+ KafkaAdapterConfig+ { topics = [env.testTopic],+ pollTimeout = Timeout 250,+ batchSize = BatchSize 100+ }+ Adapter {source, shutdown} <- kafkaAdapter config+ shutdown+ Stream.fold Fold.drain source+ case timedResult of+ Nothing -> assertFailure "idle shutdown did not terminate promptly"+ Just (Left (_cs, err)) -> assertFailure $ "idle shutdown failed: " <> show err+ Just (Right ()) -> pure () testAckRetryRedelivery :: IO () testAckRetryRedelivery = withTestEnv $ \env -> do- createTopic env- let payloads = ["r-1", "r-2", "r-3"]- produceMessages env payloads+ createTopic env+ let payloads = ["r-1", "r-2", "r-3"]+ produceMessages env payloads - retried <- newIORef False- seen <- newIORef ([] :: [ByteString])- 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)- Stream.fold Fold.drain- $ Stream.mapM- ( \(Ingested{envelope, ack = AckHandle finalize}) -> do- let payload = maybe "" id envelope.payload- liftIO $ modifyIORef' seen (payload :)- hasRetried <- liftIO $ readIORef retried- if payload == "r-2" && not hasRetried- then do- liftIO $ writeIORef retried True- finalize (AckRetry (RetryDelay 0))- else finalize AckOk- )- $ Stream.take 4 source- commitAllOffsets OffsetCommit- case result of- Left (_cs, err) -> assertFailure $ "AckRetry redelivery failed: " <> show err- Right () -> pure ()+ retried <- newIORef False+ seen <- newIORef ([] :: [ByteString])+ 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)+ Stream.fold Fold.drain+ $ Stream.mapM+ ( \(Ingested {envelope, ack = AckHandle finalize}) -> do+ let payload = maybe "" id envelope.payload+ liftIO $ modifyIORef' seen (payload :)+ hasRetried <- liftIO $ readIORef retried+ if payload == "r-2" && not hasRetried+ then do+ liftIO $ writeIORef retried True+ finalize (AckRetry (RetryDelay 0))+ else finalize AckOk+ )+ $ Stream.take 4 source+ commitAllOffsets OffsetCommit+ case result of+ Left (_cs, err) -> assertFailure $ "AckRetry redelivery failed: " <> show err+ Right () -> pure () - delivered <- reverse <$> readIORef seen- assertBool ("expected r-2 at least twice, saw " <> show delivered) (countPayload "r-2" delivered >= 2)- assertBool ("expected r-3 after retry, saw " <> show delivered) ("r-3" `elem` delivered)+ delivered <- reverse <$> readIORef seen+ assertBool ("expected r-2 at least twice, saw " <> show delivered) (countPayload "r-2" delivered >= 2)+ assertBool ("expected r-3 after retry, saw " <> show delivered) ("r-3" `elem` delivered) - noRedelivery <- 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 .. 3 :: Int] $ \_ ->- pollMessageBatch (Timeout 500) (BatchSize 100)- liftIO $ assertEqual "no redelivery after final AckOk commit" 0 (length [cr | Right cr <- concat batches])- case noRedelivery of- Left (_cs, err) -> assertFailure $ "post-commit verification failed: " <> show err- Right () -> pure ()+ noRedelivery <- 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 .. 3 :: Int] $ \_ ->+ pollMessageBatch (Timeout 500) (BatchSize 100)+ liftIO $ assertEqual "no redelivery after final AckOk commit" 0 (length [cr | Right cr <- concat batches])+ case noRedelivery of+ Left (_cs, err) -> assertFailure $ "post-commit verification failed: " <> show err+ Right () -> pure () testAckRetryAbandonedSession :: IO () testAckRetryAbandonedSession = withTestEnv $ \env -> do- createTopic env- let payloads = ["ab-1", "ab-2", "ab-3"]- produceMessages env payloads+ createTopic env+ let payloads = ["ab-1", "ab-2", "ab-3"]+ produceMessages env payloads - firstSession <- 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)- Stream.fold Fold.drain- $ Stream.mapM- ( \(Ingested{envelope, ack = AckHandle finalize}) -> do- case envelope.payload of- Just "ab-2" -> finalize (AckRetry (RetryDelay 0))- _ -> finalize AckOk- )- $ Stream.take 2 source- case firstSession of- Left (_cs, err) -> assertFailure $ "first session failed: " <> show err- Right () -> pure ()+ firstSession <- 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)+ Stream.fold Fold.drain+ $ Stream.mapM+ ( \(Ingested {envelope, ack = AckHandle finalize}) -> do+ case envelope.payload of+ Just "ab-2" -> finalize (AckRetry (RetryDelay 0))+ _ -> finalize AckOk+ )+ $ Stream.take 2 source+ case firstSession of+ Left (_cs, err) -> assertFailure $ "first session failed: " <> show err+ Right () -> pure () - redelivered <- newIORef ([] :: [ByteString])- secondSession <- 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)- Stream.fold Fold.drain- $ Stream.mapM- ( \(Ingested{envelope, ack = AckHandle finalize}) -> do- maybe (pure ()) (liftIO . modifyIORef' redelivered . (:)) envelope.payload- finalize AckOk- )- $ Stream.take 2 source- commitAllOffsets OffsetCommit- case secondSession of- Left (_cs, err) -> assertFailure $ "second session failed: " <> show err- Right () -> pure ()+ redelivered <- newIORef ([] :: [ByteString])+ secondSession <- 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)+ Stream.fold Fold.drain+ $ Stream.mapM+ ( \(Ingested {envelope, ack = AckHandle finalize}) -> do+ maybe (pure ()) (liftIO . modifyIORef' redelivered . (:)) envelope.payload+ finalize AckOk+ )+ $ Stream.take 2 source+ commitAllOffsets OffsetCommit+ case secondSession of+ Left (_cs, err) -> assertFailure $ "second session failed: " <> show err+ Right () -> pure () - delivered <- readIORef redelivered- assertBool ("expected ab-2 redelivery, saw " <> show delivered) ("ab-2" `elem` delivered)+ delivered <- readIORef redelivered+ assertBool ("expected ab-2 redelivery, saw " <> show delivered) ("ab-2" `elem` delivered) testHandlerExceptionRedelivery :: IO () testHandlerExceptionRedelivery = withTestEnv $ \env -> do- createTopic env- let payloads = ["n-1", "n-2", "n-3"]- produceMessages env payloads+ createTopic env+ let payloads = ["n-1", "n-2", "n-3"]+ produceMessages env payloads - thrown <- newIORef False- seen <- newIORef ([] :: [ByteString])- result <- runEff . runError @KafkaError . runTracingNoop $ do- let props = brokersList [env.testBroker] <> groupId env.testGroupId <> noAutoOffsetStore- sub = topics [env.testTopic] <> offsetReset Earliest- runKafkaConsumer props sub $ do- upstream <- kafkaAdapter (testConfig env)- let finiteAdapter = upstream{source = Stream.take 4 upstream.source}- handler Message{envelope} = do- let payload = maybe "" id envelope.payload- liftIO $ modifyIORef' seen (payload :)- hasThrown <- liftIO $ readIORef thrown- if payload == "n-2" && not hasThrown- then liftIO $ do- writeIORef thrown True- throwIO (userError "planned handler exception")- else pure AckOk- appResult <- runApp defaultAppConfig [(ProcessorId "handler-exception", mkProcessor finiteAdapter handler)]- case appResult of- Left appErr -> liftIO $ assertFailure $ "runApp failed: " <> show appErr- Right appHandle -> waitApp appHandle- commitAllOffsets OffsetCommit- case result of- Left (_cs, err) -> assertFailure $ "handler exception scenario failed: " <> show err- Right () -> pure ()+ thrown <- newIORef False+ seen <- newIORef ([] :: [ByteString])+ result <- runEff . runError @KafkaError . runTracingNoop $ do+ let props = brokersList [env.testBroker] <> groupId env.testGroupId <> noAutoOffsetStore+ sub = topics [env.testTopic] <> offsetReset Earliest+ runKafkaConsumer props sub $ do+ upstream <- kafkaAdapter (testConfig env)+ let finiteAdapter = upstream {source = Stream.take 4 upstream.source}+ handler Message {envelope} = do+ let payload = maybe "" id envelope.payload+ liftIO $ modifyIORef' seen (payload :)+ hasThrown <- liftIO $ readIORef thrown+ if payload == "n-2" && not hasThrown+ then liftIO $ do+ writeIORef thrown True+ throwIO (userError "planned handler exception")+ else pure AckOk+ appResult <- runApp defaultAppConfig [(ProcessorId "handler-exception", mkProcessor finiteAdapter handler)]+ case appResult of+ Left appErr -> liftIO $ assertFailure $ "runApp failed: " <> show appErr+ Right appHandle -> waitApp appHandle+ commitAllOffsets OffsetCommit+ case result of+ Left (_cs, err) -> assertFailure $ "handler exception scenario failed: " <> show err+ Right () -> pure () - delivered <- reverse <$> readIORef seen- assertBool ("expected n-2 at least twice, saw " <> show delivered) (countPayload "n-2" delivered >= 2)- assertBool ("expected n-3 after handler exception retry, saw " <> show delivered) ("n-3" `elem` delivered)+ delivered <- reverse <$> readIORef seen+ assertBool ("expected n-2 at least twice, saw " <> show delivered) (countPayload "n-2" delivered >= 2)+ assertBool ("expected n-3 after handler exception retry, saw " <> show delivered) ("n-3" `elem` delivered) testConfig :: TestEnv -> KafkaAdapterConfig testConfig env =- KafkaAdapterConfig- { topics = [env.testTopic]- , pollTimeout = Timeout 500- , batchSize = BatchSize 100- }+ KafkaAdapterConfig+ { topics = [env.testTopic],+ pollTimeout = Timeout 500,+ batchSize = BatchSize 100+ } countPayload :: ByteString -> [ByteString] -> Int countPayload target = length . filter (== target)