hw-kafka-client 3.0.0 → 3.1.0

raw patch · 13 files changed

+336/−157 lines, 13 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• README.md +68/−11
• example/ConsumerExample.hs +0/−1
• example/ProducerExample.hs +38/−6
• hw-kafka-client.cabal +106/−113
• src/Kafka/Consumer.hs +9/−2
• src/Kafka/Consumer/Callbacks.hs +0/−1
• src/Kafka/Internal/RdKafka.chs +4/−1
• src/Kafka/Metadata.hs +0/−1
• src/Kafka/Producer.hs +52/−17
• src/Kafka/Producer/Callbacks.hs +23/−2
• src/Kafka/Producer/Convert.hs +7/−0
• src/Kafka/Producer/Types.hs +9/−2
• tests-it/Kafka/IntegrationSpec.hs +20/−0

@@ -19,20 +19,21 @@ group the set of consumers attempt to "rebalance" the load to assign partitions to each consumer.  ### Example:-```++```bash $ stack build --flag hw-kafka-client:examples ```  or -```+```bash $ stack build --exec kafka-client-example --flag hw-kafka-client: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@@ -75,12 +76,14 @@ ```  # 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,27 +97,26 @@ 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".+             <> 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)+             <> 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  ```Haskell@@ -163,6 +165,56 @@                   } ``` +### Synchronous sending of messages+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.++```haskell+produceMessage' :: MonadIO m+                => KafkaProducer+                -> ProducerRecord+                -> (DeliveryReport -> IO ())+                -> m (Either ImmediateError ())+```++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.++```haskell+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+```++_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.+ # Installation  ## Installing librdkafka@@ -171,16 +223,21 @@ the distribution packages are too old to support `kafka-client`. As such, we suggest you install from the source: +```bash     git clone https://github.com/edenhill/librdkafka     cd librdkafka     ./configure     make && make install+```  Sometimes it is helpful to specify openssl includes explicitly: +```     LDFLAGS=-L/usr/local/opt/openssl/lib CPPFLAGS=-I/usr/local/opt/openssl/include ./configure+```  If you are using Stack with Nix, don't forget to declare `rdkafka` as extra package:+ ```yaml # stack.yaml nix:@@ -196,6 +253,6 @@ Alternatively `docker-compose` can be used to run Kafka locally inside a Docker container. To run Kafka inside Docker: -```+```bash $ docker-compose up ```

@@ -6,7 +6,6 @@  import Control.Arrow     ((&&&)) import Control.Exception (bracket)-import Data.Monoid       ((<>)) import Kafka.Consumer import Data.Text         (Text) 

@@ -1,15 +1,18 @@ {-# 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@@ -69,3 +72,32 @@    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

@@ -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:                3.1.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.1, 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@@ -32,102 +33,94 @@   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.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+  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+  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+  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

@@ -5,7 +5,7 @@ ( module X , runConsumer , newConsumer-, assignment, subscription+, assign, assignment, subscription , pausePartitions, resumePartitions , committed, position, seek , pollMessage, pollConsumerEvents@@ -37,7 +37,7 @@ 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.RdKafka     (RdKafkaRespErrT (..), RdKafkaTopicPartitionListTPtr, RdKafkaTypeT (..), 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) import           Kafka.Internal.Shared      (kafkaErrorToMaybe, maybeToLeft, rdKafkaErrorToEither) @@ -162,6 +162,13 @@                  -> m (Maybe KafkaError) 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]))

@@ -8,7 +8,6 @@  import Control.Arrow          ((&&&)) import Control.Monad          (forM_, void)-import Data.Monoid            ((<>)) import Foreign.ForeignPtr     (newForeignPtr_) import Foreign.Ptr            (nullPtr) import Kafka.Callbacks        as X

@@ -147,6 +147,7 @@     , offset'RdKafkaMessageT :: Int64     , payload'RdKafkaMessageT :: Word8Ptr     , key'RdKafkaMessageT :: Word8Ptr+    , opaque'RdKafkaMessageT :: Ptr ()     }     deriving (Show, Eq) @@ -162,6 +163,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 +173,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 #} @@ -893,7 +896,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 ^

