hw-kafka-client 3.1.1 → 3.1.2

raw patch · 14 files changed

+193/−70 lines, 14 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• README.md +9/−12
• hw-kafka-client.cabal +8/−1
• src/Kafka/Callbacks.hs +5/−5
• src/Kafka/Consumer.hs +11/−13
• src/Kafka/Consumer/Callbacks.hs +2/−2
• src/Kafka/Consumer/ConsumerProperties.hs +39/−24
• src/Kafka/Consumer/Subscription.hs +6/−2
• src/Kafka/Consumer/Types.hs +7/−2
• src/Kafka/Dump.hs +5/−0
• src/Kafka/Metadata.hs +5/−0
• src/Kafka/Producer.hs +55/−2
• src/Kafka/Producer/ProducerProperties.hs +19/−4
• src/Kafka/Producer/Types.hs +16/−2
• src/Kafka/Types.hs +6/−1

@@ -13,9 +13,9 @@  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.  ### Consumer example @@ -36,8 +36,6 @@  ```haskell import Control.Exception (bracket)-import Data.Monoid ((<>))-import Kafka import Kafka.Consumer  -- Global consumer properties@@ -67,12 +65,11 @@ ------------------------------------------------------------------- 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 () ``` @@ -231,10 +228,10 @@ $ docker-compose up ``` -After that, tests can be run as usual:+After that, integration tests can switched on with using 'it' flag:  ```-$ cabal test --test-show-details=direct+$ cabal test --test-show-details=direct --flag it ```  ## Credits

@@ -1,7 +1,7 @@ cabal-version: 2.2  name:                   hw-kafka-client-version:                3.1.1+version:                3.1.2 synopsis:               Kafka bindings for Haskell description:            Apache Kafka bindings backed by the librdkafka C library.                         .@@ -32,6 +32,11 @@   manual: True   default: False +flag it+  description: Run integration tests+  manual: True+  default: False+ library   hs-source-dirs:       src   ghc-options:          -Wall@@ -103,6 +108,8 @@                       , text                       , transformers   build-tool-depends:   hspec-discover:hspec-discover+  if !(flag(it))+    buildable:          False   other-modules:        Kafka.IntegrationSpec                         Kafka.TestEnv   default-language:     Haskell2010

@@ -15,9 +15,9 @@ -- -- Basic usage: ----- > setCallback (errorCallback myErrorCallback)+-- > 'setCallback' ('errorCallback' myErrorCallback) -- >--- > myErrorCallback :: KafkaError -> String -> IO ()+-- > myErrorCallback :: 'KafkaError' -> String -> IO () -- > myErrorCallback kafkaError message = print $ show kafkaError <> "|" <> message errorCallback :: HasKafkaConf k => (KafkaError -> String -> IO ()) -> k -> IO () errorCallback callback k =@@ -30,9 +30,9 @@ -- -- Basic usage: ----- > setCallback (logCallback myLogCallback)+-- > 'setCallback' ('logCallback' myLogCallback) -- >--- > myLogCallback :: KafkaLogLevel -> String -> String -> IO ()+-- > myLogCallback :: 'KafkaLogLevel' -> String -> String -> IO () -- > myLogCallback level facility message = print $ show level <> "|" <> facility <> "|" <> message logCallback :: HasKafkaConf k => (KafkaLogLevel -> String -> String -> IO ()) -> k -> IO () logCallback callback k =@@ -45,7 +45,7 @@ -- -- Basic usage: ----- > setCallback (statsCallback myStatsCallback)+-- > 'setCallback' ('statsCallback' myStatsCallback) -- > -- > myStatsCallback :: String -> IO () -- > myStatsCallback stats = print $ show stats

@@ -10,8 +10,7 @@ --  -- @ -- import Control.Exception (bracket)--- import Data.Monoid ((<>))--- import Kafka+-- import Control.Monad (replicateM_) -- import Kafka.Consumer --  -- -- Global consumer properties@@ -33,21 +32,20 @@ --     print res --     where --       mkConsumer = 'newConsumer' consumerProps consumerSub---       clConsumer (Left err) = return (Left err)+--       clConsumer (Left err) = pure (Left err) --       clConsumer (Right kc) = (maybe (Right ()) Left) \<$\> 'closeConsumer' kc---       runHandler (Left err) = return (Left err)+--       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---     mapM_ (\_ -> do---                    msg1 <- 'pollMessage' kafka ('Timeout' 1000)---                    putStrLn $ "Message: " <> show msg1---                    err <- 'commitAllOffsets' 'OffsetCommit' kafka---                    putStrLn $ "Offsets: " <> maybe "Committed." show err---           ) [0 .. 10]---     return $ Right ()+--     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@@ -159,7 +157,7 @@ -- 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@@ -169,7 +167,7 @@   pollConsumerEvents c Nothing   mbq <- readIORef qr   case mbq of-    Nothing -> return [Left $ KafkaBadSpecification "userPolls is set when calling pollMessageBatch."]+    Nothing -> return [Left $ KafkaBadSpecification "Calling pollMessageBatch while CallbackPollMode is set to CallbackPollModeSync."]     Just q  -> rdKafkaConsumeBatchQueue q ms b >>= traverse fromMessagePtr  -- | Commit message's offset on broker for the message's partition.

