diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/shibuya-kafka-adapter.cabal b/shibuya-kafka-adapter.cabal --- a/shibuya-kafka-adapter.cabal +++ b/shibuya-kafka-adapter.cabal @@ -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, diff --git a/src/Shibuya/Adapter/Kafka.hs b/src/Shibuya/Adapter/Kafka.hs --- a/src/Shibuya/Adapter/Kafka.hs +++ b/src/Shibuya/Adapter/Kafka.hs @@ -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, ()) diff --git a/src/Shibuya/Adapter/Kafka/Config.hs b/src/Shibuya/Adapter/Kafka/Config.hs --- a/src/Shibuya/Adapter/Kafka/Config.hs +++ b/src/Shibuya/Adapter/Kafka/Config.hs @@ -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 + } diff --git a/src/Shibuya/Adapter/Kafka/Convert.hs b/src/Shibuya/Adapter/Kafka/Convert.hs --- a/src/Shibuya/Adapter/Kafka/Convert.hs +++ b/src/Shibuya/Adapter/Kafka/Convert.hs @@ -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 diff --git a/src/Shibuya/Adapter/Kafka/Internal.hs b/src/Shibuya/Adapter/Kafka/Internal.hs --- a/src/Shibuya/Adapter/Kafka/Internal.hs +++ b/src/Shibuya/Adapter/Kafka/Internal.hs @@ -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 diff --git a/test/Kafka/TestEnv.hs b/test/Kafka/TestEnv.hs --- a/test/Kafka/TestEnv.hs +++ b/test/Kafka/TestEnv.hs @@ -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 diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -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 + ] diff --git a/test/Shibuya/Adapter/Kafka/AckHandleTest.hs b/test/Shibuya/Adapter/Kafka/AckHandleTest.hs --- a/test/Shibuya/Adapter/Kafka/AckHandleTest.hs +++ b/test/Shibuya/Adapter/Kafka/AckHandleTest.hs @@ -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 () diff --git a/test/Shibuya/Adapter/Kafka/AdapterTest.hs b/test/Shibuya/Adapter/Kafka/AdapterTest.hs --- a/test/Shibuya/Adapter/Kafka/AdapterTest.hs +++ b/test/Shibuya/Adapter/Kafka/AdapterTest.hs @@ -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" diff --git a/test/Shibuya/Adapter/Kafka/ConvertTest.hs b/test/Shibuya/Adapter/Kafka/ConvertTest.hs --- a/test/Shibuya/Adapter/Kafka/ConvertTest.hs +++ b/test/Shibuya/Adapter/Kafka/ConvertTest.hs @@ -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 + ] diff --git a/test/Shibuya/Adapter/Kafka/IntegrationTest.hs b/test/Shibuya/Adapter/Kafka/IntegrationTest.hs --- a/test/Shibuya/Adapter/Kafka/IntegrationTest.hs +++ b/test/Shibuya/Adapter/Kafka/IntegrationTest.hs @@ -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)