hw-kafka-client 3.0.0 → 5.3.0
raw patch · 27 files changed
Files
- README.md +101/−62
- example/ConsumerExample.hs +4/−5
- example/ProducerExample.hs +43/−14
- hw-kafka-client.cabal +116/−113
- src/Kafka/Callbacks.hs +44/−12
- src/Kafka/Consumer.hs +125/−17
- src/Kafka/Consumer/Callbacks.hs +19/−12
- src/Kafka/Consumer/ConsumerProperties.hs +48/−28
- src/Kafka/Consumer/Convert.hs +6/−3
- src/Kafka/Consumer/Subscription.hs +19/−0
- src/Kafka/Consumer/Types.hs +44/−16
- src/Kafka/Dump.hs +5/−0
- src/Kafka/Internal/RdKafka.chs +222/−28
- src/Kafka/Internal/Setup.hs +6/−0
- src/Kafka/Internal/Shared.hs +31/−4
- src/Kafka/Metadata.hs +8/−4
- src/Kafka/Producer.hs +120/−84
- src/Kafka/Producer/Callbacks.hs +51/−26
- src/Kafka/Producer/Convert.hs +16/−2
- src/Kafka/Producer/ProducerProperties.hs +36/−9
- src/Kafka/Producer/Types.hs +27/−5
- src/Kafka/Transaction.hs +132/−0
- src/Kafka/Types.hs +55/−10
- tests-it/Kafka/IntegrationSpec.hs +68/−25
- tests-it/Kafka/TestEnv.hs +2/−2
- tests/Kafka/Consumer/ConsumerRecordMapSpec.hs +2/−1
- tests/Kafka/Consumer/ConsumerRecordTraverseSpec.hs +2/−1
README.md view
@@ -1,53 +1,53 @@ # hw-kafka-client+ [](https://circleci.com/gh/haskell-works/hw-kafka-client) Kafka bindings for Haskell backed by the [librdkafka C module](https://github.com/edenhill/librdkafka). -## Credits-This project is inspired by [Haskakafka](https://github.com/cosbynator/haskakafka)-which unfortunately doesn't seem to be actively maintained.- ## Ecosystem-HaskellWorks Kafka ecosystem is described here: https://github.com/haskell-works/hw-kafka -# Consumer+HaskellWorks Kafka ecosystem is described here: <https://github.com/haskell-works/hw-kafka>++## Consumer+ High level consumers are supported by `librdkafka` starting from version 0.9. High-level consumers provide an abstraction for consuming messages from multiple-partitions and topics. They are also address scalability (up to a number of partitions)+partitions and topics. They also address scalability (up to a number of partitions) by providing automatic rebalancing functionality. When a new consumer joins a consumer-group the set of consumers attempt to "rebalance" the load to assign partitions to each consumer.+group the set of consumers attempts to "rebalance" the load to assign partitions to each consumer. -### Example:-```-$ stack build --flag hw-kafka-client:examples+### Consumer example++See [Running integration tests locally](#running-integration-tests-locally) to learn how to configure a local environment.++```bash+cabal build --flag examples ``` or -```-$ stack build --exec kafka-client-example --flag hw-kafka-client:examples+```bash+cabal run kafka-client-example --flag examples ``` A working consumer example can be found here: [ConsumerExample.hs](example/ConsumerExample.hs)</br> To run an example please compile with the `examples` flag. -```Haskell+```haskell import Control.Exception (bracket)-import Data.Monoid ((<>))-import Kafka import Kafka.Consumer -- Global consumer properties consumerProps :: ConsumerProperties-consumerProps = brokersList [BrokerAddress "localhost:9092"]- <> groupId (ConsumerGroupId "consumer_example_group")+consumerProps = brokersList ["localhost:9092"]+ <> groupId "consumer_example_group" <> noAutoCommit <> logLevel KafkaLogInfo -- Subscription to topics consumerSub :: Subscription-consumerSub = topics [TopicName "kafka-client-example-topic"]+consumerSub = topics ["kafka-client-example-topic"] <> offsetReset Earliest -- Running an example@@ -58,29 +58,30 @@ where mkConsumer = newConsumer consumerProps consumerSub clConsumer (Left err) = return (Left err)- clConsumer (Right kc) = (maybe (Right ()) Left) <$> closeConsumer kc+ clConsumer (Right kc) = maybe (Right ()) Left <$> closeConsumer kc runHandler (Left err) = return (Left err) runHandler (Right kc) = processMessages kc ------------------------------------------------------------------- processMessages :: KafkaConsumer -> IO (Either KafkaError ()) processMessages kafka = do- mapM_ (\_ -> do- msg1 <- pollMessage kafka (Timeout 1000)- putStrLn $ "Message: " <> show msg1- err <- commitAllOffsets OffsetCommit kafka- putStrLn $ "Offsets: " <> maybe "Committed." show err- ) [0 .. 10]+ replicateM_ 10 $ do+ msg <- pollMessage kafka (Timeout 1000)+ putStrLn $ "Message: " <> show msg+ err <- commitAllOffsets OffsetCommit kafka+ putStrLn $ "Offsets: " <> maybe "Committed." show err return $ Right () ``` -# Producer+## Producer+ `kafka-client` producer supports sending messages to multiple topics. Target topic name is a part of each message that is to be sent by `produceMessage`. A working producer example can be found here: [ProducerExample.hs](example/ProducerExample.hs) ### Delivery reports+ Kafka Producer maintains its own internal queue for outgoing messages. Calling `produceMessage` does not mean that the message is actually written to Kafka, it only means that the message is put to that outgoing queue and that the producer will (eventually) push it to Kafka.@@ -94,30 +95,29 @@ It is possible to configure `hw-kafka-client` to set an infinite message timeout so the message is never dropped from the queue: -```+```haskell producerProps :: ProducerProperties-producerProps = brokersList [BrokerAddress "localhost:9092"]- <> sendTimeout (Timeout 0) -- for librdkafka "0" means "infinite".+producerProps = brokersList ["localhost:9092"]+ <> sendTimeout (Timeout 0) -- for librdkafka "0" means "infinite" (see https://github.com/edenhill/librdkafka/issues/2015) ``` *Delivery reports* provide the way to detect when producer experiences problems sending messages to Kafka. Currently `hw-kafka-client` only supports delivery error callbacks:-```++```haskell producerProps :: ProducerProperties-producerProps = brokersList [BrokerAddress "localhost:9092"]- <> setCallback (deliveryErrorsCallback print)+producerProps = brokersList ["localhost:9092"]+ <> setCallback (deliveryCallback print) ```+ In the example above when the producer cannot deliver the message to Kafka, the error will be printed (and the message will be dropped). -When `sendTimeout` is not configured to `Timeout 0` (infinite), no error callbacks will be delivered.-This is because no message will ever be timing out for sending.--### Example+### Producer example -```Haskell+```haskell {-# LANGUAGE OverloadedStrings #-} import Control.Exception (bracket) import Control.Monad (forM_)@@ -126,12 +126,12 @@ -- Global producer properties producerProps :: ProducerProperties-producerProps = brokersList [BrokerAddress "localhost:9092"]+producerProps = brokersList ["localhost:9092"] <> logLevel KafkaLogDebug -- Topic to send messages to targetTopic :: TopicName-targetTopic = TopicName "kafka-client-example-topic"+targetTopic = "kafka-client-example-topic" -- Run an example runProducerExample :: IO ()@@ -163,39 +163,78 @@ } ``` -# Installation+### Synchronous sending of messages -## Installing librdkafka+Because of the asynchronous nature of librdkafka, there is no API to provide+synchronous production of messages. It is, however, possible to combine the+delivery reports feature with that of callbacks. This can be done using the+`Kafka.Producer.produceMessage'` function. -Although `librdkafka` is available on many platforms, most of-the distribution packages are too old to support `kafka-client`.-As such, we suggest you install from the source:+```haskell+produceMessage' :: MonadIO m+ => KafkaProducer+ -> ProducerRecord+ -> (DeliveryReport -> IO ())+ -> m (Either ImmediateError ())+``` - git clone https://github.com/edenhill/librdkafka- cd librdkafka- ./configure- make && make install+Using this function, you can provide a callback which will be invoked upon the+produced message's delivery report. With a little help of `MVar`s or similar,+you can in fact, create a synchronous-like interface. -Sometimes it is helpful to specify openssl includes explicitly:+```haskell+sendMessageSync :: MonadIO m+ => KafkaProducer+ -> ProducerRecord+ -> m (Either KafkaError Offset)+sendMessageSync producer record = liftIO $ do+ -- Create an empty MVar:+ var <- newEmptyMVar - LDFLAGS=-L/usr/local/opt/openssl/lib CPPFLAGS=-I/usr/local/opt/openssl/include ./configure+ -- Produce the message and use the callback to put the delivery report in the+ -- MVar:+ res <- produceMessage' producer record (putMVar var) -If you are using Stack with Nix, don't forget to declare `rdkafka` as extra package:-```yaml-# stack.yaml-nix:- enable: true- packages:- - rdkafka+ case res of+ Left (ImmediateError err) ->+ pure (Left err)+ Right () -> do+ -- Flush producer queue to make sure you don't get stuck waiting for the+ -- message to send:+ flushProducer producer++ -- Wait for the message's delivery report and map accordingly:+ takeMVar var >>= return . \case+ DeliverySuccess _ offset -> Right offset+ DeliveryFailure _ err -> Left err+ NoMessageError err -> Left err ``` -## Installing Kafka+_Note:_ this is a semi-naive solution as this waits forever (or until+librdkafka times out). You should make sure that your configuration reflects+the behavior you want out of this functionality. -The full Kafka guide is at http://kafka.apache.org/documentation.html#quickstart+## Running integration tests locally -Alternatively `docker-compose` can be used to run Kafka locally inside a Docker container.-To run Kafka inside Docker:+[shell.nix](./shell.nix) can be used to provide a working environment that is enough to build and test `hw-kafka-client`. +To be able to run tests locally, `$KAFKA_TEST_BROKER` environment variable is expected to be set (use [shell.nix](./shell.nix) or export manually).++`$KAFKA_TEST_BROKER` should contain an IP address of an accessible Kafka broker that will be used to run integration tests against.++With [Docker Compose](./docker-compose.yml) this variable is used to configure Kafka broker to listen on this address:+ ``` $ docker-compose up ```++After that, integration tests can switched on with using 'it' flag:++```+$ cabal test --test-show-details=direct --flag it+```++## Credits++This project is inspired by [Haskakafka](https://github.com/cosbynator/haskakafka)+which unfortunately doesn't seem to be actively maintained.
example/ConsumerExample.hs view
@@ -6,14 +6,13 @@ import Control.Arrow ((&&&)) import Control.Exception (bracket)-import Data.Monoid ((<>)) import Kafka.Consumer import Data.Text (Text) -- Global consumer properties consumerProps :: ConsumerProperties-consumerProps = brokersList [BrokerAddress "localhost:9092"]- <> groupId (ConsumerGroupId "consumer_example_group")+consumerProps = brokersList ["localhost:9092"]+ <> groupId "consumer_example_group" <> noAutoCommit <> setCallback (rebalanceCallback printingRebalanceCallback) <> setCallback (offsetCommitCallback printingOffsetCallback)@@ -21,7 +20,7 @@ -- Subscription to topics consumerSub :: Subscription-consumerSub = topics [TopicName "kafka-client-example-topic"]+consumerSub = topics ["kafka-client-example-topic"] <> offsetReset Earliest -- Running an example@@ -33,7 +32,7 @@ where mkConsumer = newConsumer consumerProps consumerSub clConsumer (Left err) = return (Left err)- clConsumer (Right kc) = (maybe (Right ()) Left) <$> closeConsumer kc+ clConsumer (Right kc) = maybe (Right ()) Left <$> closeConsumer kc runHandler (Left err) = return (Left err) runHandler (Right kc) = processMessages kc
example/ProducerExample.hs view
@@ -1,26 +1,29 @@ {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE LambdaCase #-} module ProducerExample where -import Control.Exception (bracket)-import Control.Monad (forM_)-import Data.ByteString (ByteString)-import Data.ByteString.Char8 (pack)-import Data.Monoid+import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)+import Control.Exception (bracket)+import Control.Monad (forM_)+import Control.Monad.IO.Class (MonadIO(..))+import Data.ByteString (ByteString)+import Data.ByteString.Char8 (pack)+import Kafka.Consumer (Offset) import Kafka.Producer-import Data.Text (Text)+import Data.Text (Text) -- Global producer properties producerProps :: ProducerProperties-producerProps = brokersList [BrokerAddress "localhost:9092"]+producerProps = brokersList ["localhost:9092"] <> sendTimeout (Timeout 10000) <> setCallback (deliveryCallback print) <> logLevel KafkaLogDebug -- Topic to send messages to targetTopic :: TopicName-targetTopic = TopicName "kafka-client-example-topic"+targetTopic = "kafka-client-example-topic" mkMessage :: Maybe ByteString -> Maybe ByteString -> ProducerRecord mkMessage k v = ProducerRecord@@ -28,6 +31,7 @@ , prPartition = UnassignedPartition , prKey = k , prValue = v+ , prHeaders = mempty } -- Run an example@@ -46,7 +50,7 @@ putStrLn "Producer is ready, send your messages!" msg1 <- getLine - err1 <- produceMessage prod (mkMessage Nothing (Just $ pack msg1))+ err1 <- produceMessage prod (mkMessage (Just "zero") (Just $ pack msg1)) forM_ err1 print putStrLn "One more time!"@@ -59,13 +63,38 @@ msg3 <- getLine err3 <- produceMessage prod (mkMessage (Just "key3") (Just $ pack msg3)) - -- errs <- produceMessageBatch prod- -- [ mkMessage (Just "b-1") (Just "batch-1")- -- , mkMessage (Just "b-2") (Just "batch-2")- -- , mkMessage Nothing (Just "batch-3")- -- ]+ err4 <- produceMessage prod ((mkMessage (Just "key4") (Just $ pack msg3)) { prHeaders = headersFromList [("fancy", "header")]}) -- forM_ errs (print . snd) putStrLn "Thank you." return $ Right ()++-- | An example for sending messages synchronously using the 'produceMessage''+-- function+--+sendMessageSync :: MonadIO m+ => KafkaProducer+ -> ProducerRecord+ -> m (Either KafkaError Offset)+sendMessageSync producer record = liftIO $ do+ -- Create an empty MVar:+ var <- newEmptyMVar++ -- Produce the message and use the callback to put the delivery report in the+ -- MVar:+ res <- produceMessage' producer record (putMVar var)++ case res of+ Left (ImmediateError err) ->+ pure (Left err)+ Right () -> do+ -- Flush producer queue to make sure you don't get stuck waiting for the+ -- message to send:+ flushProducer producer++ -- Wait for the message's delivery report and map accordingly:+ takeMVar var >>= return . \case+ DeliverySuccess _ offset -> Right offset+ DeliveryFailure _ err -> Left err+ NoMessageError err -> Left err
hw-kafka-client.cabal view
@@ -1,27 +1,28 @@-name: hw-kafka-client-version: 3.0.0-synopsis: Kafka bindings for Haskell-description: Apache Kafka bindings backed by the librdkafka C library.- .- Features include:- .- * Consumer groups: auto-rebalancing consumers- .- * Keyed and keyless messages producing/consuming- .- * Batch producing messages-category: Database-homepage: https://github.com/haskell-works/hw-kafka-client-bug-reports: https://github.com/haskell-works/hw-kafka-client/issues-author: Alexey Raga <alexey.raga@gmail.com>-maintainer: Alexey Raga <alexey.raga@gmail.com>-license: MIT-license-file: LICENSE-build-type: Simple-cabal-version: >= 1.10-extra-source-files:- README.md+cabal-version: 2.2 +name: hw-kafka-client+version: 5.3.0+synopsis: Kafka bindings for Haskell+description: Apache Kafka bindings backed by the librdkafka C library.+ .+ Features include:+ .+ * Consumer groups: auto-rebalancing consumers+ .+ * Keyed and keyless messages producing/consuming+ .+ * Batch producing messages+category: Database+homepage: https://github.com/haskell-works/hw-kafka-client+bug-reports: https://github.com/haskell-works/hw-kafka-client/issues+author: Alexey Raga <alexey.raga@gmail.com>+maintainer: Alexey Raga <alexey.raga@gmail.com>+license: MIT+license-file: LICENSE+tested-with: GHC == 8.10.2, GHC == 8.8.3, GHC == 8.6.5, GHC == 8.4.4+build-type: Simple+extra-source-files: README.md+ source-repository head type: git location: git://github.com/haskell-works/hw-kafka-client.git@@ -31,103 +32,105 @@ manual: True default: False +flag it+ description: Run integration tests+ manual: True+ default: False+ library- hs-source-dirs: src- ghc-options:- -Wall- -Wcompat- -Wincomplete-record-updates- -Wincomplete-uni-patterns- -Wredundant-constraints- extra-libraries: rdkafka- build-depends:- base >=4.6 && <5- , bifunctors- , bytestring- , containers- , text- , transformers- , unix- build-tools: c2hs+ hs-source-dirs: src+ ghc-options: -Wall+ -Wcompat+ -Wincomplete-record-updates+ -Wincomplete-uni-patterns+ -Wredundant-constraints+ extra-libraries: rdkafka+ build-depends: base >=4.6 && <5+ , bifunctors+ , bytestring+ , containers+ , text+ , transformers+ , unix+ build-tool-depends: c2hs:c2hs if impl(ghc <8.0)- build-depends:- semigroups- exposed-modules:- Kafka.Consumer- Kafka.Consumer.ConsumerProperties- Kafka.Consumer.Subscription- Kafka.Consumer.Types- Kafka.Dump- Kafka.Metadata- Kafka.Producer- Kafka.Producer.ProducerProperties- Kafka.Producer.Types- Kafka.Types- other-modules:- Kafka.Callbacks- Kafka.Consumer.Callbacks- Kafka.Consumer.Convert- Kafka.Internal.RdKafka- Kafka.Internal.Setup- Kafka.Internal.Shared- Kafka.Producer.Callbacks- Kafka.Producer.Convert- default-language: Haskell2010+ build-depends: semigroups+ exposed-modules: Kafka.Consumer+ Kafka.Consumer.ConsumerProperties+ Kafka.Consumer.Subscription+ Kafka.Consumer.Types+ Kafka.Dump+ Kafka.Metadata+ Kafka.Producer+ Kafka.Producer.ProducerProperties+ Kafka.Producer.Types+ Kafka.Transaction+ Kafka.Types+ other-modules: Kafka.Callbacks+ Kafka.Consumer.Callbacks+ Kafka.Consumer.Convert+ Kafka.Internal.RdKafka+ Kafka.Internal.Setup+ Kafka.Internal.Shared+ Kafka.Producer.Callbacks+ Kafka.Producer.Convert+ default-language: Haskell2010 executable kafka-client-example- main-is: Main.hs- hs-source-dirs: example- ghc-options: -threaded -rtsopts- build-depends:- base >=4.6 && <5- , bytestring- , hw-kafka-client- , text+ main-is: Main.hs+ hs-source-dirs: example+ ghc-options: -threaded -rtsopts+ build-depends: base >=4.6 && <5+ , bytestring+ , hw-kafka-client+ , text if !(flag(examples))- buildable: False+ buildable: False other-modules:- ConsumerExample- ProducerExample- default-language: Haskell2010+ ConsumerExample+ ProducerExample+ default-language: Haskell2010 test-suite integration-tests- type: exitcode-stdio-1.0- main-is: Spec.hs- hs-source-dirs: tests-it- ghc-options: -Wall -threaded- build-depends:- base >=4.6 && <5- , bifunctors- , bytestring- , containers- , either- , hspec- , hw-kafka-client- , monad-loops- , random- , text- , transformers- other-modules:- Kafka.IntegrationSpec- Kafka.TestEnv- default-language: Haskell2010+ type: exitcode-stdio-1.0+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ main-is: Spec.hs+ hs-source-dirs: tests-it+ ghc-options: -Wall -threaded+ build-depends: base >=4.6 && <5+ , bifunctors+ , bytestring+ , containers+ , either+ , hspec+ , hw-kafka-client+ , monad-loops+ , random+ , text+ , transformers+ build-tool-depends: hspec-discover:hspec-discover+ if !(flag(it))+ buildable: False+ other-modules: Kafka.IntegrationSpec+ Kafka.TestEnv+ default-language: Haskell2010 test-suite tests- type: exitcode-stdio-1.0- main-is: Spec.hs- hs-source-dirs: tests- ghc-options: -Wall -threaded- build-depends:- base >=4.6 && <5- , bifunctors- , bytestring- , containers- , either- , hspec- , hw-kafka-client- , text- , monad-loops- other-modules:- Kafka.Consumer.ConsumerRecordMapSpec- Kafka.Consumer.ConsumerRecordTraverseSpec- default-language: Haskell2010+ type: exitcode-stdio-1.0+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ main-is: Spec.hs+ hs-source-dirs: tests+ ghc-options: -Wall -threaded+ build-depends: base >=4.6 && <5+ , bifunctors+ , bytestring+ , containers+ , either+ , hspec+ , hw-kafka-client+ , text+ , monad-loops+ build-tool-depends: hspec-discover:hspec-discover+ other-modules: Kafka.Consumer.ConsumerRecordMapSpec+ Kafka.Consumer.ConsumerRecordTraverseSpec+ default-language: Haskell2010
src/Kafka/Callbacks.hs view
@@ -2,24 +2,56 @@ ( errorCallback , logCallback , statsCallback+, Callback ) where +import Data.ByteString (ByteString) import Kafka.Internal.RdKafka (rdKafkaConfSetErrorCb, rdKafkaConfSetLogCb, rdKafkaConfSetStatsCb)-import Kafka.Internal.Setup (HasKafkaConf(..), getRdKafkaConf)-import Kafka.Types (KafkaError(..))+import Kafka.Internal.Setup (getRdKafkaConf, Callback(..))+import Kafka.Types (KafkaError(..), KafkaLogLevel(..)) -errorCallback :: HasKafkaConf k => (KafkaError -> String -> IO ()) -> k -> IO ()-errorCallback callback k =+-- | Add a callback for errors.+--+-- ==== __Examples__+--+-- Basic usage:+--+-- > 'setCallback' ('errorCallback' myErrorCallback)+-- >+-- > myErrorCallback :: 'KafkaError' -> String -> IO ()+-- > myErrorCallback kafkaError message = print $ show kafkaError <> "|" <> message+errorCallback :: (KafkaError -> String -> IO ()) -> Callback+errorCallback callback = let realCb _ err = callback (KafkaResponseError err)- in rdKafkaConfSetErrorCb (getRdKafkaConf k) realCb+ in Callback $ \k -> rdKafkaConfSetErrorCb (getRdKafkaConf k) realCb -logCallback :: HasKafkaConf k => (Int -> String -> String -> IO ()) -> k -> IO ()-logCallback callback k =- let realCb _ = callback- in rdKafkaConfSetLogCb (getRdKafkaConf k) realCb+-- | Add a callback for logs.+--+-- ==== __Examples__+--+-- Basic usage:+--+-- > 'setCallback' ('logCallback' myLogCallback)+-- >+-- > myLogCallback :: 'KafkaLogLevel' -> String -> String -> IO ()+-- > myLogCallback level facility message = print $ show level <> "|" <> facility <> "|" <> message+logCallback :: (KafkaLogLevel -> String -> String -> IO ()) -> Callback+logCallback callback =+ let realCb _ = callback . toEnum+ in Callback $ \k -> rdKafkaConfSetLogCb (getRdKafkaConf k) realCb -statsCallback :: HasKafkaConf k => (String -> IO ()) -> k -> IO ()-statsCallback callback k =+-- | Add a callback for stats. The passed ByteString contains an UTF-8 encoded JSON document and can e.g. be parsed using Data.Aeson.decodeStrict. For more information about the content of the JSON document see <https://github.com/edenhill/librdkafka/blob/master/STATISTICS.md>.+--+-- ==== __Examples__+--+-- Basic usage:+--+-- > 'setCallback' ('statsCallback' myStatsCallback)+-- >+-- > myStatsCallback :: String -> IO ()+-- > myStatsCallback stats = print $ show stats+statsCallback :: (ByteString -> IO ()) -> Callback+statsCallback callback = let realCb _ = callback- in rdKafkaConfSetStatsCb (getRdKafkaConf k) realCb+ in Callback $ \k -> rdKafkaConfSetStatsCb (getRdKafkaConf k) realCb
src/Kafka/Consumer.hs view
@@ -1,20 +1,68 @@ {-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TupleSections #-}++-----------------------------------------------------------------------------+-- |+-- Module to consume messages from Kafka topics.+-- +-- Here's an example of code to consume messages from a topic:+-- +-- @+-- import Control.Exception (bracket)+-- import Control.Monad (replicateM_)+-- import Kafka.Consumer+-- +-- -- Global consumer properties+-- consumerProps :: 'ConsumerProperties'+-- consumerProps = 'brokersList' ["localhost:9092"]+-- <> 'groupId' ('ConsumerGroupId' "consumer_example_group")+-- <> 'noAutoCommit'+-- <> 'logLevel' 'KafkaLogInfo'+-- +-- -- Subscription to topics+-- consumerSub :: 'Subscription'+-- consumerSub = 'topics' ['TopicName' "kafka-client-example-topic"]+-- <> 'offsetReset' 'Earliest'+-- +-- -- Running an example+-- runConsumerExample :: IO ()+-- runConsumerExample = do+-- res <- bracket mkConsumer clConsumer runHandler+-- print res+-- where+-- mkConsumer = 'newConsumer' consumerProps consumerSub+-- clConsumer (Left err) = pure (Left err)+-- clConsumer (Right kc) = (maybe (Right ()) Left) \<$\> 'closeConsumer' kc+-- runHandler (Left err) = pure (Left err)+-- runHandler (Right kc) = processMessages kc+-- +-- -- Example polling 10 times before stopping+-- processMessages :: 'KafkaConsumer' -> IO (Either 'KafkaError' ())+-- processMessages kafka = do+-- replicateM_ 10 $ do+-- msg <- 'pollMessage' kafka ('Timeout' 1000)+-- putStrLn $ "Message: " <> show msg+-- err <- 'commitAllOffsets' 'OffsetCommit' kafka+-- putStrLn $ "Offsets: " <> maybe "Committed." show err+-- pure $ Right ()+-- @+----------------------------------------------------------------------------- module Kafka.Consumer-( module X+( KafkaConsumer+, module X , runConsumer , newConsumer-, assignment, subscription+, assign, assignment, subscription , pausePartitions, resumePartitions-, committed, position, seek+, committed, position, seek, seekPartitions , pollMessage, pollConsumerEvents , pollMessageBatch , commitOffsetMessage, commitAllOffsets, commitPartitionsOffsets , storeOffsets, storeOffsetMessage+, rewindConsumer , closeConsumer -- ReExport Types-, KafkaConsumer , RdKafkaRespErrT (..) ) where@@ -22,23 +70,22 @@ import Control.Arrow (left, (&&&)) import Control.Concurrent (forkIO, modifyMVar, rtsSupportsBoundThreads, withMVar) import Control.Exception (bracket)-import Control.Monad (forM_, mapM_, void, when)+import Control.Monad (forM_, void, when) import Control.Monad.IO.Class (MonadIO (liftIO)) import Control.Monad.Trans.Except (ExceptT (ExceptT), runExceptT) import Data.Bifunctor (bimap, first) import qualified Data.ByteString as BS import Data.IORef (readIORef, writeIORef)-import qualified Data.Map as M+import qualified Data.Map as M hiding (map, foldr) import Data.Maybe (fromMaybe)-import Data.Monoid ((<>)) import Data.Set (Set) import qualified Data.Set as Set import qualified Data.Text as Text import Foreign hiding (void) import Kafka.Consumer.Convert (fromMessagePtr, fromNativeTopicPartitionList'', offsetCommitToBool, offsetToInt64, toMap, toNativeTopicPartitionList, toNativeTopicPartitionList', toNativeTopicPartitionListNoDispose, topicPartitionFromMessageForCommit) import Kafka.Consumer.Types (KafkaConsumer (..))-import Kafka.Internal.RdKafka (RdKafkaRespErrT (..), RdKafkaTopicPartitionListTPtr, RdKafkaTypeT (..), newRdKafkaT, newRdKafkaTopicPartitionListT, newRdKafkaTopicT, rdKafkaAssignment, rdKafkaCommit, rdKafkaCommitted, rdKafkaConfSetDefaultTopicConf, rdKafkaConsumeBatchQueue, rdKafkaConsumeQueue, rdKafkaConsumerClose, rdKafkaConsumerPoll, rdKafkaOffsetsStore, rdKafkaPausePartitions, rdKafkaPollSetConsumer, rdKafkaPosition, rdKafkaQueueDestroy, rdKafkaQueueNew, rdKafkaResumePartitions, rdKafkaSeek, rdKafkaSetLogLevel, rdKafkaSubscribe, rdKafkaSubscription, rdKafkaTopicConfDup, rdKafkaTopicPartitionListAdd)-import Kafka.Internal.Setup (CallbackPollStatus (..), Kafka (..), KafkaConf (..), KafkaProps (..), TopicConf (..), TopicProps (..), getKafkaConf, getRdKafka, kafkaConf, topicConf)+import Kafka.Internal.RdKafka (RdKafkaRespErrT (..), RdKafkaTopicPartitionListTPtr, RdKafkaTypeT (..), rdKafkaSeekPartitions, rdKafkaErrorDestroy, rdKafkaErrorCode, newRdKafkaT, newRdKafkaTopicPartitionListT, newRdKafkaTopicT, rdKafkaAssign, rdKafkaAssignment, rdKafkaCommit, rdKafkaCommitted, rdKafkaConfSetDefaultTopicConf, rdKafkaConsumeBatchQueue, rdKafkaConsumeQueue, rdKafkaConsumerClose, rdKafkaConsumerPoll, rdKafkaOffsetsStore, rdKafkaPausePartitions, rdKafkaPollSetConsumer, rdKafkaPosition, rdKafkaQueueDestroy, rdKafkaQueueNew, rdKafkaResumePartitions, rdKafkaSeek, rdKafkaSetLogLevel, rdKafkaSubscribe, rdKafkaSubscription, rdKafkaTopicConfDup, rdKafkaTopicPartitionListAdd)+import Kafka.Internal.Setup (CallbackPollStatus (..), Kafka (..), KafkaConf (..), KafkaProps (..), TopicConf (..), TopicProps (..), getKafkaConf, getRdKafka, kafkaConf, topicConf, Callback(..)) import Kafka.Internal.Shared (kafkaErrorToMaybe, maybeToLeft, rdKafkaErrorToEither) import Kafka.Consumer.ConsumerProperties as X@@ -48,7 +95,7 @@ -- | Runs high-level kafka consumer. -- A callback provided is expected to call 'pollMessage' when convenient.-{-# DEPRECATED runConsumer "Use newConsumer/closeConsumer instead" #-}+{-# DEPRECATED runConsumer "Use 'newConsumer'/'closeConsumer' instead" #-} runConsumer :: ConsumerProperties -> Subscription -> (KafkaConsumer -> IO (Either KafkaError a)) -- ^ A callback function to poll and handle messages@@ -64,6 +111,7 @@ runHandler (Left err) = return (Left err) runHandler (Right kc) = f kc +-- | Create a `KafkaConsumer`. This consumer must be correctly released using 'closeConsumer'. newConsumer :: MonadIO m => ConsumerProperties -> Subscription@@ -94,6 +142,7 @@ runConsumerLoop kafka (Just $ Timeout 100)) >> return (Right kafka) Just err -> closeConsumer kafka >> return (Left err) +-- | Polls a single message pollMessage :: MonadIO m => KafkaConsumer -> Timeout -- ^ the timeout, in milliseconds@@ -104,11 +153,11 @@ Nothing -> rdKafkaConsumerPoll (getRdKafka c) ms >>= fromMessagePtr Just q -> rdKafkaConsumeQueue q (fromIntegral ms) >>= fromMessagePtr --- | Polls up to BatchSize messages.+-- | Polls up to 'BatchSize' messages. -- Unlike 'pollMessage' this function does not return usual "timeout" errors. -- An empty batch is returned when there are no messages available. ----- This API is not available when 'userPolls' is set.+-- This API is not available when 'CallbackPollMode' is set to 'CallbackPollModeSync'. pollMessageBatch :: MonadIO m => KafkaConsumer -> Timeout@@ -118,8 +167,8 @@ pollConsumerEvents c Nothing mbq <- readIORef qr case mbq of- Nothing -> return [Left $ KafkaBadSpecification "userPolls is set when calling pollMessageBatch."]- Just q -> rdKafkaConsumeBatchQueue q ms b >>= traverse fromMessagePtr+ Nothing -> return [Left $ KafkaBadSpecification "Calling pollMessageBatch while CallbackPollMode is set to CallbackPollModeSync."]+ Just q -> whileNoCallbackRunning c $ rdKafkaConsumeBatchQueue q ms b >>= traverse fromMessagePtr -- | Commit message's offset on broker for the message's partition. commitOffsetMessage :: MonadIO m@@ -163,6 +212,13 @@ commitPartitionsOffsets o k ps = liftIO $ toNativeTopicPartitionList ps >>= commitOffsets o k +-- | Assigns the consumer to consume from the given topics, partitions,+-- and offsets.+assign :: MonadIO m => KafkaConsumer -> [TopicPartition] -> m (Maybe KafkaError)+assign (KafkaConsumer (Kafka k) _) ps = liftIO $ do+ tps <- toNativeTopicPartitionList ps+ kafkaErrorToMaybe . KafkaResponseError <$> rdKafkaAssign k tps+ -- | Returns current consumer's assignment assignment :: MonadIO m => KafkaConsumer -> m (Either KafkaError (M.Map TopicName [PartitionId])) assignment (KafkaConsumer (Kafka k) _) = liftIO $ do@@ -198,6 +254,8 @@ mapM_ (\(TopicName topicName, PartitionId partitionId) -> rdKafkaTopicPartitionListAdd pl (Text.unpack topicName) partitionId) ps KafkaResponseError <$> rdKafkaResumePartitions k pl +-- | Seek a particular offset for each provided 'TopicPartition'+{-# DEPRECATED seek "Use seekPartitions instead" #-} seek :: MonadIO m => KafkaConsumer -> Timeout -> [TopicPartition] -> m (Maybe KafkaError) seek (KafkaConsumer (Kafka k) _) (Timeout timeout) tps = liftIO $ either Just (const Nothing) <$> seekAll@@ -214,6 +272,14 @@ nt <- newRdKafkaTopicT k (Text.unpack tn) Nothing return $ bimap KafkaError (,tpPartition tp, tpOffset tp) (first Text.pack nt) +-- | Seek consumer for partitions in partitions to the per-partition+-- offset in the offset field of partitions.+seekPartitions :: MonadIO m => KafkaConsumer -> [TopicPartition] -> Timeout -> m (Maybe KafkaError)+seekPartitions (KafkaConsumer (Kafka k) _) ps (Timeout timeout) = liftIO $ do+ tps <- toNativeTopicPartitionList ps+ err <- bracket (rdKafkaSeekPartitions k tps timeout) rdKafkaErrorDestroy rdKafkaErrorCode+ pure $ either Just (const Nothing) $ rdKafkaErrorToEither err+ -- | Retrieve committed offsets for topics+partitions. committed :: MonadIO m => KafkaConsumer -> Timeout -> [(TopicName, PartitionId)] -> m (Either KafkaError [TopicPartition]) committed (KafkaConsumer (Kafka k) _) (Timeout timeout) tps = liftIO $ do@@ -237,7 +303,7 @@ -- -- Events will cause application provided callbacks to be called. ----- The \p timeout_ms argument specifies the maximum amount of time+-- The 'Timeout' argument specifies the maximum amount of time -- (in milliseconds) that the call will block waiting for events. -- -- This function is called on each 'pollMessage' and, if runtime allows@@ -253,6 +319,8 @@ void . withCallbackPollEnabled k $ pollConsumerEvents' k timeout -- | Closes the consumer.+-- +-- See 'newConsumer' closeConsumer :: MonadIO m => KafkaConsumer -> m (Maybe KafkaError) closeConsumer (KafkaConsumer (Kafka k) (KafkaConf _ qr statusVar)) = liftIO $ -- because closing the consumer will raise callbacks,@@ -264,16 +332,51 @@ readIORef qr >>= mapM_ rdKafkaQueueDestroy res <- kafkaErrorToMaybe . KafkaResponseError <$> rdKafkaConsumerClose k pure (CallbackPollDisabled, res)++-- | Rewind consumer's consume position to the last committed offsets for the current assignment.+-- NOTE: follows https://github.com/edenhill/librdkafka/blob/master/examples/transactions.c#L166+rewindConsumer :: MonadIO m + => KafkaConsumer + -> Timeout+ -> m (Maybe KafkaError)+rewindConsumer c to = liftIO $ do+ ret <- assignment c+ case ret of+ Left err -> pure $ Just err+ Right os -> do+ if M.size os == 0+ -- No current assignment to rewind+ then pure Nothing+ else do+ let tps = foldr (\(t, ps) acc -> map (t,) ps ++ acc) [] $ M.toList os+ ret' <- committed c to tps+ case ret' of+ Left err -> pure $ Just err+ Right ps -> do+ -- Seek to committed offset, or start of partition if no+ -- committed offset is available.+ let ps' = map checkOffsets ps+ seekPartitions c ps' to+ where+ checkOffsets :: TopicPartition -> TopicPartition+ checkOffsets tp + | isUncommitedOffset $ tpOffset tp + = tp { tpOffset = PartitionOffsetBeginning }+ | otherwise = tp ++ isUncommitedOffset :: PartitionOffset -> Bool+ isUncommitedOffset (PartitionOffset _) = False+ isUncommitedOffset _ = True ----------------------------------------------------------------------------- newConsumerConf :: ConsumerProperties -> IO KafkaConf newConsumerConf ConsumerProperties {cpProps = m, cpCallbacks = cbs} = do conf <- kafkaConf (KafkaProps m)- forM_ cbs (\setCb -> setCb conf)+ forM_ cbs (\(Callback setCb) -> setCb conf) return conf -- | Subscribes to a given list of topics. ----- Wildcard (regex) topics are supported by the librdkafka assignor:+-- Wildcard (regex) topics are supported by the /librdkafka/ assignor: -- any topic name in the topics list that is prefixed with @^@ will -- be regex-matched to the full list of topics in the cluster and matching -- topics will be added to the subscription list.@@ -313,6 +416,11 @@ case st of CallbackPollEnabled -> go CallbackPollDisabled -> pure ()++whileNoCallbackRunning :: KafkaConsumer -> IO a -> IO a+whileNoCallbackRunning k f = do+ let statusVar = kcfgCallbackPollStatus (getKafkaConf k)+ withMVar statusVar $ \_ -> f withCallbackPollEnabled :: KafkaConsumer -> IO () -> IO CallbackPollStatus withCallbackPollEnabled k f = do
src/Kafka/Consumer/Callbacks.hs view
@@ -8,23 +8,23 @@ import Control.Arrow ((&&&)) import Control.Monad (forM_, void)-import Data.Monoid ((<>)) import Foreign.ForeignPtr (newForeignPtr_) import Foreign.Ptr (nullPtr) import Kafka.Callbacks as X import Kafka.Consumer.Convert (fromNativeTopicPartitionList', fromNativeTopicPartitionList'') import Kafka.Consumer.Types (KafkaConsumer (..), RebalanceEvent (..), TopicPartition (..)) import Kafka.Internal.RdKafka-import Kafka.Internal.Setup (HasKafka (..), HasKafkaConf (..), Kafka (..), KafkaConf (..), getRdMsgQueue)+import Kafka.Internal.Setup (HasKafka (..), HasKafkaConf (..), Kafka (..), KafkaConf (..), getRdMsgQueue, Callback (..)) import Kafka.Types (KafkaError (..), PartitionId (..), TopicName (..)) import qualified Data.Text as Text -- | Sets a callback that is called when rebalance is needed.-rebalanceCallback :: (KafkaConsumer -> RebalanceEvent -> IO ()) -> KafkaConf -> IO ()-rebalanceCallback callback kc@(KafkaConf conf _ _) = rdKafkaConfSetRebalanceCb conf realCb+rebalanceCallback :: (KafkaConsumer -> RebalanceEvent -> IO ()) -> Callback+rebalanceCallback callback =+ Callback $ \kc@(KafkaConf con _ _) -> rdKafkaConfSetRebalanceCb con (realCb kc) where- realCb k err pl = do+ realCb kc k err pl = do k' <- newForeignPtr_ k pls <- newForeignPtr_ pl setRebalanceCallback callback (KafkaConsumer (Kafka k') kc) (KafkaResponseError err) pls@@ -32,15 +32,16 @@ -- | Sets a callback that is called when rebalance is needed. -- -- The results of automatic or manual offset commits will be scheduled--- for this callback and is served by `pollMessage`.+-- for this callback and is served by 'Kafka.Consumer.pollMessage'. -- -- If no partitions had valid offsets to commit this callback will be called--- with `KafkaError` == `KafkaResponseError` `RdKafkaRespErrNoOffset` which is not to be considered+-- with 'KafkaResponseError' 'RdKafkaRespErrNoOffset' which is not to be considered -- an error.-offsetCommitCallback :: (KafkaConsumer -> KafkaError -> [TopicPartition] -> IO ()) -> KafkaConf -> IO ()-offsetCommitCallback callback kc@(KafkaConf conf _ _) = rdKafkaConfSetOffsetCommitCb conf realCb+offsetCommitCallback :: (KafkaConsumer -> KafkaError -> [TopicPartition] -> IO ()) -> Callback+offsetCommitCallback callback =+ Callback $ \kc@(KafkaConf conf _ _) -> rdKafkaConfSetOffsetCommitCb conf (realCb kc) where- realCb k err pl = do+ realCb kc k err pl = do k' <- newForeignPtr_ k pls <- fromNativeTopicPartitionList' pl callback (KafkaConsumer (Kafka k') kc) (KafkaResponseError err) pls@@ -65,7 +66,10 @@ case e of KafkaResponseError RdKafkaRespErrAssignPartitions -> do f k (RebalanceBeforeAssign assignment)- void $ rdKafkaAssign kptr pls+ protocol <- rdKafkaRebalanceProtocol kptr+ if protocol == "COOPERATIVE" + then void $ rdKafkaIncrementalAssign kptr pls+ else void $ rdKafkaAssign kptr pls mbq <- getRdMsgQueue $ getKafkaConf k case mbq of@@ -83,6 +87,9 @@ KafkaResponseError RdKafkaRespErrRevokePartitions -> do f k (RebalanceBeforeRevoke assignment)- void $ newForeignPtr_ nullPtr >>= rdKafkaAssign kptr+ protocol <- rdKafkaRebalanceProtocol kptr+ if protocol == "COOPERATIVE" + then void $ rdKafkaIncrementalUnassign kptr pls+ else void $ newForeignPtr_ nullPtr >>= rdKafkaAssign kptr f k (RebalanceRevoke assignment) x -> error $ "Rebalance: UNKNOWN response: " <> show x
src/Kafka/Consumer/ConsumerProperties.hs view
@@ -1,5 +1,9 @@ {-# LANGUAGE OverloadedStrings #-} +-----------------------------------------------------------------------------+-- |+-- Module with consumer properties types and functions.+----------------------------------------------------------------------------- module Kafka.Consumer.ConsumerProperties ( ConsumerProperties(..) , CallbackPollMode(..)@@ -13,6 +17,7 @@ , logLevel , compression , suppressDisconnectLogs+, statisticsInterval , extraProps , extraProp , debugOptions@@ -29,18 +34,25 @@ import Data.Text (Text) import qualified Data.Text as Text import Kafka.Consumer.Types (ConsumerGroupId (..))-import Kafka.Internal.Setup (KafkaConf (..)) import Kafka.Types (BrokerAddress (..), ClientId (..), KafkaCompressionCodec (..), KafkaDebug (..), KafkaLogLevel (..), Millis (..), kafkaCompressionCodecToText, kafkaDebugToText) import Kafka.Consumer.Callbacks as X -data CallbackPollMode = CallbackPollModeSync | CallbackPollModeAsync deriving (Show, Eq)+-- | Whether the callback polling should be done synchronously or not.+data CallbackPollMode =+ -- | You have to poll the consumer frequently to handle new messages + -- as well as rebalance and keep alive events.+ -- This enables lowering the footprint and having full control over when polling+ -- happens, at the cost of manually managing those events.+ CallbackPollModeSync+ -- | Handle polling rebalance and keep alive events for you in a background thread.+ | CallbackPollModeAsync deriving (Show, Eq) --- | Properties to create 'KafkaConsumer'.+-- | Properties to create 'Kafka.Consumer.Types.KafkaConsumer'. data ConsumerProperties = ConsumerProperties { cpProps :: Map Text Text , cpLogLevel :: Maybe KafkaLogLevel- , cpCallbacks :: [KafkaConf -> IO ()]+ , cpCallbacks :: [Callback] , cpCallbackPollMode :: CallbackPollMode } @@ -61,11 +73,13 @@ mappend = (Sem.<>) {-# INLINE mappend #-} +-- | Set the <https://kafka.apache.org/documentation/#bootstrap.servers list of brokers> to contact to connect to the Kafka cluster. brokersList :: [BrokerAddress] -> ConsumerProperties brokersList bs =- let bs' = Text.intercalate "," ((\(BrokerAddress x) -> x) <$> bs)+ let bs' = Text.intercalate "," (unBrokerAddress <$> bs) in extraProps $ M.fromList [("bootstrap.servers", bs')] +-- | Set the <https://kafka.apache.org/documentation/#auto.commit.interval.ms auto commit interval> and enables <https://kafka.apache.org/documentation/#enable.auto.commit auto commit>. autoCommit :: Millis -> ConsumerProperties autoCommit (Millis ms) = extraProps $ M.fromList@@ -73,81 +87,87 @@ , ("auto.commit.interval.ms", Text.pack $ show ms) ] --- | Disables auto commit for the consumer+-- | Disable <https://kafka.apache.org/documentation/#enable.auto.commit auto commit> for the consumer. noAutoCommit :: ConsumerProperties noAutoCommit = extraProps $ M.fromList [("enable.auto.commit", "false")] --- | Disables auto offset store for the consumer+-- | Disable auto offset store for the consumer.+--+-- See <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md enable.auto.offset.store> for more information. noAutoOffsetStore :: ConsumerProperties noAutoOffsetStore = extraProps $ M.fromList [("enable.auto.offset.store", "false")] --- | Consumer group id+-- | Set the consumer <https://kafka.apache.org/documentation/#group.id group id>. groupId :: ConsumerGroupId -> ConsumerProperties groupId (ConsumerGroupId cid) = extraProps $ M.fromList [("group.id", cid)] +-- | Set the <https://kafka.apache.org/documentation/#client.id consumer identifier>. clientId :: ClientId -> ConsumerProperties clientId (ClientId cid) = extraProps $ M.fromList [("client.id", cid)] -setCallback :: (KafkaConf -> IO ()) -> ConsumerProperties+-- | Set the consumer callback.+--+-- For examples of use, see:+--+-- * 'errorCallback'+-- * 'logCallback'+-- * 'statsCallback'+setCallback :: Callback -> ConsumerProperties setCallback cb = mempty { cpCallbacks = [cb] } --- | Sets the logging level.+-- | Set the logging level. -- Usually is used with 'debugOptions' to configure which logs are needed. logLevel :: KafkaLogLevel -> ConsumerProperties logLevel ll = mempty { cpLogLevel = Just ll } --- | Sets the compression codec for the consumer.+-- | Set the <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md compression.codec> for the consumer. compression :: KafkaCompressionCodec -> ConsumerProperties compression c = extraProps $ M.singleton "compression.codec" (kafkaCompressionCodecToText c) --- | Suppresses consumer disconnects logs.+-- | Suppresses consumer <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md log.connection.close>. -- -- It might be useful to turn this off when interacting with brokers--- with an aggressive connection.max.idle.ms value.+-- with an aggressive @connection.max.idle.ms@ value. suppressDisconnectLogs :: ConsumerProperties suppressDisconnectLogs = extraProps $ M.fromList [("log.connection.close", "false")] --- | Any configuration options that are supported by /librdkafka/.+-- | Set the <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md statistics.interval.ms> for the producer.+statisticsInterval :: Millis -> ConsumerProperties+statisticsInterval (Millis t) =+ extraProps $ M.singleton "statistics.interval.ms" (Text.pack $ show t)++-- | Set any configuration options that are supported by /librdkafka/. -- The full list can be found <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here> extraProps :: Map Text Text -> ConsumerProperties extraProps m = mempty { cpProps = m } {-# INLINE extraProps #-} --- | Any configuration options that are supported by /librdkafka/.+-- | Set any configuration option that is supported by /librdkafka/. -- The full list can be found <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here> extraProp :: Text -> Text -> ConsumerProperties extraProp k v = mempty { cpProps = M.singleton k v } {-# INLINE extraProp #-} --- | Sets debug features for the consumer.--- Usually is used with 'consumerLogLevel'.+-- | Set <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md debug> features for the consumer.+-- Usually is used with 'logLevel'. debugOptions :: [KafkaDebug] -> ConsumerProperties debugOptions [] = extraProps M.empty debugOptions d = let points = Text.intercalate "," (kafkaDebugToText <$> d) in extraProps $ M.fromList [("debug", points)] +-- | Set <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md queued.max.messages.kbytes> queuedMaxMessagesKBytes :: Int -> ConsumerProperties queuedMaxMessagesKBytes kBytes = extraProp "queued.max.messages.kbytes" (Text.pack $ show kBytes) {-# INLINE queuedMaxMessagesKBytes #-} --- | Sets the callback poll mode.------ The default 'CallbackPollModeAsync' mode handles polling rebalance--- and keep alive events for you--- in a background thread.------ With 'CallbacPollModeSync' the user will poll the consumer--- frequently to handle new messages as well as rebalance and keep alive events.--- 'CallbacPollModeSync' lets you can simplify--- hw-kafka-client's footprint and have full control over when polling--- happens at the cost of having to manage this yourself.+-- | Set the callback poll mode. Default value is 'CallbackPollModeAsync'. callbackPollMode :: CallbackPollMode -> ConsumerProperties callbackPollMode mode = mempty { cpCallbackPollMode = mode }
src/Kafka/Consumer/Convert.hs view
@@ -18,6 +18,7 @@ import Control.Monad ((>=>)) import qualified Data.ByteString as BS+import Data.Either (fromRight) import Data.Int (Int64) import Data.Map.Strict (Map, fromListWith) import qualified Data.Set as S@@ -41,7 +42,7 @@ , rdKafkaTopicPartitionListNew , peekCText )-import Kafka.Internal.Shared (kafkaRespErr, readTopic, readKey, readPayload, readTimestamp)+import Kafka.Internal.Shared (kafkaRespErr, readHeaders, readTopic, readKey, readPayload, readTimestamp) import Kafka.Types (KafkaError(..), PartitionId(..), TopicName(..)) -- | Converts offsets sync policy to integer (the way Kafka understands it):@@ -158,20 +159,22 @@ s <- peek realPtr msg <- if err'RdKafkaMessageT s /= RdKafkaRespErrNoError then return . Left . KafkaResponseError $ err'RdKafkaMessageT s- else Right <$> mkRecord s+ else Right <$> mkRecord s realPtr rdKafkaMessageDestroy realPtr return msg where- mkRecord msg = do+ mkRecord msg rptr = do topic <- readTopic msg key <- readKey msg payload <- readPayload msg timestamp <- readTimestamp ptr+ headers <- fromRight mempty <$> readHeaders rptr return ConsumerRecord { crTopic = TopicName topic , crPartition = PartitionId $ partition'RdKafkaMessageT msg , crOffset = Offset $ offset'RdKafkaMessageT msg , crTimestamp = timestamp+ , crHeaders = headers , crKey = key , crValue = payload }
src/Kafka/Consumer/Subscription.hs view
@@ -1,5 +1,9 @@ {-# LANGUAGE OverloadedStrings #-} +-----------------------------------------------------------------------------+-- |+-- Module with subscription types and functions.+----------------------------------------------------------------------------- module Kafka.Consumer.Subscription ( Subscription(..) , topics@@ -17,6 +21,18 @@ import Kafka.Consumer.Types (OffsetReset (..)) import Kafka.Types (TopicName (..)) +-- | A consumer subscription to a topic.+--+-- ==== __Examples__+--+-- Typically you don't call the constructor directly, but combine settings:+--+-- @+-- consumerSub :: 'Subscription'+-- consumerSub = 'topics' ['TopicName' "kafka-client-example-topic"]+-- <> 'offsetReset' 'Earliest'+-- <> 'extraSubscriptionProps' (fromList [("prop1", "value 1"), ("prop2", "value 2")])+-- @ data Subscription = Subscription (Set TopicName) (Map Text Text) instance Sem.Semigroup Subscription where@@ -32,9 +48,11 @@ mappend = (Sem.<>) {-# INLINE mappend #-} +-- | Build a subscription by giving the list of topic names only topics :: [TopicName] -> Subscription topics ts = Subscription (Set.fromList ts) M.empty +-- | Build a subscription by giving the offset reset parameter only offsetReset :: OffsetReset -> Subscription offsetReset o = let o' = case o of@@ -42,5 +60,6 @@ Latest -> "latest" in Subscription Set.empty (M.fromList [("auto.offset.reset", o')]) +-- | Build a subscription by giving extra properties only extraSubscriptionProps :: Map Text Text -> Subscription extraSubscriptionProps = Subscription Set.empty
src/Kafka/Consumer/Types.hs view
@@ -1,5 +1,11 @@-{-# LANGUAGE DeriveDataTypeable #-}-{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++-----------------------------------------------------------------------------+-- |+-- Module holding consumer types.+----------------------------------------------------------------------------- module Kafka.Consumer.Types ( KafkaConsumer(..) , ConsumerGroupId(..)@@ -30,14 +36,18 @@ import Data.Bifoldable (Bifoldable (..)) import Data.Bifunctor (Bifunctor (..))-import Data.Bitraversable (Bitraversable (..), bimapM, bisequenceA)+import Data.Bitraversable (Bitraversable (..), bimapM, bisequence) import Data.Int (Int64)+import Data.String (IsString) import Data.Text (Text) import Data.Typeable (Typeable) import GHC.Generics (Generic) import Kafka.Internal.Setup (HasKafka (..), HasKafkaConf (..), Kafka (..), KafkaConf (..))-import Kafka.Types (Millis (..), PartitionId (..), TopicName (..))+import Kafka.Types (Millis (..), PartitionId (..), TopicName (..), Headers) +-- | The main type for Kafka consumption, used e.g. to poll and commit messages.+-- +-- Its constructor is intentionally not exposed, instead, one should use 'Kafka.Consumer.newConsumer' to acquire such a value. data KafkaConsumer = KafkaConsumer { kcKafkaPtr :: !Kafka , kcKafkaConf :: !KafkaConf@@ -51,8 +61,19 @@ getKafkaConf = kcKafkaConf {-# INLINE getKafkaConf #-} -newtype ConsumerGroupId = ConsumerGroupId { unConsumerGroupId :: Text } deriving (Show, Ord, Eq, Generic)+-- | Consumer group ID. Different consumers with the same consumer group ID will get assigned different partitions of each subscribed topic. +-- +-- See <https://kafka.apache.org/documentation/#group.id Kafka documentation on consumer group>+newtype ConsumerGroupId = ConsumerGroupId+ { unConsumerGroupId :: Text+ } deriving (Show, Ord, Eq, IsString, Generic)++-- | A message offset in a partition newtype Offset = Offset { unOffset :: Int64 } deriving (Show, Eq, Ord, Read, Generic)++-- | Where to reset the offset when there is no initial offset in Kafka+-- +-- See <https://kafka.apache.org/documentation/#auto.offset.reset Kafka documentation on offset reset> data OffsetReset = Earliest | Latest deriving (Show, Eq, Generic) -- | A set of events which happen during the rebalancing process@@ -67,6 +88,7 @@ | RebalanceRevoke [(TopicName, PartitionId)] deriving (Eq, Show, Generic) +-- | The partition offset data PartitionOffset = PartitionOffsetBeginning | PartitionOffsetEnd@@ -75,11 +97,13 @@ | PartitionOffsetInvalid deriving (Eq, Show, Generic) +-- | Partitions subscribed by a consumer data SubscribedPartitions- = SubscribedPartitions [PartitionId]- | SubscribedPartitionsAll+ = SubscribedPartitions [PartitionId] -- ^ Subscribe only to those partitions+ | SubscribedPartitionsAll -- ^ Subscribe to all partitions deriving (Show, Eq, Generic) +-- | Consumer record timestamp data Timestamp = CreateTime !Millis | LogAppendTime !Millis@@ -119,13 +143,14 @@ , crPartition :: !PartitionId -- ^ Kafka partition this message was received from , crOffset :: !Offset -- ^ Offset within the 'crPartition' Kafka partition , crTimestamp :: !Timestamp -- ^ Message timestamp- , crKey :: !k- , crValue :: !v+ , crHeaders :: !Headers -- ^ Message headers+ , crKey :: !k -- ^ Message key+ , crValue :: !v -- ^ Message value } deriving (Eq, Show, Read, Typeable, Generic) instance Bifunctor ConsumerRecord where- bimap f g (ConsumerRecord t p o ts k v) = ConsumerRecord t p o ts (f k) (g v)+ bimap f g (ConsumerRecord t p o ts hds k v) = ConsumerRecord t p o ts hds (f k) (g v) {-# INLINE bimap #-} instance Functor (ConsumerRecord k) where@@ -148,24 +173,27 @@ bitraverse f g r = (\k v -> bimap (const k) (const v) r) <$> f (crKey r) <*> g (crValue r) {-# INLINE bitraverse #-} +{-# DEPRECATED crMapKey "Isn't concern of this library. Use 'first'" #-} crMapKey :: (k -> k') -> ConsumerRecord k v -> ConsumerRecord k' v crMapKey = first {-# INLINE crMapKey #-} +{-# DEPRECATED crMapValue "Isn't concern of this library. Use 'second'" #-} crMapValue :: (v -> v') -> ConsumerRecord k v -> ConsumerRecord k v' crMapValue = second {-# INLINE crMapValue #-} +{-# DEPRECATED crMapKV "Isn't concern of this library. Use 'bimap'" #-} crMapKV :: (k -> k') -> (v -> v') -> ConsumerRecord k v -> ConsumerRecord k' v' crMapKV = bimap {-# INLINE crMapKV #-} -{-# DEPRECATED sequenceFirst "Isn't concern of this library. Use 'bitraverse id pure'" #-}+{-# DEPRECATED sequenceFirst "Isn't concern of this library. Use @'bitraverse' 'id' 'pure'@" #-} sequenceFirst :: (Bitraversable t, Applicative f) => t (f k) v -> f (t k v) sequenceFirst = bitraverse id pure {-# INLINE sequenceFirst #-} -{-# DEPRECATED traverseFirst "Isn't concern of this library. Use 'bitraverse f pure'" #-}+{-# DEPRECATED traverseFirst "Isn't concern of this library. Use @'bitraverse' f 'pure'@" #-} traverseFirst :: (Bitraversable t, Applicative f) => (k -> f k') -> t k v@@ -173,7 +201,7 @@ traverseFirst f = bitraverse f pure {-# INLINE traverseFirst #-} -{-# DEPRECATED traverseFirstM "Isn't concern of this library. Use 'bitraverse id pure <$> bitraverse f pure r'" #-}+{-# DEPRECATED traverseFirstM "Isn't concern of this library. Use @'bitraverse' 'id' 'pure' '<$>' 'bitraverse' f 'pure' r@" #-} traverseFirstM :: (Bitraversable t, Applicative f, Monad m) => (k -> m (f k')) -> t k v@@ -181,7 +209,7 @@ traverseFirstM f r = bitraverse id pure <$> bitraverse f pure r {-# INLINE traverseFirstM #-} -{-# DEPRECATED traverseM "Isn't concern of this library. Use 'sequenceA <$> traverse f r'" #-}+{-# DEPRECATED traverseM "Isn't concern of this library. Use @'sequenceA' '<$>' 'traverse' f r@" #-} traverseM :: (Traversable t, Applicative f, Monad m) => (v -> m (f v')) -> t v@@ -189,12 +217,12 @@ traverseM f r = sequenceA <$> traverse f r {-# INLINE traverseM #-} -{-# DEPRECATED bitraverseM "Isn't concern of this library. Use 'bisequenceA <$> bimapM f g r'" #-}+{-# DEPRECATED bitraverseM "Isn't concern of this library. Use @'Data.Bitraversable.bisequenceA' '<$>' 'bimapM' f g r@" #-} bitraverseM :: (Bitraversable t, Applicative f, Monad m) => (k -> m (f k')) -> (v -> m (f v')) -> t k v -> m (f (t k' v'))-bitraverseM f g r = bisequenceA <$> bimapM f g r+bitraverseM f g r = bisequence <$> bimapM f g r {-# INLINE bitraverseM #-}
src/Kafka/Dump.hs view
@@ -1,3 +1,8 @@+-----------------------------------------------------------------------------+-- |+-- Module providing various functions to dump information. These may be useful for+-- debug/investigation but should probably not be used on production applications.+----------------------------------------------------------------------------- module Kafka.Dump ( hPrintSupportedKafkaConf , hPrintKafka
src/Kafka/Internal/RdKafka.chs view
@@ -3,19 +3,23 @@ module Kafka.Internal.RdKafka where +import Data.ByteString (ByteString)+import qualified Data.ByteString as BS import Data.Text (Text) import qualified Data.Text as Text import Control.Monad (liftM) import Data.Int (Int32, Int64) import Data.Word (Word8)+import Foreign.Concurrent (newForeignPtr)+import qualified Foreign.Concurrent as Concurrent import Foreign.Marshal.Alloc (alloca, allocaBytes)-import Foreign.Marshal.Array (peekArray, allocaArray)+import Foreign.Marshal.Array (peekArray, allocaArray, withArrayLen) import Foreign.Storable (Storable(..)) import Foreign.Ptr (Ptr, FunPtr, castPtr, nullPtr)-import Foreign.ForeignPtr (FinalizerPtr, addForeignPtrFinalizer, withForeignPtr, newForeignPtr, newForeignPtr_)+import Foreign.ForeignPtr (FinalizerPtr, addForeignPtrFinalizer, newForeignPtr_, withForeignPtr) import Foreign.C.Error (Errno(..), getErrno)-import Foreign.C.String (CString, newCString, withCAString, peekCAString, peekCAStringLen, peekCString)-import Foreign.C.Types (CFile, CInt(..), CSize, CChar)+import Foreign.C.String (CString, newCString, withCAString, peekCAString, peekCString)+import Foreign.C.Types (CFile, CInt(..), CSize, CChar, CLong) import System.IO (Handle, stdin, stdout, stderr) import System.Posix.IO (handleToFd) import System.Posix.Types (Fd(..))@@ -147,6 +151,7 @@ , offset'RdKafkaMessageT :: Int64 , payload'RdKafkaMessageT :: Word8Ptr , key'RdKafkaMessageT :: Word8Ptr+ , opaque'RdKafkaMessageT :: Ptr () } deriving (Show, Eq) @@ -162,6 +167,7 @@ <*> liftM fromIntegral ({#get rd_kafka_message_t->offset #} p) <*> liftM castPtr ({#get rd_kafka_message_t->payload #} p) <*> liftM castPtr ({#get rd_kafka_message_t->key #} p)+ <*> liftM castPtr ({#get rd_kafka_message_t->_private #} p) poke p x = do {#set rd_kafka_message_t.err#} p (enumToCInt $ err'RdKafkaMessageT x) {#set rd_kafka_message_t.rkt#} p (castPtr $ topic'RdKafkaMessageT x)@@ -171,6 +177,7 @@ {#set rd_kafka_message_t.offset#} p (fromIntegral $ offset'RdKafkaMessageT x) {#set rd_kafka_message_t.payload#} p (castPtr $ payload'RdKafkaMessageT x) {#set rd_kafka_message_t.key#} p (castPtr $ key'RdKafkaMessageT x)+ {#set rd_kafka_message_t._private#} p (castPtr $ opaque'RdKafkaMessageT x) {#pointer *rd_kafka_message_t as RdKafkaMessageTPtr foreign -> RdKafkaMessageT #} @@ -264,12 +271,14 @@ {`Int'} -> `RdKafkaTopicPartitionListTPtr' #} foreign import ccall unsafe "rdkafka.h &rd_kafka_topic_partition_list_destroy"- rdKafkaTopicPartitionListDestroy :: FinalizerPtr RdKafkaTopicPartitionListT+ rdKafkaTopicPartitionListDestroyF :: FinalizerPtr RdKafkaTopicPartitionListT+foreign import ccall unsafe "rdkafka.h rd_kafka_topic_partition_list_destroy"+ rdKafkaTopicPartitionListDestroy :: Ptr RdKafkaTopicPartitionListT -> IO () newRdKafkaTopicPartitionListT :: Int -> IO RdKafkaTopicPartitionListTPtr newRdKafkaTopicPartitionListT size = do ret <- rdKafkaTopicPartitionListNew size- addForeignPtrFinalizer rdKafkaTopicPartitionListDestroy ret+ addForeignPtrFinalizer rdKafkaTopicPartitionListDestroyF ret return ret {# fun rd_kafka_topic_partition_list_add as ^@@ -284,7 +293,7 @@ copyRdKafkaTopicPartitionList :: RdKafkaTopicPartitionListTPtr -> IO RdKafkaTopicPartitionListTPtr copyRdKafkaTopicPartitionList pl = do cp <- rdKafkaTopicPartitionListCopy pl- addForeignPtrFinalizer rdKafkaTopicPartitionListDestroy cp+ addForeignPtrFinalizer rdKafkaTopicPartitionListDestroyF cp return cp {# fun rd_kafka_topic_partition_list_set_offset as ^@@ -407,8 +416,8 @@ withForeignPtr conf $ \c -> rdKafkaConfSetLogCb' c cb' ---- Stats Callback-type StatsCallback' = Ptr RdKafkaT -> CString -> CSize -> Word8Ptr -> IO ()-type StatsCallback = Ptr RdKafkaT -> String -> IO ()+type StatsCallback' = Ptr RdKafkaT -> CString -> CSize -> Word8Ptr -> IO CInt+type StatsCallback = Ptr RdKafkaT -> ByteString -> IO () foreign import ccall safe "wrapper" mkStatsCallback :: StatsCallback' -> IO (FunPtr StatsCallback')@@ -418,12 +427,12 @@ rdKafkaConfSetStatsCb :: RdKafkaConfTPtr -> StatsCallback -> IO () rdKafkaConfSetStatsCb conf cb = do- cb' <- mkStatsCallback $ \k j jl _ -> peekCAStringLen (j, cIntConv jl) >>= cb k+ cb' <- mkStatsCallback $ \k j jl _ -> BS.packCStringLen (j, cIntConv jl) >>= cb k >> pure 0 withForeignPtr conf $ \c -> rdKafkaConfSetStatsCb' c cb' return () ---- Socket Callback-type SocketCallback = Int -> Int -> Int -> Word8Ptr -> IO ()+type SocketCallback = Int -> Int -> Int -> Word8Ptr -> IO CInt foreign import ccall safe "wrapper" mkSocketCallback :: SocketCallback -> IO (FunPtr SocketCallback)@@ -434,7 +443,7 @@ rdKafkaConfSetSocketCb :: RdKafkaConfTPtr -> SocketCallback -> IO () rdKafkaConfSetSocketCb conf cb = do cb' <- mkSocketCallback cb- withForeignPtr conf $ \c -> rdKafkaConfSetSocketCb' c cb'+ _ <- withForeignPtr conf $ \c -> rdKafkaConfSetSocketCb' c cb' >> pure (0 :: Int) return () {#fun rd_kafka_conf_set_opaque as ^@@ -557,7 +566,7 @@ rdKafkaConsumeBatchQueue qptr timeout batchSize = do allocaArray batchSize $ \pArr -> do rSize <- rdKafkaConsumeBatchQueue' qptr timeout pArr (fromIntegral batchSize)- peekArray (fromIntegral rSize) pArr >>= traverse newForeignPtr_+ peekArray (fromIntegral rSize) pArr >>= traverse (flip newForeignPtr (return ())) ------------------------------------------------------------------------------------------------- ---- High-level KafkaConsumer@@ -579,7 +588,7 @@ (err, sub) <- rdKafkaSubscription' k case err of RdKafkaRespErrNoError ->- Right <$> newForeignPtr rdKafkaTopicPartitionListDestroy sub+ Right <$> newForeignPtr sub (rdKafkaTopicPartitionListDestroy sub) e -> return (Left e) {#fun rd_kafka_consumer_poll as ^@@ -602,6 +611,9 @@ {`RdKafkaTPtr', `RdKafkaTopicPartitionListTPtr'} -> `RdKafkaRespErrT' cIntToEnum #} +{#fun rd_kafka_rebalance_protocol as ^+ {`RdKafkaTPtr'} -> `String' #}+ {#fun rd_kafka_assignment as rdKafkaAssignment' {`RdKafkaTPtr', alloca- `Ptr RdKafkaTopicPartitionListT' peekPtr* } -> `RdKafkaRespErrT' cIntToEnum #}@@ -611,7 +623,7 @@ (err, ass) <- rdKafkaAssignment' k case err of RdKafkaRespErrNoError ->- Right <$> newForeignPtr rdKafkaTopicPartitionListDestroy ass+ Right <$> newForeignPtr ass (rdKafkaTopicPartitionListDestroy ass) e -> return (Left e) {#fun rd_kafka_commit as ^@@ -718,8 +730,11 @@ -> `RdKafkaRespErrT' cIntToEnum #} foreign import ccall "rdkafka.h &rd_kafka_group_list_destroy"- rdKafkaGroupListDestroy :: FinalizerPtr RdKafkaGroupListT+ rdKafkaGroupListDestroyF :: FinalizerPtr RdKafkaGroupListT +foreign import ccall "rdkafka.h rd_kafka_group_list_destroy"+ rdKafkaGroupListDestroy :: Ptr RdKafkaGroupListT -> IO ()+ rdKafkaListGroups :: RdKafkaTPtr -> Maybe String -> Int -> IO (Either RdKafkaRespErrT RdKafkaGroupListTPtr) rdKafkaListGroups k g t = case g of Nothing -> listGroups nullPtr@@ -728,7 +743,7 @@ listGroups grp = do (err, res) <- rdKafkaListGroups' k grp t case err of- RdKafkaRespErrNoError -> Right <$> newForeignPtr rdKafkaGroupListDestroy res+ RdKafkaRespErrNoError -> Right <$> newForeignPtr res (rdKafkaGroupListDestroy res) e -> return $ Left e ------------------------------------------------------------------------------------------------- @@ -827,9 +842,6 @@ {enumToCInt `RdKafkaTypeT', `RdKafkaConfTPtr', id `CCharBufPointer', cIntConv `CSize'} -> `RdKafkaTPtr' #} -foreign import ccall safe "rdkafka.h &rd_kafka_destroy"- rdKafkaDestroy :: FunPtr (Ptr RdKafkaT -> IO ())- newRdKafkaT :: RdKafkaTypeT -> RdKafkaConfTPtr -> IO (Either Text RdKafkaTPtr) newRdKafkaT kafkaType confPtr = allocaBytes nErrorBytes $ \charPtr -> do@@ -838,10 +850,15 @@ withForeignPtr ret $ \realPtr -> do if realPtr == nullPtr then peekCText charPtr >>= return . Left else do- -- do not call 'rd_kafka_close_consumer' on destroying all Kafka.- -- when needed, applications should do it explicitly.- -- {# call rd_kafka_destroy_flags #} realPtr 0x8- addForeignPtrFinalizer rdKafkaDestroy ret+ -- Issue #151+ -- rd_kafka_destroy_flags may call back into Haskell if an+ -- error or log callback is set, so we must use a concurrent+ -- finalizer+ Concurrent.addForeignPtrFinalizer ret $ do+ -- do not call 'rd_kafka_close_consumer' on destroying all Kafka.+ -- when needed, applications should do it explicitly.+ -- RD_KAFKA_DESTROY_F_NO_CONSUMER_CLOSE = 0x8+ {# call rd_kafka_destroy_flags #} realPtr 0x8 return $ Right ret {#fun rd_kafka_brokers_add as ^@@ -893,7 +910,7 @@ {#fun rd_kafka_produce as ^ {`RdKafkaTopicTPtr', cIntConv `CInt32T', `Int', castPtr `Word8Ptr',- cIntConv `CSize', castPtr `Word8Ptr', cIntConv `CSize', castPtr `Word8Ptr'}+ cIntConv `CSize', castPtr `Word8Ptr', cIntConv `CSize', castPtr `Ptr ()'} -> `Int' #} {#fun rd_kafka_produce_batch as ^@@ -907,15 +924,15 @@ alloca- `Ptr RdKafkaMetadataT' peekPtr*, `Int'} -> `RdKafkaRespErrT' cIntToEnum #} -foreign import ccall unsafe "rdkafka.h &rd_kafka_metadata_destroy"- rdKafkaMetadataDestroy :: FinalizerPtr RdKafkaMetadataT+foreign import ccall unsafe "rdkafka.h rd_kafka_metadata_destroy"+ rdKafkaMetadataDestroy :: Ptr RdKafkaMetadataT -> IO () rdKafkaMetadata :: RdKafkaTPtr -> Bool -> Maybe RdKafkaTopicTPtr -> Int -> IO (Either RdKafkaRespErrT RdKafkaMetadataTPtr) rdKafkaMetadata k allTopics mt timeout = do tptr <- maybe (newForeignPtr_ nullPtr) pure mt (err, res) <- rdKafkaMetadata' k allTopics tptr timeout case err of- RdKafkaRespErrNoError -> Right <$> newForeignPtr rdKafkaMetadataDestroy res+ RdKafkaRespErrNoError -> Right <$> newForeignPtr res (rdKafkaMetadataDestroy res) e -> return (Left e) {#fun rd_kafka_poll as ^@@ -958,6 +975,178 @@ _ <- traverse (addForeignPtrFinalizer rdKafkaTopicDestroy') res return res +-------------------------------------------------------------------------------------------------+---- Errors++data RdKafkaErrorT+{#pointer *rd_kafka_error_t as RdKafkaErrorTPtr -> RdKafkaErrorT #}++{#fun rd_kafka_error_code as ^+ {`RdKafkaErrorTPtr'} -> `RdKafkaRespErrT' cIntToEnum #}++{#fun rd_kafka_error_destroy as ^+ {`RdKafkaErrorTPtr'} -> `()' #}+-------------------------------------------------------------------------------------------------+---- Headers++data RdKafkaHeadersT+{#pointer *rd_kafka_headers_t as RdKafkaHeadersTPtr -> RdKafkaHeadersT #}++{#fun rd_kafka_header_get_all as ^+ {`RdKafkaHeadersTPtr', cIntConv `CSize', castPtr `Ptr CString', castPtr `Ptr Word8Ptr', castPtr `CSizePtr'} -> `RdKafkaRespErrT' cIntToEnum #}++{#fun rd_kafka_message_headers as ^+ {castPtr `Ptr RdKafkaMessageT', alloca- `RdKafkaHeadersTPtr' peekPtr*} -> `RdKafkaRespErrT' cIntToEnum #}++--- Produceva api++{#enum rd_kafka_vtype_t as ^ {underscoreToCase} deriving (Show, Eq) #}++data RdKafkaVuT+ = Topic'RdKafkaVu CString+ | TopicHandle'RdKafkaVu (Ptr RdKafkaTopicT)+ | Partition'RdKafkaVu CInt32T+ | Value'RdKafkaVu Word8Ptr CSize+ | Key'RdKafkaVu Word8Ptr CSize+ | MsgFlags'RdKafkaVu CInt+ | Timestamp'RdKafkaVu CInt64T+ | Opaque'RdKafkaVu (Ptr ())+ | Header'RdKafkaVu CString Word8Ptr CSize+ | Headers'RdKafkaVu (Ptr RdKafkaHeadersT) -- The message object will assume ownership of the headers (unless produceva() fails)+ | End'RdKafkaVu++{#pointer *rd_kafka_vu_t as RdKafkaVuTPtr foreign -> RdKafkaVuT #}++instance Storable RdKafkaVuT where+ alignment _ = {#alignof rd_kafka_vu_t #}+ sizeOf _ = {#sizeof rd_kafka_vu_t #}+ peek p = {#get rd_kafka_vu_t->vtype #} p >>= \a -> case cIntToEnum a of+ RdKafkaVtypeEnd -> return End'RdKafkaVu+ RdKafkaVtypeTopic -> Topic'RdKafkaVu <$> ({#get rd_kafka_vu_t->u.cstr #} p)+ RdKafkaVtypeMsgflags -> MsgFlags'RdKafkaVu <$> ({#get rd_kafka_vu_t->u.i #} p)+ RdKafkaVtypeTimestamp -> Timestamp'RdKafkaVu <$> ({#get rd_kafka_vu_t->u.i64 #} p)+ RdKafkaVtypePartition -> Partition'RdKafkaVu <$> ({#get rd_kafka_vu_t->u.i32 #} p)+ RdKafkaVtypeHeaders -> Headers'RdKafkaVu <$> ({#get rd_kafka_vu_t->u.headers #} p)+ RdKafkaVtypeValue -> do+ nm <- liftM castPtr ({#get rd_kafka_vu_t->u.mem.ptr #} p)+ sz <- ({#get rd_kafka_vu_t->u.mem.size #} p)+ return $ Value'RdKafkaVu nm (cIntConv sz)+ RdKafkaVtypeKey -> do+ nm <- liftM castPtr ({#get rd_kafka_vu_t->u.mem.ptr #} p)+ sz <- ({#get rd_kafka_vu_t->u.mem.size #} p)+ return $ Key'RdKafkaVu nm (cIntConv sz)+ RdKafkaVtypeRkt -> TopicHandle'RdKafkaVu <$> ({#get rd_kafka_vu_t->u.rkt #} p)+ RdKafkaVtypeOpaque -> Opaque'RdKafkaVu <$> ({#get rd_kafka_vu_t->u.ptr #} p)+ RdKafkaVtypeHeader -> do+ nm <- ({#get rd_kafka_vu_t->u.header.name #} p)+ val' <- liftM castPtr ({#get rd_kafka_vu_t->u.header.val #} p)+ sz <- ({#get rd_kafka_vu_t->u.header.size #} p)+ return $ Header'RdKafkaVu nm val' (cIntConv sz)+ poke p End'RdKafkaVu =+ {#set rd_kafka_vu_t.vtype #} p (enumToCInt RdKafkaVtypeEnd)+ poke p (Topic'RdKafkaVu str) = do+ {#set rd_kafka_vu_t.vtype #} p (enumToCInt RdKafkaVtypeTopic)+ {#set rd_kafka_vu_t.u.cstr #} p str+ poke p (Timestamp'RdKafkaVu tms) = do+ {#set rd_kafka_vu_t.vtype #} p (enumToCInt RdKafkaVtypeTimestamp)+ {#set rd_kafka_vu_t.u.i64 #} p tms+ poke p (Partition'RdKafkaVu prt) = do+ {#set rd_kafka_vu_t.vtype #} p (enumToCInt RdKafkaVtypePartition)+ {#set rd_kafka_vu_t.u.i32 #} p prt+ poke p (MsgFlags'RdKafkaVu flags) = do+ {#set rd_kafka_vu_t.vtype #} p (enumToCInt RdKafkaVtypeMsgflags)+ {#set rd_kafka_vu_t.u.i #} p flags+ poke p (Headers'RdKafkaVu headers) = do+ {#set rd_kafka_vu_t.vtype #} p (enumToCInt RdKafkaVtypeHeaders)+ {#set rd_kafka_vu_t.u.headers #} p headers+ poke p (TopicHandle'RdKafkaVu tphandle) = do+ {#set rd_kafka_vu_t.vtype #} p (enumToCInt RdKafkaVtypeRkt)+ {#set rd_kafka_vu_t.u.rkt #} p tphandle+ poke p (Value'RdKafkaVu pl sz) = do+ {#set rd_kafka_vu_t.vtype #} p (enumToCInt RdKafkaVtypeValue)+ {#set rd_kafka_vu_t.u.mem.size #} p (cIntConv sz)+ {#set rd_kafka_vu_t.u.mem.ptr #} p (castPtr pl)+ poke p (Key'RdKafkaVu pl sz) = do+ {#set rd_kafka_vu_t.vtype #} p (enumToCInt RdKafkaVtypeKey)+ {#set rd_kafka_vu_t.u.mem.size #} p (cIntConv sz)+ {#set rd_kafka_vu_t.u.mem.ptr #} p (castPtr pl)+ poke p (Opaque'RdKafkaVu ptr') = do+ {#set rd_kafka_vu_t.vtype #} p (enumToCInt RdKafkaVtypeOpaque)+ {#set rd_kafka_vu_t.u.ptr #} p ptr'+ poke p (Header'RdKafkaVu nm val' sz) = do+ {#set rd_kafka_vu_t.vtype #} p (enumToCInt RdKafkaVtypeHeader)+ {#set rd_kafka_vu_t.u.header.size #} p (cIntConv sz)+ {#set rd_kafka_vu_t.u.header.name #} p nm+ {#set rd_kafka_vu_t.u.header.val #} p (castPtr val')++{#fun rd_kafka_produceva as rdKafkaMessageProduceVa'+ {`RdKafkaTPtr', `RdKafkaVuTPtr', `CLong'} -> `RdKafkaErrorTPtr' #}++rdKafkaMessageProduceVa :: RdKafkaTPtr -> [RdKafkaVuT] -> IO RdKafkaErrorTPtr+rdKafkaMessageProduceVa kafkaPtr vts = withArrayLen vts $ \i arrPtr -> do+ fptr <- newForeignPtr_ arrPtr+ rdKafkaMessageProduceVa' kafkaPtr fptr (cIntConv i)++--- Transactional api++{#fun rd_kafka_init_transactions as rdKafkaInitTransactions+ {`RdKafkaTPtr', `Int'} -> `RdKafkaErrorTPtr' #}++{#fun rd_kafka_begin_transaction as rdKafkaBeginTransaction+ {`RdKafkaTPtr'} -> `RdKafkaErrorTPtr' #}++{#fun rd_kafka_send_offsets_to_transaction as rdKafkaSendOffsetsToTransaction'+ {`RdKafkaTPtr', `RdKafkaTopicPartitionListTPtr', `Ptr ()', `Int' } -> `RdKafkaErrorTPtr' #}++{#fun rd_kafka_commit_transaction as rdKafkaCommitTransaction+ {`RdKafkaTPtr', `Int'} -> `RdKafkaErrorTPtr' #}++{#fun rd_kafka_abort_transaction as rdKafkaAbortTransaction+ {`RdKafkaTPtr', `Int'} -> `RdKafkaErrorTPtr' #}++{#fun rd_kafka_incremental_assign as ^+ {`RdKafkaTPtr', `RdKafkaTopicPartitionListTPtr'} -> `RdKafkaErrorTPtr' #}++{#fun rd_kafka_incremental_unassign as ^+ {`RdKafkaTPtr', `RdKafkaTopicPartitionListTPtr'}+ -> `RdKafkaErrorTPtr' #}++{#fun rd_kafka_consumer_group_metadata as rdKafkaConsumerGroupMetadata+ {`RdKafkaTPtr'} -> `Ptr ()' #}++{#fun rd_kafka_consumer_group_metadata_destroy as rdKafkaConsumerGroupMetadataDestroy+ {`Ptr ()'} -> `()' #}++{#fun rd_kafka_seek_partitions as rdKafkaSeekPartitions+ {`RdKafkaTPtr', `RdKafkaTopicPartitionListTPtr', `Int'} -> `RdKafkaErrorTPtr' #}++rdKafkaSendOffsetsToTransaction :: RdKafkaTPtr -> RdKafkaTPtr -> RdKafkaTopicPartitionListTPtr -> Int -> IO RdKafkaErrorTPtr+rdKafkaSendOffsetsToTransaction p c topicPartition timeOut = do+ metaData <- rdKafkaConsumerGroupMetadata c+ errPtr <- rdKafkaSendOffsetsToTransaction' p topicPartition metaData timeOut+ -- NOTE:need to destroy the metadata otherwise leaking memory+ rdKafkaConsumerGroupMetadataDestroy metaData+ pure errPtr++{#fun rd_kafka_error_is_fatal as rdKafkaErrorIsFatal'+ {`RdKafkaErrorTPtr'} -> `CInt' #}++{#fun rd_kafka_error_is_retriable as rdKafkaErrorIsRetriable'+ {`RdKafkaErrorTPtr'} -> `CInt' #}++{#fun rd_kafka_error_txn_requires_abort as rdKafkaErrorTxnRequiresAbort'+ {`RdKafkaErrorTPtr'} -> `CInt' #}++rdKafkaErrorIsFatal :: RdKafkaErrorTPtr -> IO Bool+rdKafkaErrorIsFatal ptr = boolFromCInt <$> rdKafkaErrorIsFatal' ptr++rdKafkaErrorIsRetriable :: RdKafkaErrorTPtr -> IO Bool+rdKafkaErrorIsRetriable ptr = boolFromCInt <$> rdKafkaErrorIsRetriable' ptr++rdKafkaErrorTxnRequiresAbort :: RdKafkaErrorTPtr -> IO Bool+rdKafkaErrorTxnRequiresAbort ptr = boolFromCInt <$> rdKafkaErrorTxnRequiresAbort' ptr++ -- Marshall / Unmarshall enumToCInt :: Enum a => a -> CInt enumToCInt = fromIntegral . fromEnum@@ -975,6 +1164,11 @@ boolToCInt True = CInt 1 boolToCInt False = CInt 0 {-# INLINE boolToCInt #-}++boolFromCInt :: CInt -> Bool+boolFromCInt (CInt 0) = False+boolFromCInt (CInt _) = True+{-# INLINE boolFromCInt #-} peekInt64Conv :: (Storable a, Integral a) => Ptr a -> IO Int64 peekInt64Conv = liftM cIntConv . peek
src/Kafka/Internal/Setup.hs view
@@ -7,6 +7,7 @@ , HasKafka(..) , HasKafkaConf(..) , HasTopicConf(..)+, Callback(..) , CallbackPollStatus(..) , getRdKafka , getRdKafkaConf@@ -45,6 +46,11 @@ newtype TopicProps = TopicProps (Map Text Text) deriving (Show, Eq) newtype Kafka = Kafka RdKafkaTPtr deriving Show newtype TopicConf = TopicConf RdKafkaTopicConfTPtr deriving Show++-- | Callbacks allow retrieving various information like error occurences, statistics+-- and log messages.+-- See `Kafka.Consumer.setCallback` (Consumer) and `Kafka.Producer.setCallback` (Producer) for more details.+newtype Callback = Callback (KafkaConf -> IO ()) data CallbackPollStatus = CallbackPollEnabled | CallbackPollDisabled deriving (Show, Eq)
@@ -1,3 +1,5 @@+{-# LANGUAGE LambdaCase #-}+ module Kafka.Internal.Shared ( pollEvents , word8PtrToBS@@ -8,6 +10,7 @@ , kafkaErrorToEither , kafkaErrorToMaybe , maybeToLeft+, readHeaders , readPayload , readTopic , readKey@@ -29,14 +32,14 @@ import Foreign.Ptr (Ptr, nullPtr) import Foreign.Storable (Storable (peek)) import Kafka.Consumer.Types (Timestamp (..))-import Kafka.Internal.RdKafka (RdKafkaMessageT (..), RdKafkaMessageTPtr, RdKafkaRespErrT (..), RdKafkaTimestampTypeT (..), Word8Ptr, rdKafkaErrno2err, rdKafkaMessageTimestamp, rdKafkaPoll, rdKafkaTopicName)+import Kafka.Internal.RdKafka (RdKafkaMessageT (..), RdKafkaMessageTPtr, RdKafkaRespErrT (..), RdKafkaTimestampTypeT (..), Word8Ptr, rdKafkaErrno2err, rdKafkaMessageTimestamp, rdKafkaPoll, rdKafkaTopicName, rdKafkaHeaderGetAll, rdKafkaMessageHeaders) import Kafka.Internal.Setup (HasKafka (..), Kafka (..))-import Kafka.Types (KafkaError (..), Millis (..), Timeout (..))+import Kafka.Types (KafkaError (..), Millis (..), Timeout (..), Headers, headersFromList) pollEvents :: HasKafka a => a -> Maybe Timeout -> IO () pollEvents a tm =- let timeout = maybe 0 (\(Timeout ms) -> ms) tm- (Kafka k) = getKafka a+ let timeout = maybe 0 unTimeout tm+ Kafka k = getKafka a in void (rdKafkaPoll k timeout) word8PtrToBS :: Int -> Word8Ptr -> IO BS.ByteString@@ -101,6 +104,30 @@ RdKafkaTimestampCreateTime -> CreateTime (Millis ts) RdKafkaTimestampLogAppendTime -> LogAppendTime (Millis ts) RdKafkaTimestampNotAvailable -> NoTimestamp+++readHeaders :: Ptr RdKafkaMessageT -> IO (Either RdKafkaRespErrT Headers)+readHeaders msg = do+ (err, headersPtr) <- rdKafkaMessageHeaders msg+ case err of+ RdKafkaRespErrNoent -> return $ Right mempty+ RdKafkaRespErrNoError -> fmap headersFromList <$> extractHeaders headersPtr+ e -> return . Left $ e+ where extractHeaders ptHeaders =+ alloca $ \nptr ->+ alloca $ \vptr ->+ alloca $ \szptr ->+ let go acc idx = rdKafkaHeaderGetAll ptHeaders idx nptr vptr szptr >>= \case+ RdKafkaRespErrNoent -> return $ Right acc+ RdKafkaRespErrNoError -> do+ cstr <- peek nptr+ wptr <- peek vptr+ csize <- peek szptr+ hn <- BS.packCString cstr+ hv <- word8PtrToBS (fromIntegral csize) wptr+ go ((hn, hv) : acc) (idx + 1)+ _ -> error "Unexpected error code while extracting headers"+ in go [] 0 readBS :: (t -> Int) -> (t -> Ptr Word8) -> t -> IO (Maybe BS.ByteString) readBS flen fdata s = if fdata s == nullPtr
src/Kafka/Metadata.hs view
@@ -1,5 +1,10 @@ {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE OverloadedStrings #-}++-----------------------------------------------------------------------------+-- |+-- Module with metadata types and functions.+----------------------------------------------------------------------------- module Kafka.Metadata ( KafkaMetadata(..), BrokerMetadata(..), TopicMetadata(..), PartitionMetadata(..) , WatermarkOffsets(..)@@ -19,7 +24,6 @@ import Control.Monad.IO.Class (MonadIO (liftIO)) import Data.Bifunctor (bimap) import Data.ByteString (ByteString, pack)-import Data.Monoid ((<>)) import Data.Text (Text) import Foreign (Storable (peek), peekArray, withForeignPtr) import GHC.Generics (Generic)@@ -256,8 +260,8 @@ { pmPartitionId = PartitionId (id'RdKafkaMetadataPartitionT pm) , pmError = kafkaErrorToMaybe $ KafkaResponseError (err'RdKafkaMetadataPartitionT pm) , pmLeader = BrokerId (leader'RdKafkaMetadataPartitionT pm)- , pmReplicas = (BrokerId . fromIntegral) <$> reps- , pmInSyncReplicas = (BrokerId . fromIntegral) <$> isrs+ , pmReplicas = BrokerId . fromIntegral <$> reps+ , pmInSyncReplicas = BrokerId . fromIntegral <$> isrs } @@ -292,4 +296,4 @@ "Stable" -> GroupStable "Dead" -> GroupDead "Empty" -> GroupEmpty- _ -> error $ "Unknown group state: " <> (Text.unpack s)+ _ -> error $ "Unknown group state: " <> Text.unpack s
src/Kafka/Producer.hs view
@@ -1,44 +1,97 @@-{-# LANGUAGE TupleSections #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE LambdaCase #-}++-----------------------------------------------------------------------------+-- |+-- Module to produce messages to Kafka topics.+-- +-- Here's an example of code to produce messages to a topic:+-- +-- @+-- import Control.Exception (bracket)+-- import Control.Monad (forM_)+-- import Data.ByteString (ByteString)+-- import Kafka.Producer+-- +-- -- Global producer properties+-- producerProps :: 'ProducerProperties'+-- producerProps = 'brokersList' ["localhost:9092"]+-- <> 'logLevel' 'KafkaLogDebug'+-- +-- -- Topic to send messages to+-- targetTopic :: 'TopicName'+-- targetTopic = 'TopicName' "kafka-client-example-topic"+-- +-- -- Run an example+-- runProducerExample :: IO ()+-- runProducerExample =+-- bracket mkProducer clProducer runHandler >>= print+-- where+-- mkProducer = 'newProducer' producerProps+-- clProducer (Left _) = pure ()+-- clProducer (Right prod) = 'closeProducer' prod+-- runHandler (Left err) = pure $ Left err+-- runHandler (Right prod) = sendMessages prod+-- +-- -- Example sending 2 messages and printing the response from Kafka+-- sendMessages :: 'KafkaProducer' -> IO (Either 'KafkaError' ())+-- sendMessages prod = do+-- err1 <- 'produceMessage' prod (mkMessage Nothing (Just "test from producer") )+-- forM_ err1 print+-- +-- err2 <- 'produceMessage' prod (mkMessage (Just "key") (Just "test from producer (with key)"))+-- forM_ err2 print+-- +-- pure $ Right ()+-- +-- mkMessage :: Maybe ByteString -> Maybe ByteString -> 'ProducerRecord'+-- mkMessage k v = 'ProducerRecord'+-- { 'prTopic' = targetTopic+-- , 'prPartition' = 'UnassignedPartition'+-- , 'prKey' = k+-- , 'prValue' = v+-- }+-- @+----------------------------------------------------------------------------- module Kafka.Producer-( module X+( KafkaProducer+, module X , runProducer , newProducer-, produceMessage, produceMessageBatch+, produceMessage+, produceMessage' , flushProducer , closeProducer-, KafkaProducer , RdKafkaRespErrT (..) ) where -import Control.Arrow ((&&&)) import Control.Exception (bracket)-import Control.Monad (forM, forM_, (<=<))+import Control.Monad (forM_) import Control.Monad.IO.Class (MonadIO (liftIO)) import qualified Data.ByteString as BS import qualified Data.ByteString.Internal as BSI-import Data.Function (on)-import Data.List (groupBy, sortBy)-import Data.Ord (comparing) import qualified Data.Text as Text-import Foreign.ForeignPtr (newForeignPtr_, withForeignPtr)-import Foreign.Marshal.Array (withArrayLen)+import Foreign.C.String (withCString)+import Foreign.ForeignPtr (withForeignPtr)+import Foreign.Marshal.Utils (withMany) import Foreign.Ptr (Ptr, nullPtr, plusPtr)-import Foreign.Storable (Storable (..))-import Kafka.Internal.RdKafka (RdKafkaMessageT (..), RdKafkaRespErrT (..), RdKafkaTypeT (..), destroyUnmanagedRdKafkaTopic, newRdKafkaT, newUnmanagedRdKafkaTopicT, rdKafkaOutqLen, rdKafkaProduce, rdKafkaProduceBatch, rdKafkaSetLogLevel)-import Kafka.Internal.Setup (Kafka (..), KafkaConf (..), KafkaProps (..), TopicConf (..), TopicProps (..), kafkaConf, topicConf)+import Foreign.StablePtr (newStablePtr, castStablePtrToPtr)+import Kafka.Internal.RdKafka (RdKafkaRespErrT (..), RdKafkaTypeT (..), RdKafkaVuT(..), newRdKafkaT, rdKafkaErrorCode, rdKafkaErrorDestroy, rdKafkaOutqLen, rdKafkaMessageProduceVa, rdKafkaSetLogLevel)+import Kafka.Internal.Setup (Kafka (..), KafkaConf (..), KafkaProps (..), TopicProps (..), kafkaConf, topicConf, Callback(..)) import Kafka.Internal.Shared (pollEvents)-import Kafka.Producer.Convert (copyMsgFlags, handleProduceErr, producePartitionCInt, producePartitionInt)+import Kafka.Producer.Convert (copyMsgFlags, handleProduceErrT, producePartitionCInt) import Kafka.Producer.Types (KafkaProducer (..)) + import Kafka.Producer.ProducerProperties as X import Kafka.Producer.Types as X hiding (KafkaProducer) import Kafka.Types as X -- | Runs Kafka Producer. -- The callback provided is expected to call 'produceMessage'--- or/and 'produceMessageBatch' to send messages to Kafka.-{-# DEPRECATED runProducer "Use newProducer/closeProducer instead" #-}+-- to send messages to Kafka.+{-# DEPRECATED runProducer "Use 'newProducer'/'closeProducer' instead" #-} runProducer :: ProducerProperties -> (KafkaProducer -> IO (Either KafkaError a)) -> IO (Either KafkaError a)@@ -60,8 +113,12 @@ kc@(KafkaConf kc' _ _) <- kafkaConf (KafkaProps $ (ppKafkaProps pps)) tc <- topicConf (TopicProps $ (ppTopicProps pps)) + -- add default delivery report callback+ let Callback setDeliveryCallback = deliveryCallback (const mempty)+ setDeliveryCallback kc+ -- set callbacks- forM_ (ppCallbacks pps) (\setCb -> setCb kc)+ forM_ (ppCallbacks pps) (\(Callback setCb) -> setCb kc) mbKafka <- newRdKafkaT RdKafkaProducer kc' case mbKafka of@@ -78,75 +135,47 @@ => KafkaProducer -> ProducerRecord -> m (Maybe KafkaError)-produceMessage kp@(KafkaProducer (Kafka k) _ (TopicConf tc)) m = liftIO $ do- pollEvents kp (Just $ Timeout 0) -- fire callbacks if any exist (handle delivery reports)- bracket (mkTopic $ prTopic m) clTopic withTopic- where- mkTopic (TopicName tn) = newUnmanagedRdKafkaTopicT k (Text.unpack tn) (Just tc)-- clTopic = either (return . const ()) destroyUnmanagedRdKafkaTopic-- withTopic (Left err) = return . Just . KafkaError $ Text.pack err- withTopic (Right t) =- withBS (prValue m) $ \payloadPtr payloadLength ->- withBS (prKey m) $ \keyPtr keyLength ->- handleProduceErr =<<- rdKafkaProduce t (producePartitionCInt (prPartition m))- copyMsgFlags payloadPtr (fromIntegral payloadLength)- keyPtr (fromIntegral keyLength) nullPtr----- | Sends a batch of messages.--- Returns a list of messages which it was unable to send with corresponding errors.--- Since librdkafka is backed by a queue, this function can return before messages are sent. See--- 'flushProducer' to wait for queue to empty.-produceMessageBatch :: MonadIO m- => KafkaProducer- -> [ProducerRecord]- -> m [(ProducerRecord, KafkaError)]- -- ^ An empty list when the operation is successful,- -- otherwise a list of "failed" messages with corresponsing errors.-produceMessageBatch kp@(KafkaProducer (Kafka k) _ (TopicConf tc)) messages = liftIO $ do- pollEvents kp (Just $ Timeout 0) -- fire callbacks if any exist (handle delivery reports)- concat <$> forM (mkBatches messages) sendBatch+produceMessage kp m = produceMessage' kp m (pure . mempty) >>= adjustRes where- mkSortKey = prTopic &&& prPartition- mkBatches = groupBy ((==) `on` mkSortKey) . sortBy (comparing mkSortKey)-- mkTopic (TopicName tn) = newUnmanagedRdKafkaTopicT k (Text.unpack tn) (Just tc)-- clTopic = either (return . const ()) destroyUnmanagedRdKafkaTopic+ adjustRes = \case+ Right () -> pure Nothing+ Left (ImmediateError err) -> pure (Just err) - sendBatch [] = return []- sendBatch batch = bracket (mkTopic $ prTopic (head batch)) clTopic (withTopic batch)+-- | Sends a single message with a registered callback.+--+-- The callback can be a long running process, as it is forked by the thread+-- that handles the delivery reports.+produceMessage' :: MonadIO m+ => KafkaProducer+ -> ProducerRecord+ -> (DeliveryReport -> IO ())+ -> m (Either ImmediateError ())+produceMessage' kp@(KafkaProducer (Kafka k) _ _) msg cb = liftIO $+ fireCallbacks >> produceIt+ where+ fireCallbacks =+ pollEvents kp . Just . Timeout $ 0 - withTopic ms (Left err) = return $ (, KafkaError (Text.pack err)) <$> ms- withTopic ms (Right t) = do- let (partInt, partCInt) = (producePartitionInt &&& producePartitionCInt) $ prPartition (head ms)- withForeignPtr t $ \topicPtr -> do- nativeMs <- forM ms (toNativeMessage topicPtr partInt)- withArrayLen nativeMs $ \len batchPtr -> do- batchPtrF <- newForeignPtr_ batchPtr- numRet <- rdKafkaProduceBatch t partCInt copyMsgFlags batchPtrF len- if numRet == len then return []- else do- errs <- mapM (return . err'RdKafkaMessageT <=< peekElemOff batchPtr)- [0..(fromIntegral $ len - 1)]- return [(m, KafkaResponseError e) | (m, e) <- zip messages errs, e /= RdKafkaRespErrNoError]+ produceIt =+ withBS (prValue msg) $ \payloadPtr payloadLength ->+ withBS (prKey msg) $ \keyPtr keyLength ->+ withHeaders (prHeaders msg) $ \hdrs ->+ withCString (Text.unpack . unTopicName . prTopic $ msg) $ \topicName -> do+ callbackPtr <- newStablePtr cb+ let opts = [+ Topic'RdKafkaVu topicName+ , Partition'RdKafkaVu . producePartitionCInt . prPartition $ msg+ , MsgFlags'RdKafkaVu (fromIntegral copyMsgFlags)+ , Value'RdKafkaVu payloadPtr (fromIntegral payloadLength)+ , Key'RdKafkaVu keyPtr (fromIntegral keyLength)+ , Opaque'RdKafkaVu (castStablePtrToPtr callbackPtr)+ ] - toNativeMessage t p m =- withBS (prValue m) $ \payloadPtr payloadLength ->- withBS (prKey m) $ \keyPtr keyLength ->- return RdKafkaMessageT- { err'RdKafkaMessageT = RdKafkaRespErrNoError- , topic'RdKafkaMessageT = t- , partition'RdKafkaMessageT = p- , len'RdKafkaMessageT = payloadLength- , payload'RdKafkaMessageT = payloadPtr- , offset'RdKafkaMessageT = 0- , keyLen'RdKafkaMessageT = keyLength- , key'RdKafkaMessageT = keyPtr- }+ code <- bracket (rdKafkaMessageProduceVa k (hdrs ++ opts)) rdKafkaErrorDestroy rdKafkaErrorCode+ res <- handleProduceErrT code+ pure $ case res of+ Just err -> Left . ImmediateError $ err+ Nothing -> Right () -- | Closes the producer. -- Will wait until the outbound queue is drained before returning the control.@@ -163,8 +192,15 @@ if (l == 0) then pollEvents kp (Just $ Timeout 0) -- to be sure that all the delivery reports are fired else flushProducer kp- ------------------------------------------------------------------------------------++withHeaders :: Headers -> ([RdKafkaVuT] -> IO a) -> IO a+withHeaders hds = withMany allocHeader (headersToList hds)+ where+ allocHeader (nm, val) f = + BS.useAsCString nm $ \cnm ->+ withBS (Just val) $ \vp vl ->+ f $ Header'RdKafkaVu cnm vp (fromIntegral vl) withBS :: Maybe BS.ByteString -> (Ptr a -> Int -> IO b) -> IO b withBS Nothing f = f nullPtr 0
src/Kafka/Producer/Callbacks.hs view
@@ -1,23 +1,36 @@+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE LambdaCase #-} module Kafka.Producer.Callbacks ( deliveryCallback , module X ) where +import Control.Monad (void)+import Control.Exception (bracket)+import Control.Concurrent (forkIO) import Foreign.C.Error (getErrno) import Foreign.Ptr (Ptr, nullPtr) import Foreign.Storable (Storable(peek))+import Foreign.StablePtr (castPtrToStablePtr, deRefStablePtr, freeStablePtr) import Kafka.Callbacks as X import Kafka.Consumer.Types (Offset(..)) import Kafka.Internal.RdKafka (RdKafkaMessageT(..), RdKafkaRespErrT(..), rdKafkaConfSetDrMsgCb)-import Kafka.Internal.Setup (KafkaConf(..), getRdKafkaConf)-import Kafka.Internal.Shared (kafkaRespErr, readTopic, readKey, readPayload)+import Kafka.Internal.Setup (getRdKafkaConf, Callback(..))+import Kafka.Internal.Shared (kafkaRespErr, readTopic, readKey, readPayload, readHeaders) import Kafka.Producer.Types (ProducerRecord(..), DeliveryReport(..), ProducePartition(..)) import Kafka.Types (KafkaError(..), TopicName(..))+import Data.Either (fromRight) -- | Sets the callback for delivery reports.-deliveryCallback :: (DeliveryReport -> IO ()) -> KafkaConf -> IO ()-deliveryCallback callback kc = rdKafkaConfSetDrMsgCb (getRdKafkaConf kc) realCb+--+-- /Note: A callback should not be a long-running process as it blocks+-- librdkafka from continuing on the thread that handles the delivery+-- callbacks. For callbacks to individual messsages see+-- 'Kafka.Producer.produceMessage\''./+--+deliveryCallback :: (DeliveryReport -> IO ()) -> Callback+deliveryCallback callback = Callback $ \kc -> rdKafkaConfSetDrMsgCb (getRdKafkaConf kc) realCb where realCb :: t -> Ptr RdKafkaMessageT -> IO () realCb _ mptr =@@ -25,28 +38,40 @@ then getErrno >>= (callback . NoMessageError . kafkaRespErr) else do s <- peek mptr- if err'RdKafkaMessageT s /= RdKafkaRespErrNoError- then mkErrorReport s >>= callback- else mkSuccessReport s >>= callback+ prodRec <- mkProdRec mptr+ let cbPtr = opaque'RdKafkaMessageT s+ callbacks cbPtr $ + if err'RdKafkaMessageT s /= RdKafkaRespErrNoError+ then mkErrorReport s prodRec + else mkSuccessReport s prodRec -mkErrorReport :: RdKafkaMessageT -> IO DeliveryReport-mkErrorReport msg = do- prodRec <- mkProdRec msg- pure $ DeliveryFailure prodRec (KafkaResponseError (err'RdKafkaMessageT msg))+ callbacks cbPtr rep = do+ callback rep+ if cbPtr == nullPtr then+ pure ()+ else bracket (pure $ castPtrToStablePtr cbPtr) freeStablePtr $ \stablePtr -> do+ msgCb <- deRefStablePtr @(DeliveryReport -> IO ()) stablePtr+ -- Here we fork the callback since it might be a longer action and+ -- blocking here would block librdkafka from continuing its execution+ void . forkIO $ msgCb rep -mkSuccessReport :: RdKafkaMessageT -> IO DeliveryReport-mkSuccessReport msg = do- prodRec <- mkProdRec msg- pure $ DeliverySuccess prodRec (Offset $ offset'RdKafkaMessageT msg)+mkErrorReport :: RdKafkaMessageT -> ProducerRecord -> DeliveryReport+mkErrorReport msg prodRec = DeliveryFailure prodRec (KafkaResponseError (err'RdKafkaMessageT msg)) -mkProdRec :: RdKafkaMessageT -> IO ProducerRecord-mkProdRec msg = do- topic <- readTopic msg- key <- readKey msg- payload <- readPayload msg- pure ProducerRecord- { prTopic = TopicName topic- , prPartition = SpecifiedPartition (partition'RdKafkaMessageT msg)- , prKey = key- , prValue = payload- }+mkSuccessReport :: RdKafkaMessageT -> ProducerRecord -> DeliveryReport+mkSuccessReport msg prodRec = DeliverySuccess prodRec (Offset $ offset'RdKafkaMessageT msg)++mkProdRec :: Ptr RdKafkaMessageT -> IO ProducerRecord+mkProdRec pmsg = do+ msg <- peek pmsg + topic <- readTopic msg+ key <- readKey msg+ payload <- readPayload msg+ flip fmap (fromRight mempty <$> readHeaders pmsg) $ \headers -> + ProducerRecord+ { prTopic = TopicName topic+ , prPartition = SpecifiedPartition (partition'RdKafkaMessageT msg)+ , prKey = key+ , prValue = payload+ , prHeaders = headers+ }
src/Kafka/Producer/Convert.hs view
@@ -3,12 +3,14 @@ , producePartitionInt , producePartitionCInt , handleProduceErr+, handleProduceErr'+, handleProduceErrT ) where import Foreign.C.Error (getErrno) import Foreign.C.Types (CInt)-import Kafka.Internal.RdKafka (rdKafkaMsgFlagCopy)+import Kafka.Internal.RdKafka (RdKafkaRespErrT(..), rdKafkaMsgFlagCopy) import Kafka.Internal.Shared (kafkaRespErr) import Kafka.Types (KafkaError(..)) import Kafka.Producer.Types (ProducePartition(..))@@ -27,7 +29,19 @@ {-# INLINE producePartitionCInt #-} handleProduceErr :: Int -> IO (Maybe KafkaError)-handleProduceErr (- 1) = (Just . kafkaRespErr) <$> getErrno+handleProduceErr (- 1) = Just . kafkaRespErr <$> getErrno handleProduceErr 0 = return Nothing handleProduceErr _ = return $ Just KafkaInvalidReturnValue {-# INLINE handleProduceErr #-}++handleProduceErrT :: RdKafkaRespErrT -> IO (Maybe KafkaError)+handleProduceErrT RdKafkaRespErrUnknown = Just . kafkaRespErr <$> getErrno+handleProduceErrT RdKafkaRespErrNoError = return Nothing+handleProduceErrT e = return $ Just (KafkaResponseError e)+{-# INLINE handleProduceErrT #-}++handleProduceErr' :: Int -> IO (Either KafkaError ())+handleProduceErr' (- 1) = Left . kafkaRespErr <$> getErrno+handleProduceErr' 0 = return (Right ())+handleProduceErr' _ = return $ Left KafkaInvalidReturnValue+{-# INLINE handleProduceErr' #-}
src/Kafka/Producer/ProducerProperties.hs view
@@ -1,5 +1,9 @@ {-# LANGUAGE OverloadedStrings #-} +-----------------------------------------------------------------------------+-- |+-- Module with producer properties types and functions.+----------------------------------------------------------------------------- module Kafka.Producer.ProducerProperties ( ProducerProperties(..) , brokersList@@ -8,7 +12,9 @@ , compression , topicCompression , sendTimeout+, statisticsInterval , extraProps+, extraProp , suppressDisconnectLogs , extraTopicProps , debugOptions@@ -22,17 +28,16 @@ import Data.Map (Map) import qualified Data.Map as M import Data.Semigroup as Sem-import Kafka.Internal.Setup (KafkaConf(..))-import Kafka.Types (KafkaDebug(..), Timeout(..), KafkaCompressionCodec(..), KafkaLogLevel(..), BrokerAddress(..), kafkaDebugToText, kafkaCompressionCodecToText) +import Kafka.Types (KafkaDebug(..), Timeout(..), KafkaCompressionCodec(..), KafkaLogLevel(..), BrokerAddress(..), kafkaDebugToText, kafkaCompressionCodecToText, Millis(..)) import Kafka.Producer.Callbacks --- | Properties to create 'KafkaProducer'.+-- | Properties to create 'Kafka.Producer.Types.KafkaProducer'. data ProducerProperties = ProducerProperties { ppKafkaProps :: Map Text Text , ppTopicProps :: Map Text Text , ppLogLevel :: Maybe KafkaLogLevel- , ppCallbacks :: [KafkaConf -> IO ()]+ , ppCallbacks :: [Callback] } instance Sem.Semigroup ProducerProperties where@@ -52,12 +57,20 @@ mappend = (Sem.<>) {-# INLINE mappend #-} +-- | Set the <https://kafka.apache.org/documentation/#bootstrap.servers list of brokers> to contact to connect to the Kafka cluster. brokersList :: [BrokerAddress] -> ProducerProperties brokersList bs =- let bs' = Text.intercalate "," ((\(BrokerAddress x) -> x) <$> bs)+ let bs' = Text.intercalate "," (unBrokerAddress <$> bs) in extraProps $ M.fromList [("bootstrap.servers", bs')] -setCallback :: (KafkaConf -> IO ()) -> ProducerProperties+-- | Set the producer callback.+--+-- For examples of use, see:+--+-- * 'errorCallback'+-- * 'logCallback'+-- * 'statsCallback'+setCallback :: Callback -> ProducerProperties setCallback cb = mempty { ppCallbacks = [cb] } -- | Sets the logging level.@@ -65,23 +78,37 @@ logLevel :: KafkaLogLevel -> ProducerProperties logLevel ll = mempty { ppLogLevel = Just ll } +-- | Set the <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md compression.codec> for the producer. compression :: KafkaCompressionCodec -> ProducerProperties compression c = extraProps $ M.singleton "compression.codec" (kafkaCompressionCodecToText c) +-- | Set the <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md#topic-configuration-properties compression.codec> for the topic. topicCompression :: KafkaCompressionCodec -> ProducerProperties topicCompression c = extraTopicProps $ M.singleton "compression.codec" (kafkaCompressionCodecToText c) +-- | Set the <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md#topic-configuration-properties message.timeout.ms>. sendTimeout :: Timeout -> ProducerProperties sendTimeout (Timeout t) = extraTopicProps $ M.singleton "message.timeout.ms" (Text.pack $ show t) +-- | Set the <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md statistics.interval.ms> for the producer.+statisticsInterval :: Millis -> ProducerProperties+statisticsInterval (Millis t) =+ extraProps $ M.singleton "statistics.interval.ms" (Text.pack $ show t)+ -- | Any configuration options that are supported by /librdkafka/. -- The full list can be found <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here> extraProps :: Map Text Text -> ProducerProperties extraProps m = mempty { ppKafkaProps = m } +-- | Any configuration options that are supported by /librdkafka/.+-- The full list can be found <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here>+extraProp :: Text -> Text -> ProducerProperties+extraProp k v = mempty { ppKafkaProps = M.singleton k v }+{-# INLINE extraProp #-}+ -- | Suppresses producer disconnects logs. -- -- It might be useful to turn this off when interacting with brokers@@ -90,12 +117,12 @@ suppressDisconnectLogs = extraProps $ M.fromList [("log.connection.close", "false")] --- | Any configuration options that are supported by /librdkafka/.--- The full list can be found <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here>+-- | Any *topic* configuration options that are supported by /librdkafka/.+-- The full list can be found <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md#topic-configuration-properties here> extraTopicProps :: Map Text Text -> ProducerProperties extraTopicProps m = mempty { ppTopicProps = m } --- | Sets debug features for the producer+-- | Sets <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md debug> features for the producer -- Usually is used with 'logLevel'. debugOptions :: [KafkaDebug] -> ProducerProperties debugOptions [] = extraProps M.empty
src/Kafka/Producer/Types.hs view
@@ -1,10 +1,18 @@-{-# LANGUAGE DeriveDataTypeable #-}-{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++-----------------------------------------------------------------------------+-- |+-- Module holding producer types.+----------------------------------------------------------------------------- module Kafka.Producer.Types ( KafkaProducer(..) , ProducerRecord(..) , ProducePartition(..) , DeliveryReport(..)+, ImmediateError(..) ) where @@ -13,9 +21,11 @@ import GHC.Generics (Generic) import Kafka.Consumer.Types (Offset (..)) import Kafka.Internal.Setup (HasKafka (..), HasKafkaConf (..), HasTopicConf (..), Kafka (..), KafkaConf (..), TopicConf (..))-import Kafka.Types (KafkaError (..), TopicName (..))+import Kafka.Types (KafkaError (..), TopicName (..), Headers) --- | Main pointer to Kafka object, which contains our brokers+-- | The main type for Kafka message production, used e.g. to send messages.+--+-- Its constructor is intentionally not exposed, instead, one should used 'Kafka.Producer.newProducer' to acquire such a value. data KafkaProducer = KafkaProducer { kpKafkaPtr :: !Kafka , kpKafkaConf :: !KafkaConf@@ -40,15 +50,27 @@ , prPartition :: !ProducePartition , prKey :: Maybe ByteString , prValue :: Maybe ByteString+ , prHeaders :: !Headers } deriving (Eq, Show, Typeable, Generic) +-- | data ProducePartition =- SpecifiedPartition {-# UNPACK #-} !Int -- the partition number of the topic+ -- | The partition number of the topic+ SpecifiedPartition {-# UNPACK #-} !Int+ -- | Let the Kafka broker decide the partition | UnassignedPartition deriving (Show, Eq, Ord, Typeable, Generic) +-- | Data type representing an error that is caused by pre-flight conditions not being met+newtype ImmediateError = ImmediateError KafkaError+ deriving newtype (Eq, Show)++-- | The result of sending a message to the broker, useful for callbacks data DeliveryReport+ -- | The message was successfully sent at this offset = DeliverySuccess ProducerRecord Offset+ -- | The message could not be sent | DeliveryFailure ProducerRecord KafkaError+ -- | An error occurred, but /librdkafka/ did not attach any sent message | NoMessageError KafkaError deriving (Show, Eq, Generic)
+ src/Kafka/Transaction.hs view
@@ -0,0 +1,132 @@+-----------------------------------------------------------------------------+-- |+-- Module to work wih Kafkas transactional producers.+-- +-----------------------------------------------------------------------------+module Kafka.Transaction+( initTransactions+, beginTransaction+, commitTransaction+, abortTransaction++, commitOffsetMessageTransaction+-- , commitTransactionWithOffsets++, TxError+, getKafkaError+, kafkaErrorIsFatal+, kafkaErrorIsRetriable+, kafkaErrorTxnRequiresAbort+)+where++import Control.Monad.IO.Class (MonadIO (liftIO))+import Kafka.Internal.RdKafka (RdKafkaErrorTPtr, rdKafkaErrorDestroy, rdKafkaErrorIsFatal, rdKafkaErrorIsRetriable, rdKafkaErrorTxnRequiresAbort, rdKafkaErrorCode, rdKafkaInitTransactions, rdKafkaBeginTransaction, rdKafkaCommitTransaction, rdKafkaAbortTransaction, rdKafkaSendOffsetsToTransaction)+import Kafka.Internal.Setup (getRdKafka)+import Kafka.Producer.Convert (handleProduceErrT)+import Kafka.Producer+import Kafka.Consumer.Convert (toNativeTopicPartitionList, topicPartitionFromMessageForCommit)+import Kafka.Consumer++-------------------------------------------------------------------------------------+-- Tx API++data TxError = TxError + { txErrorKafka :: !KafkaError + , txErrorFatal :: !Bool+ , txErrorRetriable :: !Bool+ , txErrorTxnReqAbort :: !Bool+ } ++-- | Initialises Kafka for transactions +initTransactions :: MonadIO m + => KafkaProducer + -> Timeout + -> m (Maybe KafkaError)+initTransactions p (Timeout to) + = liftIO $ rdKafkaInitTransactions (getRdKafka p) to >>= rdKafkaErrorCode >>= handleProduceErrT++-- | Begins a new transaction+beginTransaction :: MonadIO m + => KafkaProducer + -> m (Maybe KafkaError)+beginTransaction p + = liftIO $ rdKafkaBeginTransaction (getRdKafka p) >>= rdKafkaErrorCode >>= handleProduceErrT++-- | Commits an existing transaction+-- Pre-condition: there exists an open transaction, created with beginTransaction+commitTransaction :: MonadIO m + => KafkaProducer + -> Timeout + -> m (Maybe TxError)+commitTransaction p (Timeout to) = liftIO $ rdKafkaCommitTransaction (getRdKafka p) to >>= toTxError++-- | Aborts an existing transaction+-- Pre-condition: there exists an open transaction, created with beginTransaction+abortTransaction :: MonadIO m + => KafkaProducer + -> Timeout + -> m (Maybe KafkaError)+abortTransaction p (Timeout to) + = liftIO $ do rdKafkaAbortTransaction (getRdKafka p) to >>= rdKafkaErrorCode >>= handleProduceErrT++-- | Commits the message's offset in the current transaction+-- Similar to Kafka.Consumer.commitOffsetMessage but within a transactional context+-- Pre-condition: there exists an open transaction, created with beginTransaction+commitOffsetMessageTransaction :: MonadIO m + => KafkaProducer + -> KafkaConsumer + -> ConsumerRecord k v+ -> Timeout + -> m (Maybe TxError)+commitOffsetMessageTransaction p c m (Timeout to) = liftIO $ do+ tps <- toNativeTopicPartitionList [topicPartitionFromMessageForCommit m]+ rdKafkaSendOffsetsToTransaction (getRdKafka p) (getRdKafka c) tps to >>= toTxError++-- -- | Commit offsets for all currently assigned partitions in the current transaction+-- -- Similar to Kafka.Consumer.commitAllOffsets but within a transactional context+-- -- Pre-condition: there exists an open transaction, created with beginTransaction+-- commitAllOffsetsTransaction :: MonadIO m +-- => KafkaProducer +-- -> KafkaConsumer +-- -> Timeout +-- -> m (Maybe TxError)+-- commitAllOffsetsTransaction p c (Timeout to) = liftIO $ do+-- -- TODO: this can't be right...+-- tps <- newForeignPtr_ nullPtr+-- rdKafkaSendOffsetsToTransaction (getRdKafka p) (getRdKafka c) tps to >>= toTxError+ +getKafkaError :: TxError -> KafkaError+getKafkaError = txErrorKafka++kafkaErrorIsFatal :: TxError -> Bool+kafkaErrorIsFatal = txErrorFatal++kafkaErrorIsRetriable :: TxError -> Bool+kafkaErrorIsRetriable = txErrorRetriable++kafkaErrorTxnRequiresAbort :: TxError -> Bool+kafkaErrorTxnRequiresAbort = txErrorTxnReqAbort++----------------------------------------------------------------------------------------------------+-- Implementation detail, used internally+toTxError :: RdKafkaErrorTPtr -> IO (Maybe TxError)+toTxError errPtr = do+ ret <- rdKafkaErrorCode errPtr >>= handleProduceErrT+ case ret of+ Nothing -> do+ -- NOTE: don't forget to free error structure, otherwise we are leaking memory!+ rdKafkaErrorDestroy errPtr+ pure Nothing+ Just ke -> do+ fatal <- rdKafkaErrorIsFatal errPtr+ retriable <- rdKafkaErrorIsRetriable errPtr+ reqAbort <- rdKafkaErrorTxnRequiresAbort errPtr+ -- NOTE: don't forget to free error structure, otherwise we are leaking memory!+ rdKafkaErrorDestroy errPtr+ pure $ Just $ TxError + { txErrorKafka = ke+ , txErrorFatal = fatal+ , txErrorRetriable = retriable+ , txErrorTxnReqAbort = reqAbort+ }
src/Kafka/Types.hs view
@@ -2,6 +2,11 @@ {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE OverloadedStrings #-}++-----------------------------------------------------------------------------+-- |+-- Module holding types shared by consumer and producer modules.+----------------------------------------------------------------------------- module Kafka.Types ( BrokerId(..) , PartitionId(..)@@ -16,6 +21,7 @@ , KafkaDebug(..) , KafkaCompressionCodec(..) , TopicType(..)+, Headers, headersFromList, headersToList , topicType , kafkaDebugToText , kafkaCompressionCodecToText@@ -24,40 +30,58 @@ import Control.Exception (Exception (..)) import Data.Int (Int64)+import Data.String (IsString) import Data.Text (Text, isPrefixOf) import Data.Typeable (Typeable) import GHC.Generics (Generic) import Kafka.Internal.RdKafka (RdKafkaRespErrT, rdKafkaErr2name, rdKafkaErr2str)+import qualified Data.ByteString as BS +-- | Kafka broker ID newtype BrokerId = BrokerId { unBrokerId :: Int } deriving (Show, Eq, Ord, Read, Generic) +-- | Topic partition ID newtype PartitionId = PartitionId { unPartitionId :: Int } deriving (Show, Eq, Read, Ord, Enum, Generic)++-- | A number of milliseconds, used to represent durations and timestamps newtype Millis = Millis { unMillis :: Int64 } deriving (Show, Read, Eq, Ord, Num, Generic)-newtype ClientId = ClientId { unClientId :: Text } deriving (Show, Eq, Ord, Generic)++-- | Client ID used by Kafka to better track requests+-- +-- See <https://kafka.apache.org/documentation/#client.id Kafka documentation on client ID>+newtype ClientId = ClientId+ { unClientId :: Text+ } deriving (Show, Eq, IsString, Ord, Generic)++-- | Batch size used for polling newtype BatchSize = BatchSize { unBatchSize :: Int } deriving (Show, Read, Eq, Ord, Num, Generic) +-- | Whether the topic is created by a user or by the system data TopicType = User -- ^ Normal topics that are created by user.- | System -- ^ Topics starting with "__" (__consumer_offsets, __confluent.support.metrics) are considered "system" topics+ | System -- ^ Topics starting with a double underscore "\__" (@__consumer_offsets@, @__confluent.support.metrics@, etc.) are considered "system" topics deriving (Show, Read, Eq, Ord, Generic) --- | Topic name to be consumed+-- | Topic name to consume/produce messages -- -- Wildcard (regex) topics are supported by the /librdkafka/ assignor: -- any topic name in the topics list that is prefixed with @^@ will -- be regex-matched to the full list of topics in the cluster and matching -- topics will be added to the subscription list.-newtype TopicName =- TopicName { unTopicName :: Text } -- ^ a simple topic name or a regex if started with @^@- deriving (Show, Eq, Ord, Read, Generic)+newtype TopicName = TopicName+ { unTopicName :: Text -- ^ a simple topic name or a regex if started with @^@+ } deriving (Show, Eq, Ord, IsString, Read, Generic) +-- | Deduce the type of a topic from its name, by checking if it starts with a double underscore "\__" topicType :: TopicName -> TopicType topicType (TopicName tn) = if "__" `isPrefixOf` tn then System else User {-# INLINE topicType #-} -- | Kafka broker address string (e.g. @broker1:9092@)-newtype BrokerAddress = BrokerAddress { unBrokerAddress :: Text } deriving (Show, Eq, Generic)+newtype BrokerAddress = BrokerAddress+ { unBrokerAddress :: Text+ } deriving (Show, Eq, IsString, Generic) -- | Timeout in milliseconds newtype Timeout = Timeout { unTimeout :: Int } deriving (Show, Eq, Read, Generic)@@ -68,8 +92,7 @@ KafkaLogNotice | KafkaLogInfo | KafkaLogDebug deriving (Show, Enum, Eq) ------ | Any Kafka errors+-- | All possible Kafka errors data KafkaError = KafkaError Text | KafkaInvalidReturnValue@@ -85,6 +108,7 @@ "[" ++ rdKafkaErr2name err ++ "] " ++ rdKafkaErr2str err displayException err = show err +-- | Available /librdkafka/ debug contexts data KafkaDebug = DebugGeneric | DebugBroker@@ -100,6 +124,9 @@ | DebugAll deriving (Eq, Show, Typeable, Generic) +-- | Convert a 'KafkaDebug' into its /librdkafka/ string equivalent.+--+-- This is used internally by the library but may be useful to some developers. kafkaDebugToText :: KafkaDebug -> Text kafkaDebugToText d = case d of DebugGeneric -> "generic"@@ -115,16 +142,34 @@ DebugFeature -> "feature" DebugAll -> "all" +-- | Compression codec used by a topic+--+-- See <https://kafka.apache.org/documentation/#compression.type Kafka documentation on compression codecs> data KafkaCompressionCodec = NoCompression | Gzip | Snappy | Lz4+ | Zstd deriving (Eq, Show, Typeable, Generic) +-- | Convert a 'KafkaCompressionCodec' into its /librdkafka/ string equivalent.+--+-- This is used internally by the library but may be useful to some developers. kafkaCompressionCodecToText :: KafkaCompressionCodec -> Text kafkaCompressionCodecToText c = case c of NoCompression -> "none" Gzip -> "gzip" Snappy -> "snappy"- Lz4 -> "lz4"+ Lz4 -> "lz4" + Zstd -> "zstd"++-- | Headers that might be passed along with a record+newtype Headers = Headers { unHeaders :: [(BS.ByteString, BS.ByteString)] } + deriving (Eq, Show, Semigroup, Monoid, Read, Typeable, Generic)++headersFromList :: [(BS.ByteString, BS.ByteString)] -> Headers+headersFromList = Headers++headersToList :: Headers -> [(BS.ByteString, BS.ByteString)]+headersToList = unHeaders
tests-it/Kafka/IntegrationSpec.hs view
@@ -1,25 +1,27 @@+{-# LANGUAGE LambdaCase #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} module Kafka.IntegrationSpec where -import Control.Monad (forM, forM_)-import Control.Monad.Loops-import qualified Data.ByteString as BS-import Data.Either-import Data.Map (fromList)-import Data.Monoid ((<>))--import Kafka.Consumer as C-import Kafka.Metadata as M-import Kafka.Producer as P-+import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)+import Control.Monad (forM, forM_, void)+import Control.Monad.Loops+import Data.Either+import Data.Map (fromList)+import qualified Data.Set as Set+import Data.Monoid ((<>))+import Kafka.Consumer+import Kafka.Metadata+import Kafka.Producer import Kafka.TestEnv import Test.Hspec -{-# ANN module ("HLint: ignore Redundant do" :: String) #-}+import qualified Data.ByteString as BS +{- HLINT ignore "Redundant do" -}+ spec :: Spec spec = do describe "Per-message commit" $ do@@ -113,11 +115,30 @@ res <- sendMessages (testMessages testTopic) prod res `shouldBe` Right () + it "sends messages with callback to test topic" $ \prod -> do+ var <- newEmptyMVar+ let+ msg = ProducerRecord+ { prTopic = "callback-topic"+ , prPartition = UnassignedPartition+ , prKey = Nothing+ , prValue = Just "test from producer"+ , prHeaders = mempty+ }++ res <- produceMessage' prod msg (putMVar var)+ res `shouldBe` Right ()+ callbackRes <- flushProducer prod *> takeMVar var+ callbackRes `shouldSatisfy` \case+ DeliverySuccess _ _ -> True+ DeliveryFailure _ _ -> False+ NoMessageError _ -> False+ specWithConsumer "Run consumer with async polling" (consumerProps <> groupId (makeGroupId "async")) runConsumerSpec specWithConsumer "Run consumer with sync polling" (consumerProps <> groupId (makeGroupId "sync") <> callbackPollMode CallbackPollModeSync) runConsumerSpec describe "Kafka.Consumer.BatchSpec" $ do- specWithConsumer "Batch consumer" (consumerProps <> groupId (ConsumerGroupId "batch-consumer")) $ do+ specWithConsumer "Batch consumer" (consumerProps <> groupId "batch-consumer") $ do it "should consume first batch" $ \k -> do res <- pollMessageBatch k (Timeout 1000) (BatchSize 5) length res `shouldBe` 5@@ -132,6 +153,23 @@ it "should consume empty batch when there are no messages" $ \k -> do res <- pollMessageBatch k (Timeout 1000) (BatchSize 50) length res `shouldBe` 0++ describe "Kafka.Headers.Spec" $ do+ let testHeaders = headersFromList [("a-header-name", "a-header-value"), ("b-header-name", "b-header-value")]++ specWithKafka "Headers consumer/producer" consumerProps $ do+ it "1. sends 2 messages to test topic enriched with headers" $ \(k, prod) -> do+ void $ receiveMessages k++ res <- sendMessagesWithHeaders (testMessages testTopic) testHeaders prod+ res `shouldBe` Right ()+ it "2. should receive 2 messages enriched with headers" $ \(k, _) -> do+ res <- receiveMessages k+ (length <$> res) `shouldBe` Right 2++ forM_ res $ \rcs ->+ forM_ rcs ((`shouldBe` Set.fromList (headersToList testHeaders)) . Set.fromList . headersToList . crHeaders)+ ---------------------------------------------------------------------------------------------------------------- data ReadState = Skip | Read@@ -152,14 +190,18 @@ testMessages :: TopicName -> [ProducerRecord] testMessages t =- [ ProducerRecord t UnassignedPartition Nothing (Just "test from producer")- , ProducerRecord t UnassignedPartition (Just "key") (Just "test from producer (with key)")+ [ ProducerRecord t UnassignedPartition Nothing (Just "test from producer") mempty+ , ProducerRecord t UnassignedPartition (Just "key") (Just "test from producer (with key)") mempty ] sendMessages :: [ProducerRecord] -> KafkaProducer -> IO (Either KafkaError ()) sendMessages msgs prod = Right <$> (forM_ msgs (produceMessage prod) >> flushProducer prod) +sendMessagesWithHeaders :: [ProducerRecord] -> Headers -> KafkaProducer -> IO (Either KafkaError ())+sendMessagesWithHeaders msgs hdrs prod =+ Right <$> (forM_ msgs (\msg -> produceMessage prod (msg {prHeaders = hdrs})) >> flushProducer prod)+ runConsumerSpec :: SpecWith KafkaConsumer runConsumerSpec = do it "should receive messages" $ \k -> do@@ -211,13 +253,14 @@ hasTopic `shouldBe` True it "should return topic metadata" $ \k -> do- res <- topicMetadata k (Timeout 1000) testTopic+ res <- topicMetadata k (Timeout 2000) testTopic res `shouldSatisfy` isRight- (length . kmBrokers) <$> res `shouldBe` Right 1- (length . kmTopics) <$> res `shouldBe` Right 1+ length . kmBrokers <$> res `shouldBe` Right 1+ length . kmTopics <$> res `shouldBe` Right 1 it "should describe all consumer groups" $ \k -> do- res <- allConsumerGroupsInfo k (Timeout 1000)+ res <- allConsumerGroupsInfo k (Timeout 2000)+ res `shouldSatisfy` isRight let groups = either (const []) (fmap giGroup) res let prefixedGroups = filter isTestGroupId groups let resLen = length prefixedGroups@@ -225,15 +268,15 @@ -- fmap giGroup <$> res `shouldBe` Right [testGroupId] it "should describe a given consumer group" $ \k -> do- res <- consumerGroupInfo k (Timeout 1000) testGroupId+ res <- consumerGroupInfo k (Timeout 2000) testGroupId fmap giGroup <$> res `shouldBe` Right [testGroupId] it "should describe non-existent consumer group" $ \k -> do- res <- consumerGroupInfo k (Timeout 1000) (ConsumerGroupId "does-not-exist")+ res <- consumerGroupInfo k (Timeout 2000) "does-not-exist" res `shouldBe` Right [] it "should read topic offsets for time" $ \k -> do- res <- topicOffsetsForTime k (Timeout 1000) (Millis 1904057189508) testTopic+ res <- topicOffsetsForTime k (Timeout 2000) (Millis 1904057189508) testTopic res `shouldSatisfy` isRight fmap tpOffset <$> res `shouldBe` Right [PartitionOffsetEnd] @@ -258,13 +301,13 @@ || x == Left (KafkaResponseError RdKafkaRespErrTimedOut)) it "should respect out-of-bound offsets (invalid offset)" $ \k -> do- res <- seek k (Timeout 1000) [TopicPartition testTopic (PartitionId 0) PartitionOffsetInvalid]+ res <- seek k (Timeout 2000) [TopicPartition testTopic (PartitionId 0) PartitionOffsetInvalid] res `shouldBe` Nothing msg <- pollMessage k (Timeout 1000) crOffset <$> msg `shouldBe` Right (Offset 0) it "should respect out-of-bound offsets (huge offset)" $ \k -> do- res <- seek k (Timeout 1000) [TopicPartition testTopic (PartitionId 0) (PartitionOffset 123456)]+ res <- seek k (Timeout 3000) [TopicPartition testTopic (PartitionId 0) (PartitionOffset 123456)] res `shouldBe` Nothing- msg <- pollMessage k (Timeout 1000)+ msg <- pollMessage k (Timeout 2000) crOffset <$> msg `shouldBe` Right (Offset 0)
tests-it/Kafka/TestEnv.hs view
@@ -24,12 +24,12 @@ brokerAddress :: BrokerAddress brokerAddress = unsafePerformIO $- (BrokerAddress . Text.pack) <$> getEnv "KAFKA_TEST_BROKER" `catch` \(_ :: SomeException) -> return "localhost:9092"+ BrokerAddress . Text.pack <$> getEnv "KAFKA_TEST_BROKER" `catch` \(_ :: SomeException) -> return "localhost:9092" {-# NOINLINE brokerAddress #-} testTopic :: TopicName testTopic = unsafePerformIO $- (TopicName . Text.pack) <$> getEnv "KAFKA_TEST_TOPIC" `catch` \(_ :: SomeException) -> return $ testPrefix <> "-topic"+ TopicName . Text.pack <$> getEnv "KAFKA_TEST_TOPIC" `catch` \(_ :: SomeException) -> return $ testPrefix <> "-topic" {-# NOINLINE testTopic #-} testGroupId :: ConsumerGroupId
tests/Kafka/Consumer/ConsumerRecordMapSpec.hs view
@@ -15,10 +15,11 @@ testRecord :: ConsumerRecord (Maybe Text) (Maybe Text) testRecord = ConsumerRecord- { crTopic = TopicName "some-topic"+ { crTopic = "some-topic" , crPartition = PartitionId 0 , crOffset = Offset 5 , crTimestamp = NoTimestamp+ , crHeaders = mempty , crKey = Just testKey , crValue = Just testValue }
tests/Kafka/Consumer/ConsumerRecordTraverseSpec.hs view
@@ -17,10 +17,11 @@ testRecord :: ConsumerRecord Text Text testRecord = ConsumerRecord- { crTopic = TopicName "some-topic"+ { crTopic = "some-topic" , crPartition = PartitionId 0 , crOffset = Offset 5 , crTimestamp = NoTimestamp+ , crHeaders = mempty , crKey = testKey , crValue = testValue }