diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,35 @@ +# Changelog + +## 0.1.0.0 — 2026-04-18 + +Initial release. + +`shibuya-kafka-adapter` bridges Apache Kafka to the +[Shibuya](https://github.com/shinzui/shibuya) queue-processing framework. It +builds on [`kafka-effectful`](https://github.com/shinzui/kafka-effectful) for +the consumer effect (polling, offset store, partition pause) and +[`hw-kafka-streamly`](https://hackage.haskell.org/package/hw-kafka-streamly) +for error classification (`skipNonFatal`), on top of +[`hw-kafka-client`](https://github.com/haskell-works/hw-kafka-client). + +### Features + +- Poll-driven consumer that produces Shibuya `Envelope` values from Kafka + `ConsumerRecord`s. +- Offset-commit semantics combining `noAutoOffsetStore`, explicit + `storeOffsetMessage` on successful acknowledgement, and librdkafka + auto-commit of the stored offsets. +- Partition-aware dispatch: `Envelope`s carry topic/partition/offset, and + `AckHalt` pauses the originating partition via the consumer effect. +- W3C `traceparent` / `tracestate` header extraction from Kafka message + headers, surfaced on the Shibuya `Envelope` for OpenTelemetry propagation. +- Kafka message timestamp conversion to Shibuya's `UTCTime` representation. +- Graceful shutdown that calls `commitAllOffsets` so stored offsets are + flushed before the consumer handle closes. + +### Known Limitations + +- No automatic partition resume after `AckHalt` within the consumer session; + resumption is left to the operator or the next rebalance. +- No dead-letter queue production. `AckDead` stores the offset so the stream + advances past the poison message but does not publish it anywhere. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Nadeem Bitar + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,42 @@ +# shibuya-kafka-adapter + +Kafka adapter for the [Shibuya](https://github.com/shinzui/shibuya) queue-processing framework. + +Integrates with Apache Kafka via [`kafka-effectful`](https://github.com/shinzui/kafka-effectful) for the consumer effect (polling, offset store, partition pause) and [`hw-kafka-streamly`](https://hackage.haskell.org/package/hw-kafka-streamly) for error classification (`skipNonFatal`), on top of [`hw-kafka-client`](https://github.com/haskell-works/hw-kafka-client). Provides polling, offset commit semantics, partition awareness, and graceful shutdown. + +## Packages + +- `shibuya-kafka-adapter` — the adapter library (`Shibuya.Adapter.Kafka`, `.Config`, `.Convert`). +- `shibuya-kafka-adapter-bench` — micro-benchmarks for the conversion hot path (`ConsumerRecord` → `Envelope`, W3C header extraction, timestamps). +- `shibuya-kafka-adapter-jitsurei` — runnable examples: `BasicConsumer`, `MultiTopic`, `MultiPartition`, `OffsetManagement`. + +## Building + +The repo ships a Nix flake and `direnv` config for a reproducible toolchain. + +```sh +direnv allow # or: nix develop +cabal build all +cabal test shibuya-kafka-adapter +``` + +Benchmarks and examples: + +```sh +cabal bench shibuya-kafka-adapter-bench +cabal run BasicConsumer +``` + +## Layout + +``` +shibuya-kafka-adapter/ library sources and tests +shibuya-kafka-adapter-bench/ tasty-bench micro-benchmarks +shibuya-kafka-adapter-jitsurei/ runnable usage examples +docs/plans/ execution plans +mori.dhall project manifest (mori registry) +``` + +## License + +MIT. See package cabal files for details. diff --git a/shibuya-kafka-adapter.cabal b/shibuya-kafka-adapter.cabal new file mode 100644 --- /dev/null +++ b/shibuya-kafka-adapter.cabal @@ -0,0 +1,113 @@ +cabal-version: 3.12 +name: shibuya-kafka-adapter +version: 0.1.0.0 +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 +extra-doc-files: + CHANGELOG.md + README.md + +source-repository head + 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 + -Wmissing-deriving-strategies + +library + import: warnings + exposed-modules: + Shibuya.Adapter.Kafka + Shibuya.Adapter.Kafka.Config + Shibuya.Adapter.Kafka.Convert + Shibuya.Adapter.Kafka.Internal + + default-extensions: + DeriveAnyClass + DerivingStrategies + DuplicateRecordFields + LambdaCase + NoFieldSelectors + OverloadedLabels + OverloadedRecordDot + OverloadedStrings + QuasiQuotes + + build-depends: + , base ^>=4.21.0.0 + , bytestring ^>=0.12 + , containers ^>=0.7 + , effectful-core ^>=2.6.1.0 + , hw-kafka-client >=5.3 && <6 + , hw-kafka-streamly ^>=0.1 + , kafka-effectful ^>=0.1 + , shibuya-core ^>=0.1.0.0 + , stm ^>=2.5 + , streamly ^>=0.11 + , streamly-core ^>=0.3 + , text ^>=2.1 + , time ^>=1.14 + + 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 + default-extensions: + DeriveAnyClass + DerivingStrategies + DuplicateRecordFields + LambdaCase + NoFieldSelectors + OverloadedLabels + OverloadedRecordDot + OverloadedStrings + QuasiQuotes + + other-modules: + Kafka.TestEnv + Shibuya.Adapter.Kafka.AdapterTest + Shibuya.Adapter.Kafka.ConvertTest + Shibuya.Adapter.Kafka.IntegrationTest + + build-depends: + , base ^>=4.21.0.0 + , bytestring + , containers + , effectful-core + , hw-kafka-client + , kafka-effectful + , process + , random + , shibuya-core + , shibuya-kafka-adapter + , stm + , streamly + , streamly-core + , tasty ^>=1.5 + , tasty-hunit ^>=0.10 + , text + , time diff --git a/src/Shibuya/Adapter/Kafka.hs b/src/Shibuya/Adapter/Kafka.hs new file mode 100644 --- /dev/null +++ b/src/Shibuya/Adapter/Kafka.hs @@ -0,0 +1,135 @@ +{- | 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 (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 IgnoreFailures 100 + [ (ProcessorId \"orders\", mkProcessor adapter myHandler) + ] + ... +@ + +== Message Lifecycle + +1. Messages are polled from Kafka in batches +2. Each message is wrapped as an 'Ingested' with an 'AckHandle' +3. On 'AckOk', the offset is stored (auto-commit flushes to broker) +4. On 'AckRetry', the offset is stored (Kafka cannot un-read messages) +5. On 'AckDeadLetter', the offset is stored (DLQ production in future milestone) +6. On 'AckHalt', the partition is paused and offset is NOT stored + +== 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.Source.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@. The partition is not automatically resumed within the +current consumer session — the adapter has no side channel for a handler to +request resumption. A new consumer session (a new call to @runKafkaConsumer@) +starts with no paused partitions, so resumption happens implicitly on restart +but not mid-session. If mid-session resume is required, the caller must +manage that explicitly outside the 'Adapter' surface. +-} +module Shibuya.Adapter.Kafka ( + -- * Adapter + kafkaAdapter, + + -- * Configuration + KafkaAdapterConfig (..), + + -- * Defaults + defaultConfig, + + -- * Re-exports + TopicName (..), + BrokerAddress (..), + ConsumerGroupId (..), + OffsetReset (..), + OffsetCommit (..), + Timeout (..), + BatchSize (..), + KafkaError, +) +where + +import Control.Concurrent.STM (TVar, atomically, newTVarIO, readTVarIO, writeTVar) +import Control.Monad.IO.Class (liftIO) +import Data.ByteString (ByteString) +import Data.Text qualified as Text +import Effectful (Eff, IOE, (:>)) +import Effectful.Error.Static (Error) +import Kafka.Consumer.Types (ConsumerGroupId (..), OffsetCommit (..), OffsetReset (..)) +import Kafka.Effectful.Consumer.Effect (KafkaConsumer, commitAllOffsets) +import Kafka.Types (BatchSize (..), BrokerAddress (..), KafkaError, Timeout (..), TopicName (..)) +import Shibuya.Adapter (Adapter (..)) +import Shibuya.Adapter.Kafka.Config (KafkaAdapterConfig (..), defaultConfig) +import Shibuya.Adapter.Kafka.Internal (ingestedStream, kafkaSource, mkIngested) +import Streamly.Data.Stream (Stream) +import Streamly.Data.Stream qualified as Stream + +{- | 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 +any stored offsets to the broker. + +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)) +kafkaAdapter config = do + shutdownVar <- liftIO $ newTVarIO False + let messageSource = ingestedStream mkIngested (kafkaSource config) + pure + Adapter + { adapterName = "kafka:" <> Text.intercalate "," (map unTopicName config.topics) + , source = takeUntilShutdown shutdownVar messageSource + , shutdown = do + liftIO $ atomically $ writeTVar shutdownVar True + commitAllOffsets OffsetCommit + } + +-- | Take from stream until shutdown signal is set. +takeUntilShutdown :: + (IOE :> es) => + TVar Bool -> + Stream (Eff es) a -> + Stream (Eff es) a +takeUntilShutdown shutdownVar = + Stream.takeWhileM $ \_ -> do + isShutdown <- liftIO $ readTVarIO shutdownVar + pure (not isShutdown) diff --git a/src/Shibuya/Adapter/Kafka/Config.hs b/src/Shibuya/Adapter/Kafka/Config.hs new file mode 100644 --- /dev/null +++ b/src/Shibuya/Adapter/Kafka/Config.hs @@ -0,0 +1,48 @@ +-- | Configuration types for the Kafka adapter. +module Shibuya.Adapter.Kafka.Config ( + -- * Main Configuration + KafkaAdapterConfig (..), + + -- * Defaults + defaultConfig, +) +where + +import GHC.Generics (Generic) +import Kafka.Consumer.Types (OffsetReset (..)) +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. +-} +data KafkaAdapterConfig = KafkaAdapterConfig + { topics :: ![TopicName] + -- ^ Topics to consume from + , pollTimeout :: !Timeout + -- ^ Timeout for each poll call (default: 1000ms) + , batchSize :: !BatchSize + -- ^ Maximum messages per poll batch (default: 100) + , offsetReset :: !OffsetReset + -- ^ Where to start when no committed offset exists (default: Earliest) + } + deriving stock (Show, Eq, Generic) + +{- | Default adapter configuration for the given topics. + +Defaults: + +* @pollTimeout@: 1000ms +* @batchSize@: 100 +* @offsetReset@: Earliest +-} +defaultConfig :: [TopicName] -> KafkaAdapterConfig +defaultConfig ts = + KafkaAdapterConfig + { topics = ts + , pollTimeout = Timeout 1000 + , batchSize = BatchSize 100 + , offsetReset = Earliest + } diff --git a/src/Shibuya/Adapter/Kafka/Convert.hs b/src/Shibuya/Adapter/Kafka/Convert.hs new file mode 100644 --- /dev/null +++ b/src/Shibuya/Adapter/Kafka/Convert.hs @@ -0,0 +1,86 @@ +-- | Type conversions between Kafka and Shibuya types. +module Shibuya.Adapter.Kafka.Convert ( + -- * Message Conversion + consumerRecordToEnvelope, + + -- * Trace Context + extractTraceHeaders, + + -- * Timestamp Conversion + timestampToUTCTime, +) +where + +import Data.ByteString (ByteString) +import Data.Int (Int64) +import Data.Text qualified as Text +import Data.Time (UTCTime) +import Data.Time.Clock.POSIX (posixSecondsToUTCTime) +import Kafka.Consumer.Types (ConsumerRecord (..), Offset (..), Timestamp (..)) +import Kafka.Types ( + Headers, + Millis (..), + PartitionId (..), + TopicName (..), + headersToList, + ) +import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..), TraceHeaders) + +{- | 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 +* @payload@: the @crValue@ field (@Maybe ByteString@) +-} +consumerRecordToEnvelope :: + ConsumerRecord (Maybe ByteString) (Maybe ByteString) -> + Envelope (Maybe ByteString) +consumerRecordToEnvelope cr = + Envelope + { messageId = mkMessageId cr.crTopic cr.crPartition cr.crOffset + , cursor = Just (CursorInt (fromIntegral (unOffset cr.crOffset))) + , partition = Just (Text.pack (show (unPartitionId cr.crPartition))) + , enqueuedAt = timestampToUTCTime cr.crTimestamp + , traceContext = extractTraceHeaders cr.crHeaders + , payload = cr.crValue + } + +-- | 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. + +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 headers = + 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)] + where + headerList :: [(ByteString, ByteString)] + headerList = headersToList headers + +{- | 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 + +-- | Convert milliseconds since Unix epoch to UTCTime. +millisToUTCTime :: Int64 -> UTCTime +millisToUTCTime ms = posixSecondsToUTCTime (fromIntegral ms / 1000) diff --git a/src/Shibuya/Adapter/Kafka/Internal.hs b/src/Shibuya/Adapter/Kafka/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Shibuya/Adapter/Kafka/Internal.hs @@ -0,0 +1,115 @@ +{- | 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 ( + -- * Stream Construction + kafkaSource, + ingestedStream, + + -- * Ingested Construction + mkIngested, + + -- * AckHandle Construction + mkAckHandle, +) +where + +import Data.ByteString (ByteString) +import Data.Function ((&)) +import Effectful (Eff, (:>)) +import Effectful.Error.Static (Error, throwError) +import Kafka.Consumer.Types (ConsumerRecord (..)) +import Kafka.Effectful.Consumer.Effect ( + KafkaConsumer, + pausePartitions, + pollMessageBatch, + storeOffsetMessage, + ) +import Kafka.Streamly.Source (skipNonFatal) +import Kafka.Types (KafkaError) +import Shibuya.Adapter.Kafka.Config (KafkaAdapterConfig (..)) +import Shibuya.Adapter.Kafka.Convert (consumerRecordToEnvelope) +import Shibuya.Core.Ack (AckDecision (..)) +import Shibuya.Core.AckHandle (AckHandle (..)) +import Shibuya.Core.Ingested (Ingested (..)) +import Streamly.Data.Stream (Stream) +import Streamly.Data.Stream qualified as Stream + +{- | Create a stream of 'ConsumerRecord's by repeatedly polling the broker. + +Calls 'pollMessageBatch' in a loop, preserving errors as @Left@ values. +Non-fatal errors (timeouts, partition EOF, etc.) are filtered out via +'skipNonFatal' from hw-kafka-streamly. Fatal errors are preserved for +upstream handling. +-} +kafkaSource :: + (KafkaConsumer :> es) => + KafkaAdapterConfig -> + Stream (Eff es) (Either KafkaError (ConsumerRecord (Maybe ByteString) (Maybe ByteString))) +kafkaSource config = + skipNonFatal $ + Stream.repeatM pollBatch + & Stream.concatMap Stream.fromList + where + pollBatch = + pollMessageBatch config.pollTimeout config.batchSize + +{- | Create an 'AckHandle' for a single 'ConsumerRecord'. + +Maps 'AckDecision' to Kafka operations: + +* 'AckOk' -> 'storeOffsetMessage' (mark offset ready for commit) +* 'AckRetry' -> 'storeOffsetMessage' (Kafka cannot un-read; see Decision Log) +* 'AckDeadLetter' -> 'storeOffsetMessage' (DLQ deferred to future milestone) +* 'AckHalt' -> 'pausePartitions' (do NOT store offset; message will be re-consumed) +-} +mkAckHandle :: + (KafkaConsumer :> es) => + ConsumerRecord (Maybe ByteString) (Maybe ByteString) -> + AckHandle es +mkAckHandle cr = AckHandle $ \case + AckOk -> + storeOffsetMessage cr + AckRetry _ -> + storeOffsetMessage cr + AckDeadLetter _ -> + storeOffsetMessage cr + AckHalt _ -> + pausePartitions [(cr.crTopic, cr.crPartition)] + +{- | Combine conversion and ack handle to produce an 'Ingested'. + +Lease is always 'Nothing' for Kafka (no visibility timeout mechanism). +-} +mkIngested :: + (KafkaConsumer :> es) => + ConsumerRecord (Maybe ByteString) (Maybe ByteString) -> + Ingested es (Maybe ByteString) +mkIngested cr = + Ingested + { envelope = consumerRecordToEnvelope cr + , ack = mkAckHandle cr + , lease = Nothing + } + +{- | 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.Source.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)) +ingestedStream mkI = + 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 new file mode 100644 --- /dev/null +++ b/test/Kafka/TestEnv.hs @@ -0,0 +1,191 @@ +-- | Test environment helpers for Kafka integration tests. +module Kafka.TestEnv ( + -- * Test Environment + TestEnv (..), + withTestEnv, + + -- * Producing + produceMessages, + produceKeyedMessages, + + -- * Consuming via Adapter + consumeN, + + -- * Topic Management + createTopic, + createTopicWithPartitions, +) +where + +import Control.Monad (forM_) +import Control.Monad.IO.Class (liftIO) +import Data.ByteString (ByteString) +import Data.IORef (modifyIORef', newIORef, readIORef) +import Data.Text qualified as Text +import Effectful (runEff) +import Effectful.Error.Static (runError) +import Kafka.Consumer.ConsumerProperties (ConsumerProperties) +import Kafka.Consumer.Subscription (Subscription) +import Kafka.Consumer.Types (ConsumerGroupId (..), OffsetCommit (..), OffsetReset (..)) +import Kafka.Effectful.Consumer ( + brokersList, + groupId, + noAutoOffsetStore, + offsetReset, + runKafkaConsumer, + topics, + ) +import Kafka.Effectful.Consumer.Effect (commitAllOffsets) +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 (..), + BrokerAddress (..), + KafkaError, + Timeout (..), + TopicName (..), + ) +import Shibuya.Adapter (Adapter (..)) +import Shibuya.Adapter.Kafka (KafkaAdapterConfig (..), kafkaAdapter) +import Shibuya.Core.Ack (AckDecision (..)) +import Shibuya.Core.AckHandle (AckHandle (..)) +import Shibuya.Core.Ingested (Ingested (..)) +import Shibuya.Core.Types (Envelope (..)) +import Streamly.Data.Fold qualified as Fold +import Streamly.Data.Stream qualified as Stream +import System.Process (callCommand) +import System.Random (randomRIO) + +-- | Test environment with isolated topic and group. +data TestEnv = TestEnv + { 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 "localhost: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] + +-- | Create a topic with 1 partition via rpk. +createTopic :: TestEnv -> IO () +createTopic env = createTopicWithPartitions env 1 + +-- | 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" + +-- | Build consumer properties for testing. +mkConsumerProps :: TestEnv -> ConsumerProperties +mkConsumerProps env = + brokersList [env.testBroker] + <> groupId env.testGroupId + <> noAutoOffsetStore + +-- | Build producer properties for testing. +mkProducerProps :: TestEnv -> ProducerProperties +mkProducerProps env = + P.brokersList [env.testBroker] + +-- | Build subscription for testing. +mkSubscription :: TestEnv -> Subscription +mkSubscription env = + 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 () + +-- | 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 () + +-- | Consume N messages from the test topic via the adapter, applying the given ack decision. +consumeN :: + 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 + , offsetReset = Earliest + } + 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 new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -0,0 +1,16 @@ +module Main (main) where + +import Shibuya.Adapter.Kafka.AdapterTest qualified as AdapterTest +import Shibuya.Adapter.Kafka.ConvertTest qualified as ConvertTest +import Shibuya.Adapter.Kafka.IntegrationTest qualified as IntegrationTest +import Test.Tasty (defaultMain, testGroup) + +main :: IO () +main = + defaultMain $ + testGroup + "shibuya-kafka-adapter" + [ AdapterTest.tests + , ConvertTest.tests + , IntegrationTest.tests + ] diff --git a/test/Shibuya/Adapter/Kafka/AdapterTest.hs b/test/Shibuya/Adapter/Kafka/AdapterTest.hs new file mode 100644 --- /dev/null +++ b/test/Shibuya/Adapter/Kafka/AdapterTest.hs @@ -0,0 +1,60 @@ +{- | 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) +import Effectful (runEff) +import Effectful.Error.Static (runError) +import Kafka.Consumer.Types (ConsumerRecord) +import Kafka.Types (KafkaError (..)) +import Shibuya.Adapter.Kafka.Internal (ingestedStream) +import Shibuya.Core.Ingested (Ingested) +import Streamly.Data.Fold qualified as Fold +import Streamly.Data.Stream qualified as Stream +import Test.Tasty (TestTree, testGroup) +import Test.Tasty.HUnit (assertEqual, assertFailure, testCase) + +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 + ] + +{- | 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) +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" + +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" diff --git a/test/Shibuya/Adapter/Kafka/ConvertTest.hs b/test/Shibuya/Adapter/Kafka/ConvertTest.hs new file mode 100644 --- /dev/null +++ b/test/Shibuya/Adapter/Kafka/ConvertTest.hs @@ -0,0 +1,120 @@ +module Shibuya.Adapter.Kafka.ConvertTest (tests) where + +import Data.ByteString (ByteString) +import Data.Time.Clock.POSIX (posixSecondsToUTCTime) +import Kafka.Consumer.Types (ConsumerRecord (..), Offset (..), Timestamp (..)) +import Kafka.Types ( + Headers, + Millis (..), + PartitionId (..), + TopicName (..), + headersFromList, + ) +import Shibuya.Adapter.Kafka.Convert ( + consumerRecordToEnvelope, + extractTraceHeaders, + 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 + ] + +-- | A minimal ConsumerRecord for testing. +mkRecord :: + 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 + } + +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 + ] + +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) + ] + +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 + ] diff --git a/test/Shibuya/Adapter/Kafka/IntegrationTest.hs b/test/Shibuya/Adapter/Kafka/IntegrationTest.hs new file mode 100644 --- /dev/null +++ b/test/Shibuya/Adapter/Kafka/IntegrationTest.hs @@ -0,0 +1,177 @@ +module Shibuya.Adapter.Kafka.IntegrationTest (tests) where + +import Control.Monad (forM) +import Control.Monad.IO.Class (liftIO) +import Data.ByteString (ByteString) +import Data.ByteString.Char8 qualified as BS8 +import Data.IORef (modifyIORef', newIORef, readIORef) +import Data.List (nub, sort) +import Data.Maybe (mapMaybe) +import Data.Text qualified as Text +import Effectful (runEff) +import Effectful.Error.Static (runError) +import Kafka.Consumer.Types (OffsetReset (..)) +import Kafka.Effectful.Consumer ( + brokersList, + groupId, + noAutoOffsetStore, + offsetReset, + runKafkaConsumer, + topics, + ) +import Kafka.Effectful.Consumer.Effect (pollMessageBatch) +import Kafka.TestEnv ( + TestEnv (..), + consumeN, + createTopic, + createTopicWithPartitions, + produceKeyedMessages, + produceMessages, + withTestEnv, + ) +import Kafka.Types ( + BatchSize (..), + KafkaError, + Timeout (..), + TopicName (..), + ) +import Shibuya.Adapter (Adapter (..)) +import Shibuya.Adapter.Kafka (KafkaAdapterConfig (..), kafkaAdapter) +import Shibuya.Core.Ack (AckDecision (..)) +import Shibuya.Core.AckHandle (AckHandle (..)) +import Shibuya.Core.Ingested (Ingested (..)) +import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..)) +import Streamly.Data.Fold qualified as Fold +import Streamly.Data.Stream qualified as Stream +import Test.Tasty (TestTree, testGroup) +import Test.Tasty.HUnit (assertBool, assertEqual, testCase) + +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 + ] + +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 + + -- 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 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 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 + + -- 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 () + +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 + + assertEqual "message count" 6 (length envelopes) + + 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) + +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 + + assertEqual "message count" 20 (length envelopes) + + 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 + + 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 + , offsetReset = Earliest + } + 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)