hw-kafka-client 1.0.0 → 1.1.0
raw patch · 11 files changed
+267/−106 lines, 11 files
Files
- README.md +133/−0
- example/ConsumerExample.hs +13/−12
- example/ProducerExample.hs +21/−15
- hw-kafka-client.cabal +4/−1
- src/Kafka.hs +0/−8
- src/Kafka/Consumer.hs +2/−4
- src/Kafka/Consumer/ConsumerProperties.hs +10/−1
- src/Kafka/Consumer/Types.hs +4/−6
- src/Kafka/Producer.hs +69/−55
- src/Kafka/Producer/ProducerProperties.hs +8/−1
- src/Kafka/Types.hs +3/−3
+ README.md view
@@ -0,0 +1,133 @@+# kafka-client +[](https://circleci.com/gh/haskell-works/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+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)+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.++### Example:+A working consumer example can be found here: [ConsumerExample.hs](example/ConsumerExample.hs)++```Haskell+import Data.Monoid ((<>))+import Kafka+import Kafka.Consumer++-- Global consumer properties+consumerProps :: ConsumerProperties+consumerProps = consumerBrokersList [BrokerAddress "localhost:9092"]+ <> groupId (ConsumerGroupId "consumer_example_group")+ <> noAutoCommit+ <> consumerDebug [DebugAll]++-- Subscription to topics+consumerSub :: Subscription+consumerSub = topics [TopicName "kafka-client-example-topic"]+ <> offsetReset Earliest++-- Running an example+runConsumerExample :: IO ()+runConsumerExample = do+ res <- runConsumer consumerProps consumerSub processMessages+ print res++-------------------------------------------------------------------+processMessages :: KafkaConsumer -> IO (Either KafkaError ())+processMessages kafka = do+ mapM_ (\_ -> do+ msg1 <- pollMessage kafka (Timeout 1000)+ putStrLn $ "Message: " <> show msg1+ err <- commitAllOffsets kafka OffsetCommit+ putStrLn $ "Offsets: " <> maybe "Committed." show err+ ) [0 .. 10]+ return $ Right ()+```++# 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)++### Example++```Haskell+import Control.Monad (forM_)+import Kafka+import Kafka.Producer++-- Global producer properties+producerProps :: ProducerProperties+producerProps = producerBrokersList [BrokerAddress "localhost:9092"]++-- Topic to send messages to+targetTopic :: TopicName+targetTopic = TopicName "kafka-client-example-topic"++-- Run an example+runProducerExample :: IO ()+runProducerExample = do+ res <- runProducer producerProps sendMessages+ print res++sendMessages :: KafkaProducer -> IO (Either KafkaError ())+sendMessages prod = do+ err1 <- produceMessage prod ProducerRecord+ { prTopic = targetTopic+ , prPartition = UnassignedPartition+ , prKey = Nothing+ , prValue = Just "test from producer"+ }+ forM_ err1 print++ err2 <- produceMessage prod ProducerRecord+ { prTopic = targetTopic+ , prPartition = UnassignedPartition+ , prKey = Just "key"+ , prValue = Just "test from producer (with key)"+ }+ forM_ err2 print++ return $ Right ()+```++# Installation++## Installing librdkafka++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:++ git clone https://github.com/edenhill/librdkafka+ cd librdkafka+ ./configure+ make && sudo 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++## Installing Kafka++The full Kafka guide is at http://kafka.apache.org/documentation.html#quickstart++Alternatively `docker-compose` can be used to run Kafka locally inside a Docker container.+To run Kafka inside Docker:++```+$ docker-compose up+```
example/ConsumerExample.hs view
@@ -25,34 +25,35 @@ runConsumerExample = do print $ cpLogLevel consumerProps res <- runConsumer consumerProps consumerSub processMessages- print $ show res+ print res ------------------------------------------------------------------- processMessages :: KafkaConsumer -> IO (Either KafkaError ()) processMessages kafka = do mapM_ (\_ -> do msg1 <- pollMessage kafka (Timeout 1000)- print $ "Message: " <> show msg1+ putStrLn $ "Message: " <> show msg1 err <- commitAllOffsets OffsetCommit kafka- print $ "Offsets: " <> maybe "Committed." show err+ putStrLn $ "Offsets: " <> maybe "Committed." show err ) [0 :: Integer .. 10] return $ Right () printingRebalanceCallback :: KafkaConsumer -> KafkaError -> [TopicPartition] -> IO () printingRebalanceCallback k e ps = do- putStrLn "Rebalance callback!"- print ("Rebalance Error: " ++ show e)- mapM_ (print . show . (tpTopicName &&& tpPartition &&& tpOffset)) ps case e of- KafkaResponseError RdKafkaRespErrAssignPartitions ->- assign k ps >>= print . show- KafkaResponseError RdKafkaRespErrRevokePartitions ->- assign k [] >>= print . show- x -> print "UNKNOWN (and unlikely!)" >> print (show x)+ KafkaResponseError RdKafkaRespErrAssignPartitions -> do+ putStr "[Rebalance] Assign partitions: "+ mapM_ (print . (tpTopicName &&& tpPartition &&& tpOffset)) ps+ assign k ps >>= print+ KafkaResponseError RdKafkaRespErrRevokePartitions -> do+ putStr "[Rebalance] Revoke partitions: "+ mapM_ (print . (tpTopicName &&& tpPartition &&& tpOffset)) ps+ assign k [] >>= print+ x -> print "Rebalance: UNKNOWN (and unlikely!)" >> print x printingOffsetCallback :: KafkaConsumer -> KafkaError -> [TopicPartition] -> IO () printingOffsetCallback _ e ps = do putStrLn "Offsets callback!" print ("Offsets Error:" ++ show e)- mapM_ (print . show . (tpTopicName &&& tpPartition &&& tpOffset)) ps+ mapM_ (print . (tpTopicName &&& tpPartition &&& tpOffset)) ps
example/ProducerExample.hs view
@@ -6,6 +6,7 @@ import Control.Monad (forM_) import Data.Monoid import Kafka.Producer+import Data.ByteString (ByteString) -- Global producer properties producerProps :: ProducerProperties@@ -16,28 +17,33 @@ targetTopic :: TopicName targetTopic = TopicName "kafka-client-example-topic" +mkMessage :: Maybe ByteString -> Maybe ByteString -> ProducerRecord+mkMessage k v = ProducerRecord+ { prTopic = targetTopic+ , prPartition = UnassignedPartition+ , prKey = k+ , prValue = v+ }+ -- Run an example runProducerExample :: IO () runProducerExample = do res <- runProducer producerProps sendMessages- print $ show res+ print res -sendMessages :: KafkaProducer -> IO (Either KafkaError String)+sendMessages :: KafkaProducer -> IO (Either KafkaError ()) sendMessages prod = do- err1 <- produceMessage prod ProducerRecord- { prTopic = targetTopic- , prPartition = UnassignedPartition- , prKey = Nothing- , prValue = Just "test from producer"- }+ err1 <- produceMessage prod (mkMessage Nothing (Just "test from producer") ) forM_ err1 print - err2 <- produceMessage prod ProducerRecord- { prTopic = targetTopic- , prPartition = UnassignedPartition- , prKey = Just "key"- , prValue = Just "test from producer (with key)"- }+ err2 <- produceMessage prod (mkMessage (Just "key") (Just "test from producer (with key)")) forM_ err2 print - return $ Right "All done, Sir."+ errs <- produceMessageBatch prod+ [ mkMessage (Just "b-1") (Just "batch-1")+ , mkMessage (Just "b-2") (Just "batch-2")+ , mkMessage Nothing (Just "batch-3")+ ]++ forM_ errs (print . snd)+ return $ Right ()
hw-kafka-client.cabal view
@@ -1,5 +1,5 @@ name: hw-kafka-client-version: 1.0.0+version: 1.1.0 homepage: https://github.com/haskell-works/hw-kafka-client bug-reports: https://github.com/haskell-works/hw-kafka-client/issues license: MIT@@ -19,6 +19,9 @@ * Keyed and keyless messages producing/consuming . * Batch producing messages+++extra-source-files: README.md source-repository head type: git
src/Kafka.hs view
@@ -4,11 +4,3 @@ import Kafka.Consumer as X import Kafka.Producer as X------- -- | Sets library log level (noisiness) with respect to a kafka instance--- setLogLevel :: Kafka -> KafkaLogLevel -> IO ()--- setLogLevel (Kafka kptr _) level =--- rdKafkaSetLogLevel kptr (fromEnum level)
src/Kafka/Consumer.hs view
@@ -64,7 +64,6 @@ -> Subscription -> m (Either KafkaError KafkaConsumer) newConsumer cp (Subscription ts tp) = liftIO $ do- -- (Kafka kkk (KafkaConf kc)) <- newKafka (KafkaProps $ M.toList m) (KafkaConf kc) <- newConsumerConf cp tp' <- topicConf (TopicProps $ M.toList tp) _ <- setDefaultTopicConf kc tp'@@ -81,7 +80,7 @@ -- | Polls the next message from a subscription pollMessage :: MonadIO m => KafkaConsumer- -> Timeout -- ^ the timeout, in milliseconds (@10^3@ per second)+ -> Timeout -- ^ the timeout, in milliseconds -> m (Either KafkaError (ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString))) -- ^ Left on error or timeout, right for success pollMessage (KafkaConsumer k _) (Timeout ms) = liftIO $ rdKafkaConsumerPoll k (fromIntegral ms) >>= fromMessagePtr@@ -105,7 +104,6 @@ -- | Assigns specified partitions to a current consumer. -- Assigning an empty list means unassigning from all partitions that are currently assigned.--- See 'setRebalanceCallback' for more details. assign :: MonadIO m => KafkaConsumer -> [TopicPartition] -> m KafkaError assign (KafkaConsumer k _) ps = let pl = if null ps@@ -114,7 +112,7 @@ in liftIO $ KafkaResponseError <$> (pl >>= rdKafkaAssign k) --- | Closes the consumer and destroys it.+-- | Closes the consumer. closeConsumer :: MonadIO m => KafkaConsumer -> m (Maybe KafkaError) closeConsumer (KafkaConsumer k _) = liftIO $ (kafkaErrorToMaybe . KafkaResponseError) <$> rdKafkaConsumerClose k
src/Kafka/Consumer/ConsumerProperties.hs view
@@ -9,6 +9,7 @@ import qualified Data.Map as M import qualified Data.List as L +-- | Properties to create 'KafkaConsumer'. data ConsumerProperties = ConsumerProperties { cpProps :: Map String String , cpRebalanceCallback :: Maybe ReballanceCallback@@ -26,10 +27,12 @@ let bs' = L.intercalate "," ((\(BrokerAddress x) -> x) <$> bs) in extraConsumerProps $ M.fromList [("bootstrap.servers", bs')] +-- | Disables auto commit for the consumer noAutoCommit :: ConsumerProperties noAutoCommit = extraConsumerProps $ M.fromList [("enable.auto.commit", "false")] +-- | Consumer group id groupId :: ConsumerGroupId -> ConsumerProperties groupId (ConsumerGroupId cid) = extraConsumerProps $ M.fromList [("group.id", cid)]@@ -66,18 +69,24 @@ offsetsCommitCallback :: OffsetsCommitCallback -> ConsumerProperties offsetsCommitCallback cb = ConsumerProperties M.empty Nothing (Just cb) Nothing +-- | Sets the logging level.+-- Usually is used with 'consumerDebug' to configure which logs are needed. consumerLogLevel :: KafkaLogLevel -> ConsumerProperties consumerLogLevel ll = ConsumerProperties M.empty Nothing Nothing (Just ll) +-- | Sets the compression codec for the consumer. consumerCompression :: KafkaCompressionCodec -> ConsumerProperties consumerCompression c = extraConsumerProps $ M.singleton "compression.codec" (kafkaCompressionCodecToString c) +-- | Any configuration options that are supported by /librdkafka/.+-- The full list can be found <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here> extraConsumerProps :: Map String String -> ConsumerProperties extraConsumerProps m = ConsumerProperties m Nothing Nothing Nothing {-# INLINE extraConsumerProps #-} --- | Sets debug features for the consumer+-- | Sets debug features for the consumer.+-- Usually is used with 'consumerLogLevel'. consumerDebug :: [KafkaDebug] -> ConsumerProperties consumerDebug [] = extraConsumerProps M.empty consumerDebug d =
src/Kafka/Consumer/Types.hs view
@@ -47,13 +47,11 @@ , tpPartition :: PartitionId , tpOffset :: PartitionOffset } deriving (Show, Eq) --- | Represents /received/ messages from a Kafka broker (i.e. used in a consumer)+-- | Represents a /received/ message from Kafka (i.e. used in a consumer) data ConsumerRecord k v = ConsumerRecord- { crTopic :: !TopicName- -- | Kafka partition this message was received from- , crPartition :: !PartitionId- -- | Offset within the 'crPartition' Kafka partition- , crOffset :: !Offset+ { crTopic :: !TopicName -- ^ Kafka topic this message was received from+ , crPartition :: !PartitionId -- ^ Kafka partition this message was received from+ , crOffset :: !Offset -- ^ Offset within the 'crPartition' Kafka partition , crKey :: !k , crValue :: !v }
src/Kafka/Producer.hs view
@@ -1,14 +1,16 @@+{-# LANGUAGE TupleSections #-} module Kafka.Producer ( module X , runProducer , newProducer-, produceMessage+, produceMessage, produceMessageBatch , flushProducer , closeProducer , RDE.RdKafkaRespErrT (..) ) where +import Control.Arrow ((&&&)) import Control.Exception import Control.Monad import Control.Monad.IO.Class@@ -20,9 +22,9 @@ import Kafka.Internal.RdKafkaEnum import Kafka.Internal.Setup import Kafka.Producer.Convert--- import Data.Function (on)--- import Data.List (sortBy, groupBy)--- import Data.Ord (comparing)+import Data.Function (on)+import Data.List (sortBy, groupBy)+import Data.Ord (comparing) import qualified Kafka.Internal.RdKafkaEnum as RDE @@ -30,6 +32,9 @@ import Kafka.Producer.Types as X import Kafka.Producer.ProducerProperties as X +-- | Runs Kafka Producer.+-- The callback provided is expected to call 'produceMessage'+-- or/and 'produceMessageBatch' to send messages to Kafka. runProducer :: ProducerProperties -> (KafkaProducer -> IO (Either KafkaError a)) -> IO (Either KafkaError a)@@ -45,6 +50,7 @@ runHandler (Right prod) = f prod -- | Creates a new kafka producer+-- A newly created producer must be closed with 'closeProducer' function. newProducer :: MonadIO m => ProducerProperties -> m (Either KafkaError KafkaProducer) newProducer (ProducerProperties kp tp ll) = liftIO $ do (KafkaConf kc) <- kafkaConf (KafkaProps $ M.toList kp)@@ -56,7 +62,7 @@ forM_ ll (rdKafkaSetLogLevel kafka . fromEnum) return .Right $ KafkaProducer kafka kc tc --- | Produce a single message.+-- | Sends a single message. -- Since librdkafka is backed by a queue, this function can return before messages are sent. See -- 'flushProducer' to wait for queue to empty. produceMessage :: MonadIO m@@ -68,8 +74,7 @@ where mkTopic (TopicName tn) = newUnmanagedRdKafkaTopicT k tn tc - clTopic (Left _) = return ()- clTopic (Right t) = destroyUnmanagedRdKafkaTopic t+ clTopic = either (return . const ()) destroyUnmanagedRdKafkaTopic withTopic (Left err) = return . Just . KafkaError $ err withTopic (Right t) =@@ -79,58 +84,67 @@ rdKafkaProduce t (producePartitionCInt (prPartition m)) copyMsgFlags payloadPtr (fromIntegral payloadLength) keyPtr (fromIntegral keyLength) nullPtr------ let (topic, key, partition, payload) = keyAndPayload m--- in withBS (Just payload) $ \payloadPtr payloadLength ->--- withBS key $ \keyPtr keyLength ->--- handleProduceErr =<<--- rdKafkaProduce t (producePartitionCInt partition)--- copyMsgFlags payloadPtr (fromIntegral payloadLength)--- keyPtr (fromIntegral keyLength) nullPtr------ | Produce a batch of messages. Since librdkafka is backed by a queue, this function can return--- before messages are sent. See 'flushProducer' to wait for the queue to be empty.--- produceMessageBatch :: KafkaTopic -- ^ topic pointer--- -> [ProducerRecord] -- ^ list of messages to enqueue.--- -> IO [(ProducerRecord, KafkaError)] -- list of failed messages with their errors. This will be empty on success.--- produceMessageBatch (KafkaTopic t _ _) pms =--- concat <$> mapM sendBatch (partBatches pms)--- where--- partBatches msgs = (\x -> (pmPartition (head x), x)) <$> batches msgs--- batches = groupBy ((==) `on` pmPartition) . sortBy (comparing pmPartition)--- sendBatch (part, ms) = do--- msgs <- forM ms toNativeMessage--- let msgsCount = length msgs--- withArray msgs $ \batchPtr -> do--- batchPtrF <- newForeignPtr_ batchPtr--- numRet <- rdKafkaProduceBatch t (producePartitionCInt part) copyMsgFlags batchPtrF msgsCount--- if numRet == msgsCount then return []--- else do--- errs <- mapM (return . err'RdKafkaMessageT <=< peekElemOff batchPtr)--- [0..(fromIntegral $ msgsCount - 1)]--- return [(m, KafkaResponseError e) | (m, e) <- zip pms errs, e /= RdKafkaRespErrNoError]--- where--- toNativeMessage msg =--- let (key, partition, payload) = keyAndPayload msg--- in withBS (Just payload) $ \payloadPtr payloadLength ->--- withBS key $ \keyPtr keyLength ->--- withForeignPtr t $ \ptrTopic ->--- return RdKafkaMessageT--- { err'RdKafkaMessageT = RdKafkaRespErrNoError--- , topic'RdKafkaMessageT = ptrTopic--- , partition'RdKafkaMessageT = producePartitionInt partition--- , len'RdKafkaMessageT = payloadLength--- , payload'RdKafkaMessageT = payloadPtr--- , offset'RdKafkaMessageT = 0--- , keyLen'RdKafkaMessageT = keyLength--- , key'RdKafkaMessageT = keyPtr--- } ++-- | 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 (KafkaProducer k _ tc) messages = liftIO $+ concat <$> forM (mkBatches messages) sendBatch+ where+ mkSortKey = prTopic &&& prPartition+ mkBatches = groupBy ((==) `on` mkSortKey) . sortBy (comparing mkSortKey)++ mkTopic (TopicName tn) = newUnmanagedRdKafkaTopicT k tn tc++ clTopic = either (return . const ()) destroyUnmanagedRdKafkaTopic++ sendBatch [] = return []+ sendBatch batch = bracket (mkTopic $ prTopic (head batch)) clTopic (withTopic batch)++ withTopic ms (Left err) = return $ (, KafkaError 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]++ 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+ }++-- | Closes the producer.+-- Will wait until the outbound queue is drained before returning the control. closeProducer :: MonadIO m => KafkaProducer -> m () closeProducer = flushProducer --- | Drains the outbound queue for a producer. This function is called automatically at the end of--- 'runKafkaProducer' or 'runKafkaProducerConf' and usually doesn't need to be called directly.+-- | Drains the outbound queue for a producer.+-- This function is also called automatically when the producer is closed+-- with 'closeProducer' to ensure that all queued messages make it to Kafka. flushProducer :: MonadIO m => KafkaProducer -> m () flushProducer kp@(KafkaProducer k _ _) = liftIO $ do pollEvents k 100
src/Kafka/Producer/ProducerProperties.hs view
@@ -1,13 +1,13 @@ module Kafka.Producer.ProducerProperties where --- import Control.Monad import Data.Map (Map) import Kafka.Types import qualified Data.Map as M import qualified Data.List as L +-- | Properties to create 'KafkaProducer'. data ProducerProperties = ProducerProperties { ppKafkaProps :: Map String String , ppTopicProps :: Map String String@@ -24,6 +24,8 @@ let bs' = L.intercalate "," ((\(BrokerAddress x) -> x) <$> bs) in extraProducerProps $ M.fromList [("bootstrap.servers", bs')] +-- | Sets the logging level.+-- Usually is used with 'producerDebug' to configure which logs are needed. producerLogLevel :: KafkaLogLevel -> ProducerProperties producerLogLevel ll = ProducerProperties M.empty M.empty (Just ll) @@ -35,13 +37,18 @@ producerTopicCompression c = extraProducerTopicProps $ M.singleton "compression.codec" (kafkaCompressionCodecToString c) +-- | Any configuration options that are supported by /librdkafka/.+-- The full list can be found <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here> extraProducerProps :: Map String String -> ProducerProperties extraProducerProps m = ProducerProperties m M.empty Nothing +-- | Any configuration options that are supported by /librdkafka/.+-- The full list can be found <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here> extraProducerTopicProps :: Map String String -> ProducerProperties extraProducerTopicProps m = ProducerProperties M.empty m Nothing -- | Sets debug features for the producer+-- Usually is used with 'producerLogLevel'. producerDebug :: [KafkaDebug] -> ProducerProperties producerDebug [] = extraProducerProps M.empty producerDebug d =
src/Kafka/Types.hs view
@@ -8,13 +8,13 @@ -- | Topic name to be consumed ----- 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. newtype TopicName = TopicName String -- ^ a simple topic name or a regex if started with @^@- deriving (Show, Eq, Read)+ deriving (Show, Eq, Ord, Read) -- | Kafka broker address string (e.g. @broker1:9092@) newtype BrokerAddress = BrokerAddress String deriving (Show, Eq)@@ -22,7 +22,7 @@ -- | Timeout in milliseconds newtype Timeout = Timeout Int deriving (Show, Eq, Read) --- | Log levels for the RdKafkaLibrary used in 'setKafkaLogLevel'+-- | Log levels for /librdkafka/. data KafkaLogLevel = KafkaLogEmerg | KafkaLogAlert | KafkaLogCrit | KafkaLogErr | KafkaLogWarning | KafkaLogNotice | KafkaLogInfo | KafkaLogDebug