diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,29 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to the [Haskell Package Versioning Policy](https://pvp.haskell.org/). -## 0.1.0.0 (unreleased) +## 0.2.0.0 — 2026-05-07 + +### Changed (BREAKING) + +- Renamed modules and functions to use streamly's vocabulary instead of + conduit's. The functions return Streamly `Stream`s and `Fold`s, so the + modules are now named `Kafka.Streamly.Stream` and `Kafka.Streamly.Fold`, + and the function families are `kafkaStream*` and `kafkaFold*`. + + | Old (0.1.0.0) | New (0.2.0.0) | + |--------------------------------|------------------------------| + | `Kafka.Streamly.Source` | `Kafka.Streamly.Stream` | + | `Kafka.Streamly.Sink` | `Kafka.Streamly.Fold` | + | `kafkaSource` | `kafkaStream` | + | `kafkaSourceAutoClose` | `kafkaStreamAutoClose` | + | `kafkaSourceNoClose` | `kafkaStreamNoClose` | + | `kafkaSink` | `kafkaFold` | + | `kafkaBatchSink` | `kafkaBatchFold` | + + No backward-compatible re-exports are provided. Callers porting from + 0.1.0.0 should mechanically substitute identifiers per the table. + +## 0.1.0.0 — 2026-04-17 Initial release. diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -15,43 +15,41 @@ ```cabal build-depends: - , hw-kafka-streamly >=0.1 && <0.2 + , hw-kafka-streamly >=0.2 && <0.3 , hw-kafka-client >=5.3 && <6 - , streamly-core >=0.4 && <0.5 + , streamly-core >=0.3 && <0.5 ``` -> **Note:** `streamly-core` 0.4 is not yet published to Hackage at the time of -> writing. Until it is, depend on a `source-repository-package` for upstream -> Streamly in your `cabal.project`. The plan to land Streamly 0.4 on Hackage -> is tracked in the project's master plan -> (`docs/masterplans/1-release-hw-kafka-streamly-0.1.0.0.md`, EP-5/EP-6). +`hw-kafka-client` needs the `librdkafka` C library at build and run time +(Homebrew: `brew install librdkafka`; Debian/Ubuntu: `apt install +librdkafka-dev`). ## Modules -- `Kafka.Streamly.Source` — consumer streams, error predicates and filters, +- `Kafka.Streamly.Stream` — consumer streams, error predicates and filters, value-mapping helpers built on `Bifunctor`/`Bitraversable`. -- `Kafka.Streamly.Sink` — producer folds and a `withKafkaProducer` bracket +- `Kafka.Streamly.Fold` — producer folds and a `withKafkaProducer` bracket helper. - `Kafka.Streamly.Combinators` — batching combinators and helpers that throw `Left` values as exceptions. ## Consuming -The source module ships three variants that differ only in how they manage +The stream module ships three variants that differ only in how they manage the underlying `KafkaConsumer`: -- `kafkaSource` — creates the consumer from `ConsumerProperties` and +- `kafkaStream` — creates the consumer from `ConsumerProperties` and `Subscription`, closes it when the stream ends. Use this when the stream fully owns the consumer's lifecycle. -- `kafkaSourceAutoClose` — wraps a caller-supplied `KafkaConsumer` and closes +- `kafkaStreamAutoClose` — wraps a caller-supplied `KafkaConsumer` and closes it on stream end. Use this when the consumer is created elsewhere but its lifetime matches the stream. -- `kafkaSourceNoClose` — wraps a caller-supplied `KafkaConsumer` and leaves +- `kafkaStreamNoClose` — wraps a caller-supplied `KafkaConsumer` and leaves it open. Use this when the consumer outlives the stream. ```haskell import Kafka.Consumer -import Kafka.Streamly.Source (kafkaSource, skipNonFatal) +import Kafka.Streamly.Stream (kafkaStream, skipNonFatal) import Streamly.Data.Stream qualified as Stream main :: IO () @@ -62,16 +60,16 @@ sub = topics ["events"] <> offsetReset Earliest Stream.fold (Fold.drainBy print) . skipNonFatal - $ kafkaSource props sub (Timeout 1000) + $ kafkaStream props sub (Timeout 1000) ``` ## Producing -Producer sinks are Streamly `Fold`s: +Producer folds are Streamly `Fold`s: ```haskell import Kafka.Producer -import Kafka.Streamly.Sink (kafkaSink, withKafkaProducer) +import Kafka.Streamly.Fold (kafkaFold, withKafkaProducer) import Streamly.Data.Fold qualified as Fold import Streamly.Data.Stream qualified as Stream @@ -79,7 +77,7 @@ main = do let props = brokersList ["localhost:9092"] result <- withKafkaProducer props $ \producer -> - Stream.fold (kafkaSink producer) + Stream.fold (kafkaFold producer) . fmap mkRecord $ Stream.fromList ["a", "b", "c"] print result diff --git a/hw-kafka-streamly.cabal b/hw-kafka-streamly.cabal --- a/hw-kafka-streamly.cabal +++ b/hw-kafka-streamly.cabal @@ -1,12 +1,11 @@ cabal-version: 3.4 name: hw-kafka-streamly -version: 0.1.0.0 +version: 0.2.0.0 synopsis: Streamly bindings for hw-kafka-client description: Streamly streaming integration for hw-kafka-client, a Haskell - binding to Apache Kafka via librdkafka. Provides composable stream - sources for consuming and fold-based sinks for producing, with safe - resource management. + binding to Apache Kafka via librdkafka. Provides composable streams + for consuming and folds for producing, with safe resource management. license: MIT license-file: LICENSE @@ -37,8 +36,8 @@ import: warnings exposed-modules: Kafka.Streamly.Combinators - Kafka.Streamly.Sink - Kafka.Streamly.Source + Kafka.Streamly.Fold + Kafka.Streamly.Stream build-depends: , base >=4.21 && <5 @@ -65,7 +64,7 @@ hs-source-dirs: test other-modules: Kafka.Streamly.CombinatorsTest - Kafka.Streamly.SourceTest + Kafka.Streamly.StreamTest build-depends: , base >=4.21 && <5 diff --git a/src/Kafka/Streamly/Combinators.hs b/src/Kafka/Streamly/Combinators.hs --- a/src/Kafka/Streamly/Combinators.hs +++ b/src/Kafka/Streamly/Combinators.hs @@ -5,7 +5,7 @@ Auxiliary combinators that bridge consumer streams and producer folds: size-bounded batching with an explicit flush token, and helpers for callers who prefer exceptions over the @Either KafkaError a@ values that -'Kafka.Streamly.Source' yields. +'Kafka.Streamly.Stream' yields. Re-exports t'Kafka.Types.BatchSize' from "Kafka.Types" for convenience. -} diff --git a/src/Kafka/Streamly/Fold.hs b/src/Kafka/Streamly/Fold.hs new file mode 100644 --- /dev/null +++ b/src/Kafka/Streamly/Fold.hs @@ -0,0 +1,146 @@ +{- | +Module : Kafka.Streamly.Fold +Description : Producer folds and bracket helper for Streamly–Kafka pipelines. + +Streamly expresses consumers of a stream as 'Fold' values, the dual of +@Streamly.Data.Stream.Stream@. This module provides folds that send each incoming +'Kafka.Producer.ProducerRecord' to a Kafka broker, plus a bracket helper +('withKafkaProducer') for the producer's lifecycle. + +== Worked example + +Fold five records through 'kafkaFold': + +> import Kafka.Producer (ProducerRecord (..), TopicName (..), ProducePartition (..)) +> import Kafka.Streamly.Fold (kafkaFold, withKafkaProducer) +> import Streamly.Data.Fold qualified as Fold +> import Streamly.Data.Stream qualified as Stream +> +> mkRecord :: Int -> ProducerRecord +> mkRecord i = ProducerRecord +> { prTopic = TopicName "example-topic" +> , prPartition = UnassignedPartition +> , prKey = Nothing +> , prValue = Just (encodeUtf8 (pack ("payload-" <> show i))) +> , prHeaders = mempty +> } +> +> main :: IO () +> main = do +> result <- withKafkaProducer producerProps $ \\producer -> +> Stream.fold (kafkaFold producer) (Stream.fromList (map mkRecord [1..5])) +> print result -- Right Nothing on success + +== Delivery semantics + +A successful fold (@Nothing@) means every 'Kafka.Producer.produceMessage' call +returned without an error — which in librdkafka terms means the records were +/queued/ for delivery, not that the broker has acknowledged them. End-to-end +delivery requires a flush; 'withKafkaProducer' handles this on exit, or call +'Kafka.Producer.flushProducer' explicitly. +-} +module Kafka.Streamly.Fold ( + -- * Producer folds + kafkaFold, + kafkaBatchFold, + + -- * Resource management + withKafkaProducer, +) where + +import Control.Exception (bracket) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Kafka.Producer ( + KafkaError, + KafkaProducer, + ProducerProperties, + ProducerRecord, + closeProducer, + newProducer, + produceMessage, + ) +import Streamly.Data.Fold (Fold) +import Streamly.Data.Fold qualified as Fold + +{- | A 'Fold' that sends each 'ProducerRecord' to Kafka via the given producer. + +Returns 'Nothing' if every 'Kafka.Producer.produceMessage' call succeeded, or +'Just' the first t'KafkaError' encountered. After an error the fold still +consumes the remaining input but sends nothing. + +A 'Nothing' result means librdkafka has /queued/ all records for delivery; it +does not mean the broker has acknowledged them. For end-to-end delivery, run +the fold inside 'withKafkaProducer' (which flushes on exit) or call +'Kafka.Producer.flushProducer' directly. + +@since 0.1.0.0 +-} +kafkaFold :: + (MonadIO m) => + KafkaProducer -> + Fold m ProducerRecord (Maybe KafkaError) +kafkaFold producer = Fold.foldlM' step (pure Nothing) + where + step Nothing record = liftIO $ produceMessage producer record + step err@(Just _) _ = pure err +{-# INLINE kafkaFold #-} + +{- | A 'Fold' that sends batches of 'ProducerRecord' to Kafka. + +Returns 'Nothing' if every record in every batch was accepted by +'Kafka.Producer.produceMessage', or 'Just' the first t'KafkaError'. After an +error the fold still consumes remaining input but sends nothing. + +As with 'kafkaFold', 'Nothing' means records are queued in librdkafka, not +acknowledged by the broker — use 'withKafkaProducer' or +'Kafka.Producer.flushProducer' for delivery guarantees. + +Note: as of @hw-kafka-client-5.3.0@ each batch is sent as individual +'Kafka.Producer.produceMessage' calls because @produceMessageBatch@ is not +exported from that release. A future version of this library may switch to a +true broker-side batch send once the upstream dependency supports it. Today +this fold is a convenience for accepting @[ProducerRecord]@ input (for example +the output of 'Kafka.Streamly.Combinators.batchByOrFlush') — it does not +reduce the number of network round-trips compared to 'kafkaFold'. + +@since 0.1.0.0 +-} +kafkaBatchFold :: + (MonadIO m) => + KafkaProducer -> + Fold m [ProducerRecord] (Maybe KafkaError) +kafkaBatchFold producer = Fold.foldlM' step (pure Nothing) + where + step Nothing batch = sendBatch batch + step err@(Just _) _ = pure err + + sendBatch [] = pure Nothing + sendBatch (r : rs) = do + result <- liftIO $ produceMessage producer r + case result of + Nothing -> sendBatch rs + Just err -> pure (Just err) +{-# INLINE kafkaBatchFold #-} + +{- | Bracket producer creation and destruction around an action. + +Creates a producer from the given 'ProducerProperties', passes it to @action@, +then closes the producer on exit. Closing a @hw-kafka-client@ producer flushes +any queued records before releasing resources, so callers do not need to flush +explicitly. Returns 'Left' if producer creation fails, otherwise 'Right' with +the action's result. + +@since 0.1.0.0 +-} +withKafkaProducer :: + ProducerProperties -> + (KafkaProducer -> IO a) -> + IO (Either KafkaError a) +withKafkaProducer props action = + newProducer props >>= \case + Left err -> pure (Left err) + Right producer -> + bracket + (pure producer) + closeProducer + (\p -> Right <$> action p) diff --git a/src/Kafka/Streamly/Sink.hs b/src/Kafka/Streamly/Sink.hs deleted file mode 100644 --- a/src/Kafka/Streamly/Sink.hs +++ /dev/null @@ -1,146 +0,0 @@ -{- | -Module : Kafka.Streamly.Sink -Description : Producer folds and bracket helper for Streamly–Kafka pipelines. - -Streamly expresses consumers of a stream as 'Fold' values, the dual of -@Streamly.Data.Stream.Stream@. This module provides folds that send each incoming -'Kafka.Producer.ProducerRecord' to a Kafka broker, plus a bracket helper -('withKafkaProducer') for the producer's lifecycle. - -== Worked example - -Fold five records through 'kafkaSink': - -> import Kafka.Producer (ProducerRecord (..), TopicName (..), ProducePartition (..)) -> import Kafka.Streamly.Sink (kafkaSink, withKafkaProducer) -> import Streamly.Data.Fold qualified as Fold -> import Streamly.Data.Stream qualified as Stream -> -> mkRecord :: Int -> ProducerRecord -> mkRecord i = ProducerRecord -> { prTopic = TopicName "example-topic" -> , prPartition = UnassignedPartition -> , prKey = Nothing -> , prValue = Just (encodeUtf8 (pack ("payload-" <> show i))) -> , prHeaders = mempty -> } -> -> main :: IO () -> main = do -> result <- withKafkaProducer producerProps $ \\producer -> -> Stream.fold (kafkaSink producer) (Stream.fromList (map mkRecord [1..5])) -> print result -- Right Nothing on success - -== Delivery semantics - -A successful fold (@Nothing@) means every 'Kafka.Producer.produceMessage' call -returned without an error — which in librdkafka terms means the records were -/queued/ for delivery, not that the broker has acknowledged them. End-to-end -delivery requires a flush; 'withKafkaProducer' handles this on exit, or call -'Kafka.Producer.flushProducer' explicitly. --} -module Kafka.Streamly.Sink ( - -- * Producer folds - kafkaSink, - kafkaBatchSink, - - -- * Resource management - withKafkaProducer, -) where - -import Control.Exception (bracket) -import Control.Monad.IO.Class (MonadIO, liftIO) -import Kafka.Producer ( - KafkaError, - KafkaProducer, - ProducerProperties, - ProducerRecord, - closeProducer, - newProducer, - produceMessage, - ) -import Streamly.Data.Fold (Fold) -import Streamly.Data.Fold qualified as Fold - -{- | A 'Fold' that sends each 'ProducerRecord' to Kafka via the given producer. - -Returns 'Nothing' if every 'Kafka.Producer.produceMessage' call succeeded, or -'Just' the first t'KafkaError' encountered. After an error the fold still -consumes the remaining input but sends nothing. - -A 'Nothing' result means librdkafka has /queued/ all records for delivery; it -does not mean the broker has acknowledged them. For end-to-end delivery, run -the fold inside 'withKafkaProducer' (which flushes on exit) or call -'Kafka.Producer.flushProducer' directly. - -@since 0.1.0.0 --} -kafkaSink :: - (MonadIO m) => - KafkaProducer -> - Fold m ProducerRecord (Maybe KafkaError) -kafkaSink producer = Fold.foldlM' step (pure Nothing) - where - step Nothing record = liftIO $ produceMessage producer record - step err@(Just _) _ = pure err -{-# INLINE kafkaSink #-} - -{- | A 'Fold' that sends batches of 'ProducerRecord' to Kafka. - -Returns 'Nothing' if every record in every batch was accepted by -'Kafka.Producer.produceMessage', or 'Just' the first t'KafkaError'. After an -error the fold still consumes remaining input but sends nothing. - -As with 'kafkaSink', 'Nothing' means records are queued in librdkafka, not -acknowledged by the broker — use 'withKafkaProducer' or -'Kafka.Producer.flushProducer' for delivery guarantees. - -Note: as of @hw-kafka-client-5.3.0@ each batch is sent as individual -'Kafka.Producer.produceMessage' calls because @produceMessageBatch@ is not -exported from that release. A future version of this library may switch to a -true broker-side batch send once the upstream dependency supports it. Today -this fold is a convenience for accepting @[ProducerRecord]@ input (for example -the output of 'Kafka.Streamly.Combinators.batchByOrFlush') — it does not -reduce the number of network round-trips compared to 'kafkaSink'. - -@since 0.1.0.0 --} -kafkaBatchSink :: - (MonadIO m) => - KafkaProducer -> - Fold m [ProducerRecord] (Maybe KafkaError) -kafkaBatchSink producer = Fold.foldlM' step (pure Nothing) - where - step Nothing batch = sendBatch batch - step err@(Just _) _ = pure err - - sendBatch [] = pure Nothing - sendBatch (r : rs) = do - result <- liftIO $ produceMessage producer r - case result of - Nothing -> sendBatch rs - Just err -> pure (Just err) -{-# INLINE kafkaBatchSink #-} - -{- | Bracket producer creation and destruction around an action. - -Creates a producer from the given 'ProducerProperties', passes it to @action@, -then closes the producer on exit. Closing a @hw-kafka-client@ producer flushes -any queued records before releasing resources, so callers do not need to flush -explicitly. Returns 'Left' if producer creation fails, otherwise 'Right' with -the action's result. - -@since 0.1.0.0 --} -withKafkaProducer :: - ProducerProperties -> - (KafkaProducer -> IO a) -> - IO (Either KafkaError a) -withKafkaProducer props action = - newProducer props >>= \case - Left err -> pure (Left err) - Right producer -> - bracket - (pure producer) - closeProducer - (\p -> Right <$> action p) diff --git a/src/Kafka/Streamly/Source.hs b/src/Kafka/Streamly/Source.hs deleted file mode 100644 --- a/src/Kafka/Streamly/Source.hs +++ /dev/null @@ -1,329 +0,0 @@ -{- | -Module : Kafka.Streamly.Source -Description : Consumer stream sources and helpers for Streamly–Kafka pipelines. - -Stream sources backed by a @hw-kafka-client@ 'KafkaConsumer'. Each source -yields @Either t'Kafka.Consumer.KafkaError' ('ConsumerRecord' (Maybe ByteString) (Maybe ByteString))@ — -errors are kept in-band rather than raised as exceptions, matching the shape -exposed by 'Kafka.Consumer.pollMessage'. - -== Three tiers of resource management - -* 'kafkaSource' — fully managed: creates a consumer, polls, closes on stream - termination. If consumer creation fails the t'KafkaError' is thrown as an - exception (see the function's docstring for how this differs from - @hw-kafka-conduit@). -* 'kafkaSourceAutoClose' — takes a pre-built consumer, polls, closes on - termination. The caller owns creation, the stream owns destruction. -* 'kafkaSourceNoClose' — polls a consumer the caller fully owns. The stream - never touches the consumer's lifecycle. - -== Worked example - -Consume five records from a topic, skipping non-fatal errors: - -> import Kafka.Consumer -> ( ConsumerGroupId (..), ConsumerRecord, KafkaError -> , OffsetReset (..), brokersList, groupId, offsetReset, topics -> ) -> import Kafka.Streamly.Source (kafkaSource, skipNonFatal) -> import Streamly.Data.Fold qualified as Fold -> import Streamly.Data.Stream qualified as Stream -> -> main :: IO () -> main = do -> let props = brokersList ["localhost:9092"] -> <> groupId (ConsumerGroupId "example-group") -> sub = topics ["example-topic"] <> offsetReset Earliest -> Stream.fold (Fold.drainMapM print) $ -> Stream.take 5 $ -> skipNonFatal $ -> kafkaSource props sub (Timeout 1000) - -== Transforming the stream - -Because the element type is @Either KafkaError (ConsumerRecord k v)@, the -idiomatic way to map the inner record value is a double @fmap@: - -> fmap (fmap (\\r -> r { crValue = Nothing })) stream - -The helpers 'mapFirst', 'mapValue', and 'bimapValue' save one level of wrapping -by lifting a function directly into the outer stream element — see their -docstrings. --} -module Kafka.Streamly.Source ( - -- * Stream sources - kafkaSource, - kafkaSourceAutoClose, - kafkaSourceNoClose, - - -- * Error predicates - isFatal, - isPollTimeout, - isPartitionEOF, - - -- * Error filters - skipNonFatal, - skipNonFatalExcept, - - -- * Value mapping - mapFirst, - mapValue, - bimapValue, -) where - -import Control.Exception (throwIO) -import Control.Monad.Catch (MonadCatch) -import Control.Monad.IO.Class (MonadIO, liftIO) -import Data.Bifunctor (Bifunctor, bimap, first) -import Data.ByteString qualified as BS -import Kafka.Consumer ( - ConsumerProperties, - ConsumerRecord, - KafkaConsumer, - KafkaError (..), - RdKafkaRespErrT (..), - Subscription, - Timeout, - closeConsumer, - newConsumer, - pollMessage, - ) -import Streamly.Data.Stream (Stream) -import Streamly.Data.Stream qualified as Stream - -{- | Create a 'Stream' for a given 'KafkaConsumer'. - -The consumer will /not/ be closed when the stream ends; the caller owns its -lifecycle. The stream terminates the first time 'pollMessage' returns a fatal -t'KafkaError' (see 'isFatal'); non-fatal errors are yielded in-band as @Left@ -values and polling continues. - -@since 0.1.0.0 --} -kafkaSourceNoClose :: - (MonadIO m) => - KafkaConsumer -> - Timeout -> - Stream m (Either KafkaError (ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString))) -kafkaSourceNoClose consumer timeout = - Stream.unfoldrM step True - where - step False = pure Nothing - step True = do - msg <- liftIO $ pollMessage consumer timeout - case msg of - Left err | isFatal err -> pure $ Just (Left err, False) - _ -> pure $ Just (msg, True) -{-# INLINE kafkaSourceNoClose #-} - -{- | Create a 'Stream' for a given 'KafkaConsumer'. - -The consumer is passed in by the caller but will be closed automatically when -the stream ends (via 'Streamly.Data.Stream.bracketIO'). Use this when you want -the stream to own destruction but not creation. - -@since 0.1.0.0 --} -kafkaSourceAutoClose :: - (MonadIO m, MonadCatch m) => - KafkaConsumer -> - Timeout -> - Stream m (Either KafkaError (ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString))) -kafkaSourceAutoClose consumer timeout = - Stream.bracketIO - (pure consumer) - (\c -> () <$ closeConsumer c) - (\c -> kafkaSourceNoClose c timeout) -{-# INLINE kafkaSourceAutoClose #-} - -{- | Create a fully managed consumer 'Stream' from 'ConsumerProperties' and a -'Subscription'. - -The stream creates the consumer on first demand, polls messages, and closes -the consumer when the stream ends. - -If consumer creation fails, the t'KafkaError' is thrown as an exception (via -'throwIO'). This diverges from @hw-kafka-conduit@'s @kafkaSource@, which -yields @Left err@ as the first stream value and then terminates. Callers -migrating from conduit who prefer the in-band error should create the -consumer manually with 'Kafka.Consumer.newConsumer' and, on @Right c@, pass -@c@ to 'kafkaSourceAutoClose'. - -Per-poll errors (non-fatal or fatal) are yielded in-band as @Left@ values, -as with the other sources in this module. - -@since 0.1.0.0 --} -kafkaSource :: - (MonadIO m, MonadCatch m) => - ConsumerProperties -> - Subscription -> - Timeout -> - Stream m (Either KafkaError (ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString))) -kafkaSource props sub timeout = - Stream.bracketIO - ( newConsumer props sub >>= \case - Left err -> throwIO err - Right c -> pure c - ) - (\c -> () <$ closeConsumer c) - (\c -> kafkaSourceNoClose c timeout) -{-# INLINE kafkaSource #-} - -------------------------------------------------------------------------------- --- Error predicates -------------------------------------------------------------------------------- - -{- | Checks if the error is fatal in a way that it doesn't make sense to retry -after or is unsafe to ignore. - -@since 0.1.0.0 --} -isFatal :: KafkaError -> Bool -isFatal = \case - KafkaUnknownConfigurationKey _ -> True - KafkaInvalidConfigurationValue _ -> True - KafkaBadConfiguration -> True - KafkaBadSpecification _ -> True - KafkaResponseError RdKafkaRespErrDestroy -> True - KafkaResponseError RdKafkaRespErrFail -> True - KafkaResponseError RdKafkaRespErrInvalidArg -> True - KafkaResponseError RdKafkaRespErrSsl -> True - KafkaResponseError RdKafkaRespErrUnknownProtocol -> True - KafkaResponseError RdKafkaRespErrNotImplemented -> True - KafkaResponseError RdKafkaRespErrAuthentication -> True - KafkaResponseError RdKafkaRespErrInconsistentGroupProtocol -> True - KafkaResponseError RdKafkaRespErrTopicAuthorizationFailed -> True - KafkaResponseError RdKafkaRespErrGroupAuthorizationFailed -> True - KafkaResponseError RdKafkaRespErrClusterAuthorizationFailed -> True - KafkaResponseError RdKafkaRespErrUnsupportedSaslMechanism -> True - KafkaResponseError RdKafkaRespErrIllegalSaslState -> True - KafkaResponseError RdKafkaRespErrUnsupportedVersion -> True - _ -> False -{-# INLINE isFatal #-} - -{- | Checks if the error is a poll timeout ('RdKafkaRespErrTimedOut'). -Timeout errors are not fatal and occur when 'pollMessage' has no messages -to return within the specified timeout. - -@since 0.1.0.0 --} -isPollTimeout :: KafkaError -> Bool -isPollTimeout e = KafkaResponseError RdKafkaRespErrTimedOut == e -{-# INLINE isPollTimeout #-} - -{- | Checks if the error indicates reaching the end of a partition -('RdKafkaRespErrPartitionEof'). Not fatal; occurs every time a consumer -reaches the current end of a partition. - -@since 0.1.0.0 --} -isPartitionEOF :: KafkaError -> Bool -isPartitionEOF e = KafkaResponseError RdKafkaRespErrPartitionEof == e -{-# INLINE isPartitionEOF #-} - -------------------------------------------------------------------------------- --- Error filters -------------------------------------------------------------------------------- - -{- | Filter out non-fatal errors, keeping only fatal errors and successful -records. - -This drops every t'KafkaError' for which 'isFatal' returns 'False', which -includes poll timeouts ('RdKafkaRespErrTimedOut') /and/ partition-EOF markers -('RdKafkaRespErrPartitionEof'). If your consumer relies on observing EOFs — -for example, to terminate a bounded read — use 'skipNonFatalExcept' with -'isPartitionEOF' instead. - -@since 0.1.0.0 --} -skipNonFatal :: - (Monad m) => - Stream m (Either KafkaError b) -> - Stream m (Either KafkaError b) -skipNonFatal = Stream.filter (either isFatal (const True)) -{-# INLINE skipNonFatal #-} - -{- | Filter out non-fatal errors except those matching any of the given -predicates. Fatal errors always pass through. - -> skipNonFatalExcept [isPollTimeout, isPartitionEOF] - -@since 0.1.0.0 --} -skipNonFatalExcept :: - (Monad m) => - [KafkaError -> Bool] -> - Stream m (Either KafkaError b) -> - Stream m (Either KafkaError b) -skipNonFatalExcept fs = - let fun e = or $ (\f -> f e) <$> (isFatal : fs) - in Stream.filter (either fun (const True)) -{-# INLINE skipNonFatalExcept #-} - -------------------------------------------------------------------------------- --- Value mapping -------------------------------------------------------------------------------- - -{- | Lift a function into the left position of a 'Bifunctor' stream element. - -Given a stream of @t k v@ (for example @Either KafkaError (ConsumerRecord k v)@), -transform the left component. To map the key of a 'ConsumerRecord' inside an -@Either@ the caller still has to compose with an outer 'fmap': - -> mapFirst (fmap prefixKey) :: Stream m (Either e (ConsumerRecord k v)) -> ... - -@since 0.1.0.0 --} -mapFirst :: - (Bifunctor t, Monad m) => - (k -> k') -> - Stream m (t k v) -> - Stream m (t k' v) -mapFirst f = fmap (first f) -{-# INLINE mapFirst #-} - -{- | Lift a function into the 'Functor' position of a stream element. - -For @Stream m (Either e (ConsumerRecord k v))@, @mapValue f@ rewrites the -@Right@ side, leaving errors untouched. To transform the record /value/ -itself the caller composes with another 'fmap': - -> mapValue (fmap uppercase) :: Stream m (Either e (ConsumerRecord k v)) -> ... - -@since 0.1.0.0 --} -mapValue :: - (Functor t, Monad m) => - (v -> v') -> - Stream m (t v) -> - Stream m (t v') -mapValue f = fmap (fmap f) -{-# INLINE mapValue #-} - -{- | Lift a pair of functions into the two positions of a 'Bifunctor' stream -element. - -Acts on the outer element. For @Stream m (Either KafkaError x)@ this maps -both the error and the success: - -> bimapValue renderError handleSuccess -> :: Stream m (Either KafkaError x) -> Stream m (Either String y) - -To bimap the key and value /inside/ an outer 'Either', wrap with 'fmap': - -> mapValue (bimap prefixKey uppercase) -> :: Stream m (Either e (ConsumerRecord k v)) -> -> Stream m (Either e (ConsumerRecord k' v')) - -@since 0.1.0.0 --} -bimapValue :: - (Bifunctor t, Monad m) => - (k -> k') -> - (v -> v') -> - Stream m (t k v) -> - Stream m (t k' v') -bimapValue f g = fmap (bimap f g) -{-# INLINE bimapValue #-} diff --git a/src/Kafka/Streamly/Stream.hs b/src/Kafka/Streamly/Stream.hs new file mode 100644 --- /dev/null +++ b/src/Kafka/Streamly/Stream.hs @@ -0,0 +1,329 @@ +{- | +Module : Kafka.Streamly.Stream +Description : Consumer streams and helpers for Streamly–Kafka pipelines. + +Streams backed by a @hw-kafka-client@ 'KafkaConsumer'. Each stream +yields @Either t'Kafka.Consumer.KafkaError' ('ConsumerRecord' (Maybe ByteString) (Maybe ByteString))@ — +errors are kept in-band rather than raised as exceptions, matching the shape +exposed by 'Kafka.Consumer.pollMessage'. + +== Three tiers of resource management + +* 'kafkaStream' — fully managed: creates a consumer, polls, closes on stream + termination. If consumer creation fails the t'KafkaError' is thrown as an + exception (see the function's docstring for how this differs from + @hw-kafka-conduit@). +* 'kafkaStreamAutoClose' — takes a pre-built consumer, polls, closes on + termination. The caller owns creation, the stream owns destruction. +* 'kafkaStreamNoClose' — polls a consumer the caller fully owns. The stream + never touches the consumer's lifecycle. + +== Worked example + +Consume five records from a topic, skipping non-fatal errors: + +> import Kafka.Consumer +> ( ConsumerGroupId (..), ConsumerRecord, KafkaError +> , OffsetReset (..), brokersList, groupId, offsetReset, topics +> ) +> import Kafka.Streamly.Stream (kafkaStream, skipNonFatal) +> import Streamly.Data.Fold qualified as Fold +> import Streamly.Data.Stream qualified as Stream +> +> main :: IO () +> main = do +> let props = brokersList ["localhost:9092"] +> <> groupId (ConsumerGroupId "example-group") +> sub = topics ["example-topic"] <> offsetReset Earliest +> Stream.fold (Fold.drainMapM print) $ +> Stream.take 5 $ +> skipNonFatal $ +> kafkaStream props sub (Timeout 1000) + +== Transforming the stream + +Because the element type is @Either KafkaError (ConsumerRecord k v)@, the +idiomatic way to map the inner record value is a double @fmap@: + +> fmap (fmap (\\r -> r { crValue = Nothing })) stream + +The helpers 'mapFirst', 'mapValue', and 'bimapValue' save one level of wrapping +by lifting a function directly into the outer stream element — see their +docstrings. +-} +module Kafka.Streamly.Stream ( + -- * Streams + kafkaStream, + kafkaStreamAutoClose, + kafkaStreamNoClose, + + -- * Error predicates + isFatal, + isPollTimeout, + isPartitionEOF, + + -- * Error filters + skipNonFatal, + skipNonFatalExcept, + + -- * Value mapping + mapFirst, + mapValue, + bimapValue, +) where + +import Control.Exception (throwIO) +import Control.Monad.Catch (MonadCatch) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Data.Bifunctor (Bifunctor, bimap, first) +import Data.ByteString qualified as BS +import Kafka.Consumer ( + ConsumerProperties, + ConsumerRecord, + KafkaConsumer, + KafkaError (..), + RdKafkaRespErrT (..), + Subscription, + Timeout, + closeConsumer, + newConsumer, + pollMessage, + ) +import Streamly.Data.Stream (Stream) +import Streamly.Data.Stream qualified as Stream + +{- | Create a 'Stream' for a given 'KafkaConsumer'. + +The consumer will /not/ be closed when the stream ends; the caller owns its +lifecycle. The stream terminates the first time 'pollMessage' returns a fatal +t'KafkaError' (see 'isFatal'); non-fatal errors are yielded in-band as @Left@ +values and polling continues. + +@since 0.1.0.0 +-} +kafkaStreamNoClose :: + (MonadIO m) => + KafkaConsumer -> + Timeout -> + Stream m (Either KafkaError (ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString))) +kafkaStreamNoClose consumer timeout = + Stream.unfoldrM step True + where + step False = pure Nothing + step True = do + msg <- liftIO $ pollMessage consumer timeout + case msg of + Left err | isFatal err -> pure $ Just (Left err, False) + _ -> pure $ Just (msg, True) +{-# INLINE kafkaStreamNoClose #-} + +{- | Create a 'Stream' for a given 'KafkaConsumer'. + +The consumer is passed in by the caller but will be closed automatically when +the stream ends (via 'Streamly.Data.Stream.bracketIO'). Use this when you want +the stream to own destruction but not creation. + +@since 0.1.0.0 +-} +kafkaStreamAutoClose :: + (MonadIO m, MonadCatch m) => + KafkaConsumer -> + Timeout -> + Stream m (Either KafkaError (ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString))) +kafkaStreamAutoClose consumer timeout = + Stream.bracketIO + (pure consumer) + (\c -> () <$ closeConsumer c) + (\c -> kafkaStreamNoClose c timeout) +{-# INLINE kafkaStreamAutoClose #-} + +{- | Create a fully managed consumer 'Stream' from 'ConsumerProperties' and a +'Subscription'. + +The stream creates the consumer on first demand, polls messages, and closes +the consumer when the stream ends. + +If consumer creation fails, the t'KafkaError' is thrown as an exception (via +'throwIO'). This diverges from @hw-kafka-conduit@'s @kafkaSource@, which +yields @Left err@ as the first stream value and then terminates. Callers +migrating from conduit who prefer the in-band error should create the +consumer manually with 'Kafka.Consumer.newConsumer' and, on @Right c@, pass +@c@ to 'kafkaStreamAutoClose'. + +Per-poll errors (non-fatal or fatal) are yielded in-band as @Left@ values, +as with the other streams in this module. + +@since 0.1.0.0 +-} +kafkaStream :: + (MonadIO m, MonadCatch m) => + ConsumerProperties -> + Subscription -> + Timeout -> + Stream m (Either KafkaError (ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString))) +kafkaStream props sub timeout = + Stream.bracketIO + ( newConsumer props sub >>= \case + Left err -> throwIO err + Right c -> pure c + ) + (\c -> () <$ closeConsumer c) + (\c -> kafkaStreamNoClose c timeout) +{-# INLINE kafkaStream #-} + +------------------------------------------------------------------------------- +-- Error predicates +------------------------------------------------------------------------------- + +{- | Checks if the error is fatal in a way that it doesn't make sense to retry +after or is unsafe to ignore. + +@since 0.1.0.0 +-} +isFatal :: KafkaError -> Bool +isFatal = \case + KafkaUnknownConfigurationKey _ -> True + KafkaInvalidConfigurationValue _ -> True + KafkaBadConfiguration -> True + KafkaBadSpecification _ -> True + KafkaResponseError RdKafkaRespErrDestroy -> True + KafkaResponseError RdKafkaRespErrFail -> True + KafkaResponseError RdKafkaRespErrInvalidArg -> True + KafkaResponseError RdKafkaRespErrSsl -> True + KafkaResponseError RdKafkaRespErrUnknownProtocol -> True + KafkaResponseError RdKafkaRespErrNotImplemented -> True + KafkaResponseError RdKafkaRespErrAuthentication -> True + KafkaResponseError RdKafkaRespErrInconsistentGroupProtocol -> True + KafkaResponseError RdKafkaRespErrTopicAuthorizationFailed -> True + KafkaResponseError RdKafkaRespErrGroupAuthorizationFailed -> True + KafkaResponseError RdKafkaRespErrClusterAuthorizationFailed -> True + KafkaResponseError RdKafkaRespErrUnsupportedSaslMechanism -> True + KafkaResponseError RdKafkaRespErrIllegalSaslState -> True + KafkaResponseError RdKafkaRespErrUnsupportedVersion -> True + _ -> False +{-# INLINE isFatal #-} + +{- | Checks if the error is a poll timeout ('RdKafkaRespErrTimedOut'). +Timeout errors are not fatal and occur when 'pollMessage' has no messages +to return within the specified timeout. + +@since 0.1.0.0 +-} +isPollTimeout :: KafkaError -> Bool +isPollTimeout e = KafkaResponseError RdKafkaRespErrTimedOut == e +{-# INLINE isPollTimeout #-} + +{- | Checks if the error indicates reaching the end of a partition +('RdKafkaRespErrPartitionEof'). Not fatal; occurs every time a consumer +reaches the current end of a partition. + +@since 0.1.0.0 +-} +isPartitionEOF :: KafkaError -> Bool +isPartitionEOF e = KafkaResponseError RdKafkaRespErrPartitionEof == e +{-# INLINE isPartitionEOF #-} + +------------------------------------------------------------------------------- +-- Error filters +------------------------------------------------------------------------------- + +{- | Filter out non-fatal errors, keeping only fatal errors and successful +records. + +This drops every t'KafkaError' for which 'isFatal' returns 'False', which +includes poll timeouts ('RdKafkaRespErrTimedOut') /and/ partition-EOF markers +('RdKafkaRespErrPartitionEof'). If your consumer relies on observing EOFs — +for example, to terminate a bounded read — use 'skipNonFatalExcept' with +'isPartitionEOF' instead. + +@since 0.1.0.0 +-} +skipNonFatal :: + (Monad m) => + Stream m (Either KafkaError b) -> + Stream m (Either KafkaError b) +skipNonFatal = Stream.filter (either isFatal (const True)) +{-# INLINE skipNonFatal #-} + +{- | Filter out non-fatal errors except those matching any of the given +predicates. Fatal errors always pass through. + +> skipNonFatalExcept [isPollTimeout, isPartitionEOF] + +@since 0.1.0.0 +-} +skipNonFatalExcept :: + (Monad m) => + [KafkaError -> Bool] -> + Stream m (Either KafkaError b) -> + Stream m (Either KafkaError b) +skipNonFatalExcept fs = + let fun e = or $ (\f -> f e) <$> (isFatal : fs) + in Stream.filter (either fun (const True)) +{-# INLINE skipNonFatalExcept #-} + +------------------------------------------------------------------------------- +-- Value mapping +------------------------------------------------------------------------------- + +{- | Lift a function into the left position of a 'Bifunctor' stream element. + +Given a stream of @t k v@ (for example @Either KafkaError (ConsumerRecord k v)@), +transform the left component. To map the key of a 'ConsumerRecord' inside an +@Either@ the caller still has to compose with an outer 'fmap': + +> mapFirst (fmap prefixKey) :: Stream m (Either e (ConsumerRecord k v)) -> ... + +@since 0.1.0.0 +-} +mapFirst :: + (Bifunctor t, Monad m) => + (k -> k') -> + Stream m (t k v) -> + Stream m (t k' v) +mapFirst f = fmap (first f) +{-# INLINE mapFirst #-} + +{- | Lift a function into the 'Functor' position of a stream element. + +For @Stream m (Either e (ConsumerRecord k v))@, @mapValue f@ rewrites the +@Right@ side, leaving errors untouched. To transform the record /value/ +itself the caller composes with another 'fmap': + +> mapValue (fmap uppercase) :: Stream m (Either e (ConsumerRecord k v)) -> ... + +@since 0.1.0.0 +-} +mapValue :: + (Functor t, Monad m) => + (v -> v') -> + Stream m (t v) -> + Stream m (t v') +mapValue f = fmap (fmap f) +{-# INLINE mapValue #-} + +{- | Lift a pair of functions into the two positions of a 'Bifunctor' stream +element. + +Acts on the outer element. For @Stream m (Either KafkaError x)@ this maps +both the error and the success: + +> bimapValue renderError handleSuccess +> :: Stream m (Either KafkaError x) -> Stream m (Either String y) + +To bimap the key and value /inside/ an outer 'Either', wrap with 'fmap': + +> mapValue (bimap prefixKey uppercase) +> :: Stream m (Either e (ConsumerRecord k v)) +> -> Stream m (Either e (ConsumerRecord k' v')) + +@since 0.1.0.0 +-} +bimapValue :: + (Bifunctor t, Monad m) => + (k -> k') -> + (v -> v') -> + Stream m (t k v) -> + Stream m (t k' v') +bimapValue f g = fmap (bimap f g) +{-# INLINE bimapValue #-} diff --git a/test/Kafka/Streamly/SourceTest.hs b/test/Kafka/Streamly/SourceTest.hs deleted file mode 100644 --- a/test/Kafka/Streamly/SourceTest.hs +++ /dev/null @@ -1,204 +0,0 @@ -module Kafka.Streamly.SourceTest (tests) where - -import Kafka.Consumer (KafkaError (..), RdKafkaRespErrT (..)) -import Kafka.Streamly.Source ( - isFatal, - isPartitionEOF, - isPollTimeout, - skipNonFatal, - skipNonFatalExcept, - ) -import Streamly.Data.Fold qualified as Fold -import Streamly.Data.Stream (Stream) -import Streamly.Data.Stream qualified as Stream -import Test.Tasty (TestTree, testGroup) -import Test.Tasty.HUnit (Assertion, testCase, (@?=)) - -tests :: TestTree -tests = - testGroup - "Source" - [ testGroup "isFatal" isFatalTests - , testGroup "isPollTimeout" isPollTimeoutTests - , testGroup "isPartitionEOF" isPartitionEOFTests - , testGroup "skipNonFatal" skipNonFatalTests - , testGroup "skipNonFatalExcept" skipNonFatalExceptTests - ] - -------------------------------------------------------------------------------- --- isFatal -------------------------------------------------------------------------------- - -fatalErrors :: [(String, KafkaError)] -fatalErrors = - [ ("KafkaUnknownConfigurationKey", KafkaUnknownConfigurationKey "") - , ("KafkaInvalidConfigurationValue", KafkaInvalidConfigurationValue "") - , ("KafkaBadConfiguration", KafkaBadConfiguration) - , ("KafkaBadSpecification", KafkaBadSpecification "") - , ("RdKafkaRespErrDestroy", KafkaResponseError RdKafkaRespErrDestroy) - , ("RdKafkaRespErrFail", KafkaResponseError RdKafkaRespErrFail) - , ("RdKafkaRespErrInvalidArg", KafkaResponseError RdKafkaRespErrInvalidArg) - , ("RdKafkaRespErrSsl", KafkaResponseError RdKafkaRespErrSsl) - , ("RdKafkaRespErrUnknownProtocol", KafkaResponseError RdKafkaRespErrUnknownProtocol) - , ("RdKafkaRespErrNotImplemented", KafkaResponseError RdKafkaRespErrNotImplemented) - , ("RdKafkaRespErrAuthentication", KafkaResponseError RdKafkaRespErrAuthentication) - , ("RdKafkaRespErrInconsistentGroupProtocol", KafkaResponseError RdKafkaRespErrInconsistentGroupProtocol) - , ("RdKafkaRespErrTopicAuthorizationFailed", KafkaResponseError RdKafkaRespErrTopicAuthorizationFailed) - , ("RdKafkaRespErrGroupAuthorizationFailed", KafkaResponseError RdKafkaRespErrGroupAuthorizationFailed) - , ("RdKafkaRespErrClusterAuthorizationFailed", KafkaResponseError RdKafkaRespErrClusterAuthorizationFailed) - , ("RdKafkaRespErrUnsupportedSaslMechanism", KafkaResponseError RdKafkaRespErrUnsupportedSaslMechanism) - , ("RdKafkaRespErrIllegalSaslState", KafkaResponseError RdKafkaRespErrIllegalSaslState) - , ("RdKafkaRespErrUnsupportedVersion", KafkaResponseError RdKafkaRespErrUnsupportedVersion) - ] - -nonFatalErrors :: [(String, KafkaError)] -nonFatalErrors = - [ ("RdKafkaRespErrTimedOut", KafkaResponseError RdKafkaRespErrTimedOut) - , ("RdKafkaRespErrPartitionEof", KafkaResponseError RdKafkaRespErrPartitionEof) - , ("RdKafkaRespErrNoError", KafkaResponseError RdKafkaRespErrNoError) - ] - -isFatalTests :: [TestTree] -isFatalTests = - [ testCase ("fatal: " <> label) (isFatal err @?= True) - | (label, err) <- fatalErrors - ] - <> [ testCase ("non-fatal: " <> label) (isFatal err @?= False) - | (label, err) <- nonFatalErrors - ] - -------------------------------------------------------------------------------- --- isPollTimeout -------------------------------------------------------------------------------- - -isPollTimeoutTests :: [TestTree] -isPollTimeoutTests = - [ testCase "matches RdKafkaRespErrTimedOut" $ - isPollTimeout (KafkaResponseError RdKafkaRespErrTimedOut) @?= True - , testCase "rejects RdKafkaRespErrPartitionEof" $ - isPollTimeout (KafkaResponseError RdKafkaRespErrPartitionEof) @?= False - , testCase "rejects RdKafkaRespErrAuthentication" $ - isPollTimeout (KafkaResponseError RdKafkaRespErrAuthentication) @?= False - , testCase "rejects KafkaBadConfiguration" $ - isPollTimeout KafkaBadConfiguration @?= False - ] - -------------------------------------------------------------------------------- --- isPartitionEOF -------------------------------------------------------------------------------- - -isPartitionEOFTests :: [TestTree] -isPartitionEOFTests = - [ testCase "matches RdKafkaRespErrPartitionEof" $ - isPartitionEOF (KafkaResponseError RdKafkaRespErrPartitionEof) @?= True - , testCase "rejects RdKafkaRespErrTimedOut" $ - isPartitionEOF (KafkaResponseError RdKafkaRespErrTimedOut) @?= False - , testCase "rejects RdKafkaRespErrAuthentication" $ - isPartitionEOF (KafkaResponseError RdKafkaRespErrAuthentication) @?= False - , testCase "rejects KafkaBadConfiguration" $ - isPartitionEOF KafkaBadConfiguration @?= False - ] - -------------------------------------------------------------------------------- --- skipNonFatal / skipNonFatalExcept -------------------------------------------------------------------------------- - -runStream :: - (Stream IO a -> Stream IO b) -> - [a] -> - IO [b] -runStream f xs = Stream.fold Fold.toList (f (Stream.fromList xs)) - -timedOut, partitionEof, authentication :: KafkaError -timedOut = KafkaResponseError RdKafkaRespErrTimedOut -partitionEof = KafkaResponseError RdKafkaRespErrPartitionEof -authentication = KafkaResponseError RdKafkaRespErrAuthentication - -skipNonFatalTests :: [TestTree] -skipNonFatalTests = - [ testCase "drops poll-timeout, keeps authentication, keeps all Rights" dropsAndKeeps - , testCase "passes all Rights through unchanged" passesAllRights - , testCase "drops non-fatal Lefts even with empty extension list" dropsNonFatal - , testCase "drops partition-EOF (non-fatal)" dropsPartitionEof - ] - where - dropsAndKeeps :: Assertion - dropsAndKeeps = do - let input = - [ Right (1 :: Int) - , Left timedOut - , Right 2 - , Left authentication - , Right 3 - ] - result <- runStream skipNonFatal input - result @?= [Right 1, Right 2, Left authentication, Right 3] - - passesAllRights :: Assertion - passesAllRights = do - let input = [Right (n :: Int) | n <- [1 .. 5]] - result <- runStream skipNonFatal input - result @?= input - - dropsNonFatal :: Assertion - dropsNonFatal = do - let input :: [Either KafkaError Int] - input = - [ Left timedOut - , Left partitionEof - , Left (KafkaResponseError RdKafkaRespErrNoError) - ] - result <- runStream skipNonFatal input - result @?= [] - - dropsPartitionEof :: Assertion - dropsPartitionEof = do - let input = - [ Right (1 :: Int) - , Left partitionEof - , Right 2 - ] - result <- runStream skipNonFatal input - result @?= [Right 1, Right 2] - -skipNonFatalExceptTests :: [TestTree] -skipNonFatalExceptTests = - [ testCase "keeps poll-timeout when isPollTimeout is in the list" keepsTimeouts - , testCase "keeps partition-EOF when isPartitionEOF is in the list" keepsPartitionEof - , testCase "empty predicate list behaves like skipNonFatal" emptyListEquivalent - ] - where - keepsTimeouts :: Assertion - keepsTimeouts = do - let input = - [ Right (1 :: Int) - , Left timedOut - , Right 2 - , Left authentication - , Right 3 - ] - result <- runStream (skipNonFatalExcept [isPollTimeout]) input - result @?= [Right 1, Left timedOut, Right 2, Left authentication, Right 3] - - keepsPartitionEof :: Assertion - keepsPartitionEof = do - let input = - [ Right (1 :: Int) - , Left partitionEof - , Left timedOut - , Right 2 - ] - result <- runStream (skipNonFatalExcept [isPartitionEOF]) input - result @?= [Right 1, Left partitionEof, Right 2] - - emptyListEquivalent :: Assertion - emptyListEquivalent = do - let input = - [ Right (1 :: Int) - , Left timedOut - , Left authentication - , Right 2 - ] - viaExcept <- runStream (skipNonFatalExcept []) input - viaPlain <- runStream skipNonFatal input - viaExcept @?= viaPlain diff --git a/test/Kafka/Streamly/StreamTest.hs b/test/Kafka/Streamly/StreamTest.hs new file mode 100644 --- /dev/null +++ b/test/Kafka/Streamly/StreamTest.hs @@ -0,0 +1,204 @@ +module Kafka.Streamly.StreamTest (tests) where + +import Kafka.Consumer (KafkaError (..), RdKafkaRespErrT (..)) +import Kafka.Streamly.Stream ( + isFatal, + isPartitionEOF, + isPollTimeout, + skipNonFatal, + skipNonFatalExcept, + ) +import Streamly.Data.Fold qualified as Fold +import Streamly.Data.Stream (Stream) +import Streamly.Data.Stream qualified as Stream +import Test.Tasty (TestTree, testGroup) +import Test.Tasty.HUnit (Assertion, testCase, (@?=)) + +tests :: TestTree +tests = + testGroup + "Stream" + [ testGroup "isFatal" isFatalTests + , testGroup "isPollTimeout" isPollTimeoutTests + , testGroup "isPartitionEOF" isPartitionEOFTests + , testGroup "skipNonFatal" skipNonFatalTests + , testGroup "skipNonFatalExcept" skipNonFatalExceptTests + ] + +------------------------------------------------------------------------------- +-- isFatal +------------------------------------------------------------------------------- + +fatalErrors :: [(String, KafkaError)] +fatalErrors = + [ ("KafkaUnknownConfigurationKey", KafkaUnknownConfigurationKey "") + , ("KafkaInvalidConfigurationValue", KafkaInvalidConfigurationValue "") + , ("KafkaBadConfiguration", KafkaBadConfiguration) + , ("KafkaBadSpecification", KafkaBadSpecification "") + , ("RdKafkaRespErrDestroy", KafkaResponseError RdKafkaRespErrDestroy) + , ("RdKafkaRespErrFail", KafkaResponseError RdKafkaRespErrFail) + , ("RdKafkaRespErrInvalidArg", KafkaResponseError RdKafkaRespErrInvalidArg) + , ("RdKafkaRespErrSsl", KafkaResponseError RdKafkaRespErrSsl) + , ("RdKafkaRespErrUnknownProtocol", KafkaResponseError RdKafkaRespErrUnknownProtocol) + , ("RdKafkaRespErrNotImplemented", KafkaResponseError RdKafkaRespErrNotImplemented) + , ("RdKafkaRespErrAuthentication", KafkaResponseError RdKafkaRespErrAuthentication) + , ("RdKafkaRespErrInconsistentGroupProtocol", KafkaResponseError RdKafkaRespErrInconsistentGroupProtocol) + , ("RdKafkaRespErrTopicAuthorizationFailed", KafkaResponseError RdKafkaRespErrTopicAuthorizationFailed) + , ("RdKafkaRespErrGroupAuthorizationFailed", KafkaResponseError RdKafkaRespErrGroupAuthorizationFailed) + , ("RdKafkaRespErrClusterAuthorizationFailed", KafkaResponseError RdKafkaRespErrClusterAuthorizationFailed) + , ("RdKafkaRespErrUnsupportedSaslMechanism", KafkaResponseError RdKafkaRespErrUnsupportedSaslMechanism) + , ("RdKafkaRespErrIllegalSaslState", KafkaResponseError RdKafkaRespErrIllegalSaslState) + , ("RdKafkaRespErrUnsupportedVersion", KafkaResponseError RdKafkaRespErrUnsupportedVersion) + ] + +nonFatalErrors :: [(String, KafkaError)] +nonFatalErrors = + [ ("RdKafkaRespErrTimedOut", KafkaResponseError RdKafkaRespErrTimedOut) + , ("RdKafkaRespErrPartitionEof", KafkaResponseError RdKafkaRespErrPartitionEof) + , ("RdKafkaRespErrNoError", KafkaResponseError RdKafkaRespErrNoError) + ] + +isFatalTests :: [TestTree] +isFatalTests = + [ testCase ("fatal: " <> label) (isFatal err @?= True) + | (label, err) <- fatalErrors + ] + <> [ testCase ("non-fatal: " <> label) (isFatal err @?= False) + | (label, err) <- nonFatalErrors + ] + +------------------------------------------------------------------------------- +-- isPollTimeout +------------------------------------------------------------------------------- + +isPollTimeoutTests :: [TestTree] +isPollTimeoutTests = + [ testCase "matches RdKafkaRespErrTimedOut" $ + isPollTimeout (KafkaResponseError RdKafkaRespErrTimedOut) @?= True + , testCase "rejects RdKafkaRespErrPartitionEof" $ + isPollTimeout (KafkaResponseError RdKafkaRespErrPartitionEof) @?= False + , testCase "rejects RdKafkaRespErrAuthentication" $ + isPollTimeout (KafkaResponseError RdKafkaRespErrAuthentication) @?= False + , testCase "rejects KafkaBadConfiguration" $ + isPollTimeout KafkaBadConfiguration @?= False + ] + +------------------------------------------------------------------------------- +-- isPartitionEOF +------------------------------------------------------------------------------- + +isPartitionEOFTests :: [TestTree] +isPartitionEOFTests = + [ testCase "matches RdKafkaRespErrPartitionEof" $ + isPartitionEOF (KafkaResponseError RdKafkaRespErrPartitionEof) @?= True + , testCase "rejects RdKafkaRespErrTimedOut" $ + isPartitionEOF (KafkaResponseError RdKafkaRespErrTimedOut) @?= False + , testCase "rejects RdKafkaRespErrAuthentication" $ + isPartitionEOF (KafkaResponseError RdKafkaRespErrAuthentication) @?= False + , testCase "rejects KafkaBadConfiguration" $ + isPartitionEOF KafkaBadConfiguration @?= False + ] + +------------------------------------------------------------------------------- +-- skipNonFatal / skipNonFatalExcept +------------------------------------------------------------------------------- + +runStream :: + (Stream IO a -> Stream IO b) -> + [a] -> + IO [b] +runStream f xs = Stream.fold Fold.toList (f (Stream.fromList xs)) + +timedOut, partitionEof, authentication :: KafkaError +timedOut = KafkaResponseError RdKafkaRespErrTimedOut +partitionEof = KafkaResponseError RdKafkaRespErrPartitionEof +authentication = KafkaResponseError RdKafkaRespErrAuthentication + +skipNonFatalTests :: [TestTree] +skipNonFatalTests = + [ testCase "drops poll-timeout, keeps authentication, keeps all Rights" dropsAndKeeps + , testCase "passes all Rights through unchanged" passesAllRights + , testCase "drops non-fatal Lefts even with empty extension list" dropsNonFatal + , testCase "drops partition-EOF (non-fatal)" dropsPartitionEof + ] + where + dropsAndKeeps :: Assertion + dropsAndKeeps = do + let input = + [ Right (1 :: Int) + , Left timedOut + , Right 2 + , Left authentication + , Right 3 + ] + result <- runStream skipNonFatal input + result @?= [Right 1, Right 2, Left authentication, Right 3] + + passesAllRights :: Assertion + passesAllRights = do + let input = [Right (n :: Int) | n <- [1 .. 5]] + result <- runStream skipNonFatal input + result @?= input + + dropsNonFatal :: Assertion + dropsNonFatal = do + let input :: [Either KafkaError Int] + input = + [ Left timedOut + , Left partitionEof + , Left (KafkaResponseError RdKafkaRespErrNoError) + ] + result <- runStream skipNonFatal input + result @?= [] + + dropsPartitionEof :: Assertion + dropsPartitionEof = do + let input = + [ Right (1 :: Int) + , Left partitionEof + , Right 2 + ] + result <- runStream skipNonFatal input + result @?= [Right 1, Right 2] + +skipNonFatalExceptTests :: [TestTree] +skipNonFatalExceptTests = + [ testCase "keeps poll-timeout when isPollTimeout is in the list" keepsTimeouts + , testCase "keeps partition-EOF when isPartitionEOF is in the list" keepsPartitionEof + , testCase "empty predicate list behaves like skipNonFatal" emptyListEquivalent + ] + where + keepsTimeouts :: Assertion + keepsTimeouts = do + let input = + [ Right (1 :: Int) + , Left timedOut + , Right 2 + , Left authentication + , Right 3 + ] + result <- runStream (skipNonFatalExcept [isPollTimeout]) input + result @?= [Right 1, Left timedOut, Right 2, Left authentication, Right 3] + + keepsPartitionEof :: Assertion + keepsPartitionEof = do + let input = + [ Right (1 :: Int) + , Left partitionEof + , Left timedOut + , Right 2 + ] + result <- runStream (skipNonFatalExcept [isPartitionEOF]) input + result @?= [Right 1, Left partitionEof, Right 2] + + emptyListEquivalent :: Assertion + emptyListEquivalent = do + let input = + [ Right (1 :: Int) + , Left timedOut + , Left authentication + , Right 2 + ] + viaExcept <- runStream (skipNonFatalExcept []) input + viaPlain <- runStream skipNonFatal input + viaExcept @?= viaPlain diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -1,7 +1,7 @@ module Main (main) where import Kafka.Streamly.CombinatorsTest qualified as CombinatorsTest -import Kafka.Streamly.SourceTest qualified as SourceTest +import Kafka.Streamly.StreamTest qualified as StreamTest import Test.Tasty (TestTree, defaultMain, testGroup) main :: IO () @@ -11,6 +11,6 @@ tests = testGroup "hw-kafka-streamly" - [ SourceTest.tests + [ StreamTest.tests , CombinatorsTest.tests ]