@@ -19,7 +19,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)

@@ -1,9 +1,11 @@-{-# LANGUAGE TupleSections #-}+{-# LANGUAGE TupleSections              #-}+{-# LANGUAGE LambdaCase                 #-} module Kafka.Producer ( module X , runProducer , newProducer , produceMessage, produceMessageBatch+, produceMessage' , flushProducer , closeProducer , KafkaProducer@@ -25,11 +27,12 @@ import           Foreign.Marshal.Array    (withArrayLen) import           Foreign.Ptr              (Ptr, nullPtr, plusPtr) import           Foreign.Storable         (Storable (..))+import           Foreign.StablePtr        (newStablePtr, castStablePtrToPtr) 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           Kafka.Internal.Shared    (pollEvents)-import           Kafka.Producer.Convert   (copyMsgFlags, handleProduceErr, producePartitionCInt, producePartitionInt)-import           Kafka.Producer.Types     (KafkaProducer (..))+import           Kafka.Producer.Convert   (copyMsgFlags, handleProduceErr', producePartitionCInt, producePartitionInt)+import           Kafka.Producer.Types     (KafkaProducer (..), ImmediateError(..))  import Kafka.Producer.ProducerProperties as X import Kafka.Producer.Types              as X hiding (KafkaProducer)@@ -60,6 +63,9 @@   kc@(KafkaConf kc' _ _) <- kafkaConf (KafkaProps $ (ppKafkaProps pps))   tc <- topicConf (TopicProps $ (ppTopicProps pps)) +  -- add default delivery report callback+  deliveryCallback (const mempty) kc+   -- set callbacks   forM_ (ppCallbacks pps) (\setCb -> setCb kc) @@ -78,24 +84,52 @@                => 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)+produceMessage kp m = produceMessage' kp m (pure . mempty) >>= adjustRes+  where+    adjustRes = \case+      Right () -> pure Nothing+      Left (ImmediateError err) -> pure (Just err) -      clTopic = either (return . const ()) destroyUnmanagedRdKafkaTopic+-- | 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) _ (TopicConf tc)) msg cb = liftIO $+  fireCallbacks >> bracket (mkTopic . prTopic $ msg) closeTopic withTopic+  where+    fireCallbacks =+      pollEvents kp . Just . Timeout $ 0 -      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+    mkTopic (TopicName tn) =+      newUnmanagedRdKafkaTopicT k (Text.unpack tn) (Just tc) +    closeTopic = either mempty destroyUnmanagedRdKafkaTopic +    withTopic (Left err) = return . Left . ImmediateError . KafkaError . Text.pack $ err+    withTopic (Right topic) =+      withBS (prValue msg) $ \payloadPtr payloadLength ->+        withBS (prKey msg) $ \keyPtr keyLength -> do+          callbackPtr <- newStablePtr cb+          res <- handleProduceErr' =<< rdKafkaProduce+            topic+            (producePartitionCInt (prPartition msg))+            copyMsgFlags+            payloadPtr+            (fromIntegral payloadLength)+            keyPtr+            (fromIntegral keyLength)+            (castStablePtrToPtr callbackPtr)++          pure $ case res of+            Left err -> Left . ImmediateError $ err+            Right () -> Right ()+ -- | 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@@ -146,6 +180,7 @@             , offset'RdKafkaMessageT    = 0             , keyLen'RdKafkaMessageT    = keyLength             , key'RdKafkaMessageT       = keyPtr+            , opaque'RdKafkaMessageT    = nullPtr             }  -- | Closes the producer.

@@ -1,12 +1,16 @@+{-# LANGUAGE TypeApplications #-} module Kafka.Producer.Callbacks ( deliveryCallback , module X ) where +import           Control.Monad          (void)+import           Control.Concurrent     (forkIO) import           Foreign.C.Error        (getErrno) import           Foreign.Ptr            (Ptr, nullPtr) import           Foreign.Storable       (Storable(peek))+import           Foreign.StablePtr      (castPtrToStablePtr, deRefStablePtr) import           Kafka.Callbacks        as X import           Kafka.Consumer.Types   (Offset(..)) import           Kafka.Internal.RdKafka (RdKafkaMessageT(..), RdKafkaRespErrT(..), rdKafkaConfSetDrMsgCb)@@ -16,6 +20,12 @@ import           Kafka.Types            (KafkaError(..), TopicName(..))  -- | Sets the callback for delivery reports.+--+--   /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 ()) -> KafkaConf -> IO () deliveryCallback callback kc = rdKafkaConfSetDrMsgCb (getRdKafkaConf kc) realCb   where@@ -25,9 +35,20 @@         then getErrno >>= (callback . NoMessageError . kafkaRespErr)         else do           s <- peek mptr+          let cbPtr = opaque'RdKafkaMessageT s           if err'RdKafkaMessageT s /= RdKafkaRespErrNoError-            then mkErrorReport s   >>= callback-            else mkSuccessReport s >>= callback+            then mkErrorReport s   >>= callbacks cbPtr+            else mkSuccessReport s >>= callbacks cbPtr++    callbacks cbPtr rep = do+      callback rep+      if cbPtr == nullPtr then+        pure ()+      else do+        msgCb <- deRefStablePtr @(DeliveryReport -> IO ()) $ castPtrToStablePtr $ cbPtr+        -- 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  mkErrorReport :: RdKafkaMessageT -> IO DeliveryReport mkErrorReport msg = do

@@ -3,6 +3,7 @@ , producePartitionInt , producePartitionCInt , handleProduceErr+, handleProduceErr' ) where @@ -31,3 +32,9 @@ handleProduceErr 0 = return Nothing handleProduceErr _ = return $ Just KafkaInvalidReturnValue {-# INLINE handleProduceErr #-}++handleProduceErr' :: Int -> IO (Either KafkaError ())+handleProduceErr' (- 1) = (Left . kafkaRespErr) <$> getErrno+handleProduceErr' 0 = return (Right ())+handleProduceErr' _ = return $ Left KafkaInvalidReturnValue+{-# INLINE handleProduceErr' #-}

@@ -1,10 +1,13 @@-{-# LANGUAGE DeriveDataTypeable #-}-{-# LANGUAGE DeriveGeneric      #-}+{-# LANGUAGE DeriveDataTypeable         #-}+{-# LANGUAGE DeriveGeneric              #-}+{-# LANGUAGE DerivingStrategies         #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} module Kafka.Producer.Types ( KafkaProducer(..) , ProducerRecord(..) , ProducePartition(..) , DeliveryReport(..)+, ImmediateError(..) ) where @@ -46,6 +49,10 @@     SpecifiedPartition {-# UNPACK #-} !Int  -- the partition number of the topic   | 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)  data DeliveryReport   = DeliverySuccess ProducerRecord Offset

@@ -1,3 +1,4 @@+{-# LANGUAGE LambdaCase          #-} {-# LANGUAGE OverloadedStrings   #-} {-# LANGUAGE ScopedTypeVariables #-} @@ -6,6 +7,7 @@  import           Control.Monad       (forM, forM_) import           Control.Monad.Loops+import           Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar) import qualified Data.ByteString     as BS import           Data.Either import           Data.Map            (fromList)@@ -112,6 +114,24 @@             it "sends messages to test topic" $ \prod -> do                 res    <- sendMessages (testMessages testTopic) prod                 res `shouldBe` Right ()++            it "sends messages with callback to test topic" $ \prod -> do+                var <- newEmptyMVar+                let+                  msg = ProducerRecord+                    { prTopic = TopicName "callback-topic"+                    , prPartition = UnassignedPartition+                    , prKey = Nothing+                    , prValue = Just "test from producer"+                    }++                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