diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,133 @@ +# kafka-client +[![Circle CI](https://circleci.com/gh/haskell-works/kafka-client.svg?style=svg&circle-token=5f3ada2650dd600bc0fd4787143024867b2afc4e)](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 +``` diff --git a/example/ConsumerExample.hs b/example/ConsumerExample.hs --- a/example/ConsumerExample.hs +++ b/example/ConsumerExample.hs @@ -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 diff --git a/example/ProducerExample.hs b/example/ProducerExample.hs --- a/example/ProducerExample.hs +++ b/example/ProducerExample.hs @@ -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 () diff --git a/hw-kafka-client.cabal b/hw-kafka-client.cabal --- a/hw-kafka-client.cabal +++ b/hw-kafka-client.cabal @@ -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 diff --git a/src/Kafka.hs b/src/Kafka.hs --- a/src/Kafka.hs +++ b/src/Kafka.hs @@ -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) diff --git a/src/Kafka/Consumer.hs b/src/Kafka/Consumer.hs --- a/src/Kafka/Consumer.hs +++ b/src/Kafka/Consumer.hs @@ -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 diff --git a/src/Kafka/Consumer/ConsumerProperties.hs b/src/Kafka/Consumer/ConsumerProperties.hs --- a/src/Kafka/Consumer/ConsumerProperties.hs +++ b/src/Kafka/Consumer/ConsumerProperties.hs @@ -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 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 = diff --git a/src/Kafka/Consumer/Types.hs b/src/Kafka/Consumer/Types.hs --- a/src/Kafka/Consumer/Types.hs +++ b/src/Kafka/Consumer/Types.hs @@ -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 } diff --git a/src/Kafka/Producer.hs b/src/Kafka/Producer.hs --- a/src/Kafka/Producer.hs +++ b/src/Kafka/Producer.hs @@ -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 diff --git a/src/Kafka/Producer/ProducerProperties.hs b/src/Kafka/Producer/ProducerProperties.hs --- a/src/Kafka/Producer/ProducerProperties.hs +++ b/src/Kafka/Producer/ProducerProperties.hs @@ -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 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 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 = diff --git a/src/Kafka/Types.hs b/src/Kafka/Types.hs --- a/src/Kafka/Types.hs +++ b/src/Kafka/Types.hs @@ -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