@@ -31,10 +31,10 @@ -- | 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

@@ -1,5 +1,9 @@ {-# LANGUAGE OverloadedStrings #-} +-----------------------------------------------------------------------------+-- |+-- Module with consumer properties types and functions.+----------------------------------------------------------------------------- module Kafka.Consumer.ConsumerProperties ( ConsumerProperties(..) , CallbackPollMode(..)@@ -34,9 +38,17 @@  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@@ -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)    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,82 @@     , ("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)] +-- | Set the consumer callback.+--+-- For examples of use, see:+--+-- * 'errorCallback'+-- * 'logCallback'+-- * 'statsCallback' setCallback :: (KafkaConf -> IO ()) -> 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 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 }

@@ -1,5 +1,9 @@ {-# LANGUAGE OverloadedStrings #-} +-----------------------------------------------------------------------------+-- |+-- Module with subscription types and functions.+----------------------------------------------------------------------------- module Kafka.Consumer.Subscription ( Subscription(..) , topics@@ -26,8 +30,8 @@ -- @ -- consumerSub :: 'Subscription' -- consumerSub = 'topics' ['TopicName' "kafka-client-example-topic"]---         <> 'offsetReset' 'Earliest'---         <> 'extraSubscriptionProps' (fromList [("prop1", "value 1"), ("prop2", "value 2")])+--            <> 'offsetReset' 'Earliest'+--            <> 'extraSubscriptionProps' (fromList [("prop1", "value 1"), ("prop2", "value 2")]) -- @ data Subscription = Subscription (Set TopicName) (Map Text Text) 

@@ -1,5 +1,10 @@ {-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE DeriveGeneric      #-}++-----------------------------------------------------------------------------+-- |+-- Module holding consumer types.+----------------------------------------------------------------------------- module Kafka.Consumer.Types ( KafkaConsumer(..) , ConsumerGroupId(..)@@ -40,7 +45,7 @@  -- | The main type for Kafka consumption, used e.g. to poll and commit messages. -- --- Its constructor is intentionally not exposed, instead, one should used 'newConsumer' to acquire such a value.+-- 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@@ -207,7 +212,7 @@ 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'))

@@ -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

@@ -1,5 +1,10 @@ {-# LANGUAGE DeriveGeneric     #-} {-# LANGUAGE OverloadedStrings #-}++-----------------------------------------------------------------------------+-- |+-- Module with metadata types and functions.+----------------------------------------------------------------------------- module Kafka.Metadata ( KafkaMetadata(..), BrokerMetadata(..), TopicMetadata(..), PartitionMetadata(..) , WatermarkOffsets(..)

@@ -1,14 +1,67 @@ {-# 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' ['BrokerAddress' "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' , flushProducer , closeProducer-, KafkaProducer , RdKafkaRespErrT (..) ) where

@@ -1,5 +1,9 @@ {-# LANGUAGE OverloadedStrings #-} +-----------------------------------------------------------------------------+-- |+-- Module with producer properties types and functions.+----------------------------------------------------------------------------- module Kafka.Producer.ProducerProperties ( ProducerProperties(..) , brokersList@@ -27,7 +31,7 @@  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@@ -52,11 +56,19 @@   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)    in extraProps $ M.fromList [("bootstrap.servers", bs')] +-- | Set the producer callback.+--+-- For examples of use, see:+--+-- * 'errorCallback'+-- * 'logCallback'+-- * 'statsCallback' setCallback :: (KafkaConf -> IO ()) -> ProducerProperties setCallback cb = mempty { ppCallbacks = [cb] } @@ -65,14 +77,17 @@ 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)@@ -90,12 +105,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

@@ -2,6 +2,11 @@ {-# LANGUAGE DeriveGeneric              #-} {-# LANGUAGE DerivingStrategies         #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}++-----------------------------------------------------------------------------+-- |+-- Module holding producer types.+----------------------------------------------------------------------------- module Kafka.Producer.Types ( KafkaProducer(..) , ProducerRecord(..)@@ -18,7 +23,9 @@ import Kafka.Internal.Setup (HasKafka (..), HasKafkaConf (..), HasTopicConf (..), Kafka (..), KafkaConf (..), TopicConf (..)) import Kafka.Types          (KafkaError (..), TopicName (..)) --- | 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@@ -45,8 +52,11 @@   , prValue     :: Maybe ByteString   } 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) @@ -54,8 +64,12 @@ 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)

@@ -2,6 +2,11 @@ {-# LANGUAGE DeriveGeneric              #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE OverloadedStrings          #-}++-----------------------------------------------------------------------------+-- |+-- Module holding types shared by consumer and producer modules.+----------------------------------------------------------------------------- module Kafka.Types ( BrokerId(..) , PartitionId(..)@@ -52,7 +57,7 @@   | 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