diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -73,6 +73,41 @@ 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. + +However, it is not always possible for the producer to send messages to Kafka. Network problems +or Kafka cluster being offline can prevent the producer from doing it. + +When a message cannot be sent to Kafka for some time (see `message.timeout.ms` [configuration](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md) option), +the message is *dropped from the outgoing queue* and the *delivery report* indicating an error is raised. + +It is possible to configure `hw-kafka-client` to set an infinite message timeout so the message is +never dropped from the queue: + +``` +producerProps :: ProducerProperties +producerProps = brokersList [BrokerAddress "localhost:9092"] + <> sendTimeout (Timeout 0) -- for librdkafka "0" means "infinite". +``` + +*Delivery reports* provide the way to detect when producer experiences problems sending messages +to Kafka. + +Currently `hw-kafka-client` only supports delivery error callbacks: +``` +producerProps :: ProducerProperties +producerProps = brokersList [BrokerAddress "localhost:9092"] + <> setCallback (deliveryErrorsCallback 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 diff --git a/example/ConsumerExample.hs b/example/ConsumerExample.hs --- a/example/ConsumerExample.hs +++ b/example/ConsumerExample.hs @@ -39,19 +39,12 @@ ) [0 :: Integer .. 10] return $ Right () -printingRebalanceCallback :: KafkaConsumer -> KafkaError -> [TopicPartition] -> IO () -printingRebalanceCallback k e ps = - case e of - 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 - +printingRebalanceCallback :: KafkaConsumer -> RebalanceEvent -> IO () +printingRebalanceCallback _ e = case e of + RebalanceAssign ps -> + putStrLn $ "[Rebalance] Assign partitions: " <> show ps + RebalanceRevoke ps -> + putStrLn $ "[Rebalance] Revoke partitions: " <> show ps printingOffsetCallback :: KafkaConsumer -> KafkaError -> [TopicPartition] -> IO () printingOffsetCallback _ e ps = do 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: 2.2.0 +version: 2.3.0 homepage: https://github.com/haskell-works/hw-kafka-client bug-reports: https://github.com/haskell-works/hw-kafka-client/issues license: MIT diff --git a/src/Kafka/Callbacks.hs b/src/Kafka/Callbacks.hs --- a/src/Kafka/Callbacks.hs +++ b/src/Kafka/Callbacks.hs @@ -10,12 +10,10 @@ errorCallback :: HasKafkaConf k => (KafkaError -> String -> IO ()) -> k -> IO () errorCallback callback k = - let (KafkaConf c _) = getKafkaConf k - realCb _ err = callback (KafkaResponseError err) - in rdKafkaConfSetErrorCb c realCb + let realCb _ err = callback (KafkaResponseError err) + in rdKafkaConfSetErrorCb (getRdKafkaConf k) realCb logCallback :: HasKafkaConf k => (Int -> String -> String -> IO ()) -> k -> IO () logCallback callback k = - let (KafkaConf c _) = getKafkaConf k - realCb _ = callback - in rdKafkaConfSetLogCb c realCb + let realCb _ = callback + in rdKafkaConfSetLogCb (getRdKafkaConf k) realCb diff --git a/src/Kafka/Consumer.hs b/src/Kafka/Consumer.hs --- a/src/Kafka/Consumer.hs +++ b/src/Kafka/Consumer.hs @@ -3,13 +3,12 @@ ( module X , runConsumer , newConsumer -, assign, assignment, subscription +, assignment, subscription , pausePartitions, resumePartitions , committed, position, seek -, pollMessage +, pollMessage, pollConsumerEvents , commitOffsetMessage, commitAllOffsets, commitPartitionsOffsets , closeConsumer - -- ReExport Types , KafkaConsumer , RdKafkaRespErrT (..) @@ -17,13 +16,17 @@ where import Control.Arrow +import Control.Concurrent (forkIO, rtsSupportsBoundThreads) import Control.Exception -import Control.Monad (forM_) +import Control.Monad (forM_, void, when) import Control.Monad.IO.Class import Control.Monad.Trans.Except import Data.Bifunctor import qualified Data.ByteString as BS +import Data.IORef import qualified Data.Map as M +import Data.Maybe (fromMaybe) +import Data.Monoid ((<>)) import Foreign hiding (void) import Kafka.Consumer.Convert import Kafka.Consumer.Types @@ -32,9 +35,6 @@ import Kafka.Internal.Setup import Kafka.Internal.Shared - -import qualified Kafka.Consumer.Types as CIT - import Kafka.Consumer.ConsumerProperties as X import Kafka.Consumer.Subscription as X import Kafka.Consumer.Types as X hiding (KafkaConsumer) @@ -62,28 +62,38 @@ => ConsumerProperties -> Subscription -> m (Either KafkaError KafkaConsumer) -newConsumer cp (Subscription ts tp) = liftIO $ do - kc@(KafkaConf kc' ct) <- newConsumerConf cp +newConsumer props (Subscription ts tp) = liftIO $ do + let cp = setCallback (rebalanceCallback (\_ _ -> return ())) <> props + kc@(KafkaConf kc' qref ct) <- newConsumerConf cp tp' <- topicConf (TopicProps $ M.toList tp) _ <- setDefaultTopicConf kc tp' - rdk <- bimap KafkaError Kafka <$> newRdKafkaT RdKafkaConsumer kc' - case flip KafkaConsumer kc <$> rdk of - Left err -> return $ Left err - Right kafka -> do - forM_ (cpLogLevel cp) (setConsumerLogLevel kafka) - sub <- subscribe kafka ts - case sub of - Nothing -> runEventLoop kafka ct (Just $ Timeout 100) >> return (Right kafka) + rdk <- newRdKafkaT RdKafkaConsumer kc' + case rdk of + Left err -> return . Left $ KafkaError err + Right rdk' -> do + msgq <- rdKafkaQueueNew rdk' + writeIORef qref (Just msgq) + let kafka = KafkaConsumer (Kafka rdk') kc + redErr <- redirectCallbacksPoll kafka + case redErr of Just err -> closeConsumer kafka >> return (Left err) + Nothing -> do + forM_ (cpLogLevel cp) (setConsumerLogLevel kafka) + sub <- subscribe kafka ts + case sub of + Nothing -> runConsumerLoop kafka ct (Just $ Timeout 100) >> return (Right kafka) + Just err -> closeConsumer kafka >> return (Left err) --- | Polls the next message from a subscription pollMessage :: MonadIO m => KafkaConsumer -> Timeout -- ^ the timeout, in milliseconds -> m (Either KafkaError (ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString))) -- ^ Left on error or timeout, right for success -pollMessage c@(KafkaConsumer (Kafka k) _) (Timeout ms) = - liftIO $ pollEvents c Nothing >> rdKafkaConsumerPoll k (fromIntegral ms) >>= fromMessagePtr - +pollMessage c@(KafkaConsumer _ (KafkaConf _ qr _)) (Timeout ms) = liftIO $ do + pollConsumerEvents c Nothing + mbq <- readIORef qr + case mbq of + Nothing -> return . Left $ KafkaBadSpecification "Messages queue is not configured, internal error, fatal." + Just q -> rdKafkaConsumeQueue q (fromIntegral ms) >>= fromMessagePtr -- | Commit message's offset on broker for the message's partition. commitOffsetMessage :: MonadIO m @@ -111,15 +121,6 @@ commitPartitionsOffsets o k ps = liftIO $ toNativeTopicPartitionList ps >>= commitOffsets o k --- | Assigns specified partitions to a current consumer. --- Assigning an empty list means unassigning from all partitions that are currently assigned. -assign :: MonadIO m => KafkaConsumer -> [TopicPartition] -> m KafkaError -assign (KafkaConsumer (Kafka k) _) ps = - let pl = if null ps - then newForeignPtr_ nullPtr - else toNativeTopicPartitionList ps - in liftIO $ KafkaResponseError <$> (pl >>= rdKafkaAssign k) - -- | Returns current consumer's assignment assignment :: MonadIO m => KafkaConsumer -> m (Either KafkaError (M.Map TopicName [PartitionId])) assignment (KafkaConsumer (Kafka k) _) = liftIO $ do @@ -192,11 +193,33 @@ RdKafkaRespErrNoError -> Right <$> fromNativeTopicPartitionList'' ntps err -> return $ Left (KafkaResponseError err) +-- | Polls the provided kafka consumer for events. +-- +-- Events will cause application provided callbacks to be called. +-- +-- The \p timeout_ms argument specifies the maximum amount of time +-- (in milliseconds) that the call will block waiting for events. +-- +-- This function is called on each 'pollMessage' and, if runtime allows +-- multi threading, it is called periodically in a separate thread +-- to ensure the callbacks are handled ASAP. +-- +-- There is no particular need to call this function manually +-- unless some special cases in a single-threaded environment +-- when polling for events on each 'pollMessage' is not +-- frequent enough. +pollConsumerEvents :: KafkaConsumer -> Maybe Timeout -> IO () +pollConsumerEvents k timeout = + let (Timeout tm) = fromMaybe (Timeout 0) timeout + in void $ rdKafkaConsumerPoll (getRdKafka k) tm -- | Closes the consumer. closeConsumer :: MonadIO m => KafkaConsumer -> m (Maybe KafkaError) -closeConsumer (KafkaConsumer (Kafka k) (KafkaConf _ ct)) = - liftIO $ CToken.cancel ct >> (kafkaErrorToMaybe . KafkaResponseError) <$> rdKafkaConsumerClose k +closeConsumer (KafkaConsumer (Kafka k) (KafkaConf _ qr ct)) = liftIO $ do + CToken.cancel ct + mbq <- readIORef qr + void $ traverse rdKafkaQueueDestroy mbq + (kafkaErrorToMaybe . KafkaResponseError) <$> rdKafkaConsumerClose k ----------------------------------------------------------------------------- newConsumerConf :: ConsumerProperties -> IO KafkaConf @@ -219,7 +242,7 @@ return $ kafkaErrorToMaybe res setDefaultTopicConf :: KafkaConf -> TopicConf -> IO () -setDefaultTopicConf (KafkaConf kc _) (TopicConf tc) = +setDefaultTopicConf (KafkaConf kc _ _) (TopicConf tc) = rdKafkaTopicConfDup tc >>= rdKafkaConfSetDefaultTopicConf kc commitOffsets :: OffsetCommit -> KafkaConsumer -> RdKafkaTopicPartitionListTPtr -> IO (Maybe KafkaError) @@ -229,3 +252,18 @@ setConsumerLogLevel :: KafkaConsumer -> KafkaLogLevel -> IO () setConsumerLogLevel (KafkaConsumer (Kafka k) _) level = liftIO $ rdKafkaSetLogLevel k (fromEnum level) + +redirectCallbacksPoll :: KafkaConsumer -> IO (Maybe KafkaError) +redirectCallbacksPoll (KafkaConsumer (Kafka k) _) = + (kafkaErrorToMaybe . KafkaResponseError) <$> rdKafkaPollSetConsumer k + +runConsumerLoop :: KafkaConsumer -> CancellationToken -> Maybe Timeout -> IO () +runConsumerLoop k ct timeout = + when rtsSupportsBoundThreads $ void $ forkIO go + where + go = do + token <- CToken.status ct + case token of + Running -> pollConsumerEvents k timeout >> go + Cancelled -> return () + diff --git a/src/Kafka/Consumer/Callbacks.hs b/src/Kafka/Consumer/Callbacks.hs --- a/src/Kafka/Consumer/Callbacks.hs +++ b/src/Kafka/Consumer/Callbacks.hs @@ -5,12 +5,16 @@ ) where -import Foreign +import Control.Arrow ((&&&)) +import Control.Monad (forM_, void) +import Data.Monoid ((<>)) +import Foreign hiding (void) import Kafka.Callbacks as X import Kafka.Consumer.Convert import Kafka.Consumer.Types import Kafka.Internal.RdKafka import Kafka.Internal.Setup +import Kafka.Internal.Shared import Kafka.Types -- | Sets a callback that is called when rebalance is needed. @@ -25,13 +29,14 @@ -- It is OK to alter partitions offsets before calling 'assign'. -- -- * When 'RdKafkaRespErrRevokePartitions' happens 'assign' should be called with an empty list of partitions. -rebalanceCallback :: (KafkaConsumer -> KafkaError -> [TopicPartition] -> IO ()) -> KafkaConf -> IO () -rebalanceCallback callback (KafkaConf conf ct) = rdKafkaConfSetRebalanceCb conf realCb +-- rebalanceCallback :: (KafkaConsumer -> KafkaError -> [TopicPartition] -> IO ()) -> KafkaConf -> IO () +rebalanceCallback :: (KafkaConsumer -> RebalanceEvent -> IO ()) -> KafkaConf -> IO () +rebalanceCallback callback kc@(KafkaConf conf _ _) = rdKafkaConfSetRebalanceCb conf realCb where realCb k err pl = do k' <- newForeignPtr_ k pls <- fromNativeTopicPartitionList' pl - callback (KafkaConsumer (Kafka k') (KafkaConf conf ct)) (KafkaResponseError err) pls + setRebalanceCallback callback (KafkaConsumer (Kafka k') kc) (KafkaResponseError err) pls -- | Sets a callback that is called when rebalance is needed. -- @@ -44,10 +49,45 @@ -- with `KafkaError` == `KafkaResponseError` `RdKafkaRespErrNoOffset` which is not to be considered -- an error. offsetCommitCallback :: (KafkaConsumer -> KafkaError -> [TopicPartition] -> IO ()) -> KafkaConf -> IO () -offsetCommitCallback callback (KafkaConf conf ct) = rdKafkaConfSetOffsetCommitCb conf realCb +offsetCommitCallback callback kc@(KafkaConf conf _ _) = rdKafkaConfSetOffsetCommitCb conf realCb where realCb k err pl = do k' <- newForeignPtr_ k pls <- fromNativeTopicPartitionList' pl - callback (KafkaConsumer (Kafka k') (KafkaConf conf ct)) (KafkaResponseError err) pls + callback (KafkaConsumer (Kafka k') kc) (KafkaResponseError err) pls +------------------------------------------------------------------------------- +redirectPartitionQueue :: Kafka -> TopicName -> PartitionId -> RdKafkaQueueTPtr -> IO () +redirectPartitionQueue (Kafka k) (TopicName t) (PartitionId p) q = do + mpq <- rdKafkaQueueGetPartition k t p + case mpq of + Nothing -> return () + Just pq -> rdKafkaQueueForward pq q + +setRebalanceCallback :: (KafkaConsumer -> RebalanceEvent -> IO ()) + -> KafkaConsumer + -> KafkaError + -> [TopicPartition] -> IO () +setRebalanceCallback f k e ps = + case e of + KafkaResponseError RdKafkaRespErrAssignPartitions -> do + mbq <- getRdMsgQueue $ getKafkaConf k + case mbq of + Nothing -> pure () + Just mq -> forM_ ps (\tp -> redirectPartitionQueue (getKafka k) (tpTopicName tp) (tpPartition tp) mq) + void $ assign k ps + f k (RebalanceAssign ((tpTopicName &&& tpPartition) <$> ps)) + KafkaResponseError RdKafkaRespErrRevokePartitions -> do + void $ assign k [] + f k (RebalanceRevoke ((tpTopicName &&& tpPartition) <$> ps)) + x -> error $ "Rebalance: UNKNOWN response: " <> show x + +-- | Assigns specified partitions to a current consumer. +-- Assigning an empty list means unassigning from all partitions that are currently assigned. +assign :: KafkaConsumer -> [TopicPartition] -> IO (Maybe KafkaError) +assign (KafkaConsumer (Kafka k) _) ps = + let pl = if null ps + then newForeignPtr_ nullPtr + else toNativeTopicPartitionList ps + er = KafkaResponseError <$> (pl >>= rdKafkaAssign k) + in kafkaErrorToMaybe <$> er 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 @@ -1,19 +1,20 @@ module Kafka.Consumer.ConsumerProperties ( module Kafka.Consumer.ConsumerProperties -, module Kafka.Consumer.Callbacks +, module X ) where -- import Control.Monad -import qualified Data.List as L -import Data.Map (Map) -import qualified Data.Map as M -import Kafka.Consumer.Callbacks +import qualified Data.List as L +import Data.Map (Map) +import qualified Data.Map as M import Kafka.Consumer.Types import Kafka.Internal.Setup import Kafka.Types +import Kafka.Consumer.Callbacks as X + -- | Properties to create 'KafkaConsumer'. data ConsumerProperties = ConsumerProperties { cpProps :: Map String String @@ -30,7 +31,7 @@ } {-# INLINE mempty #-} mappend (ConsumerProperties m1 ll1 cb1) (ConsumerProperties m2 ll2 cb2) = - ConsumerProperties (M.union m2 m1) (ll2 `mplus` ll1) (cb2 `mplus` cb1) + ConsumerProperties (M.union m2 m1) (ll2 `mplus` ll1) (cb1 `mplus` cb2) {-# INLINE mappend #-} brokersList :: [BrokerAddress] -> ConsumerProperties 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 @@ -28,6 +28,11 @@ newtype Offset = Offset Int64 deriving (Show, Eq, Read) data OffsetReset = Earliest | Latest deriving (Show, Eq) +data RebalanceEvent = + RebalanceAssign [(TopicName, PartitionId)] + | RebalanceRevoke [(TopicName, PartitionId)] + deriving (Eq, Show) + data PartitionOffset = PartitionOffsetBeginning | PartitionOffsetEnd diff --git a/src/Kafka/Internal/RdKafka.chs b/src/Kafka/Internal/RdKafka.chs --- a/src/Kafka/Internal/RdKafka.chs +++ b/src/Kafka/Internal/RdKafka.chs @@ -505,13 +505,36 @@ {`RdKafkaTPtr'} -> `RdKafkaQueueTPtr' #} foreign import ccall unsafe "rdkafka.h &rd_kafka_queue_destroy" - rdKafkaQueueDestroy :: FinalizerPtr RdKafkaQueueT + rdKafkaQueueDestroyF :: FinalizerPtr RdKafkaQueueT +{#fun rd_kafka_queue_destroy as ^ + {`RdKafkaQueueTPtr'} -> `()'#} + newRdKafkaQueue :: RdKafkaTPtr -> IO RdKafkaQueueTPtr newRdKafkaQueue k = do q <- rdKafkaQueueNew k - addForeignPtrFinalizer rdKafkaQueueDestroy q + addForeignPtrFinalizer rdKafkaQueueDestroyF q return q + +{#fun rd_kafka_consume_queue as ^ + {`RdKafkaQueueTPtr', `Int'} -> `RdKafkaMessageTPtr' #} + +{#fun rd_kafka_queue_forward as ^ + {`RdKafkaQueueTPtr', `RdKafkaQueueTPtr'} -> `()' #} + +{#fun rd_kafka_queue_get_partition as rdKafkaQueueGetPartition' + {`RdKafkaTPtr', `String', `Int'} -> `RdKafkaQueueTPtr' #} + +rdKafkaQueueGetPartition :: RdKafkaTPtr -> String -> Int -> IO (Maybe RdKafkaQueueTPtr) +rdKafkaQueueGetPartition k t p = do + ret <- rdKafkaQueueGetPartition' k t p + withForeignPtr ret $ \realPtr -> + if realPtr == nullPtr then return Nothing + else do + addForeignPtrFinalizer rdKafkaQueueDestroyF ret + return $ Just ret + + ------------------------------------------------------------------------------------------------- ---- High-level KafkaConsumer diff --git a/src/Kafka/Internal/Setup.hs b/src/Kafka/Internal/Setup.hs --- a/src/Kafka/Internal/Setup.hs +++ b/src/Kafka/Internal/Setup.hs @@ -5,6 +5,7 @@ import Control.Exception import Control.Monad +import Data.IORef import Foreign import Foreign.C.String import Kafka.Internal.CancellationToken @@ -15,7 +16,7 @@ newtype KafkaProps = KafkaProps [(String, String)] deriving (Show, Eq) newtype TopicProps = TopicProps [(String, String)] deriving (Show, Eq) newtype Kafka = Kafka RdKafkaTPtr deriving Show -data KafkaConf = KafkaConf RdKafkaConfTPtr CancellationToken +data KafkaConf = KafkaConf RdKafkaConfTPtr (IORef (Maybe RdKafkaQueueTPtr)) CancellationToken newtype TopicConf = TopicConf RdKafkaTopicConfTPtr deriving Show class HasKafka a where @@ -44,9 +45,14 @@ {-# INLINE getRdKafka #-} getRdKafkaConf :: HasKafkaConf k => k -> RdKafkaConfTPtr -getRdKafkaConf k = let (KafkaConf k' _) = getKafkaConf k in k' +getRdKafkaConf k = let (KafkaConf k' _ _) = getKafkaConf k in k' {-# INLINE getRdKafkaConf #-} +getRdMsgQueue :: HasKafkaConf k => k -> IO (Maybe RdKafkaQueueTPtr) +getRdMsgQueue k = + let (KafkaConf _ rq _) = getKafkaConf k + in readIORef rq + getRdTopicConf :: HasTopicConf t => t -> RdKafkaTopicConfTPtr getRdTopicConf t = let (TopicConf t') = getTopicConf t in t' {-# INLINE getRdTopicConf #-} @@ -55,7 +61,7 @@ newTopicConf = TopicConf <$> newRdKafkaTopicConfT newKafkaConf :: IO KafkaConf -newKafkaConf = KafkaConf <$> newRdKafkaConfT <*> newCancellationToken +newKafkaConf = KafkaConf <$> newRdKafkaConfT <*> newIORef Nothing <*> newCancellationToken kafkaConf :: KafkaProps -> IO KafkaConf kafkaConf overrides = do @@ -80,7 +86,7 @@ throw $ KafkaUnknownConfigurationKey str setKafkaConfValue :: KafkaConf -> String -> String -> IO () -setKafkaConfValue (KafkaConf confPtr _) key value = +setKafkaConfValue (KafkaConf confPtr _ _) key value = allocaBytes nErrorBytes $ \charPtr -> do err <- rdKafkaConfSet confPtr key value charPtr (fromIntegral nErrorBytes) checkConfSetValue err charPtr diff --git a/src/Kafka/Producer.hs b/src/Kafka/Producer.hs --- a/src/Kafka/Producer.hs +++ b/src/Kafka/Producer.hs @@ -56,7 +56,7 @@ -- A newly created producer must be closed with 'closeProducer' function. newProducer :: MonadIO m => ProducerProperties -> m (Either KafkaError KafkaProducer) newProducer pps = liftIO $ do - kc@(KafkaConf kc' _) <- kafkaConf (KafkaProps $ M.toList (ppKafkaProps pps)) + kc@(KafkaConf kc' _ _) <- kafkaConf (KafkaProps $ M.toList (ppKafkaProps pps)) tc <- topicConf (TopicProps $ M.toList (ppTopicProps pps)) -- set callbacks @@ -151,7 +151,7 @@ -- Will wait until the outbound queue is drained before returning the control. closeProducer :: MonadIO m => KafkaProducer -> m () closeProducer p = - let (KafkaConf _ ct) = kpKafkaConf p + let (KafkaConf _ _ ct) = kpKafkaConf p in liftIO (CToken.cancel ct) >> flushProducer p -- | Drains the outbound queue for a producer. diff --git a/src/Kafka/Producer/Callbacks.hs b/src/Kafka/Producer/Callbacks.hs --- a/src/Kafka/Producer/Callbacks.hs +++ b/src/Kafka/Producer/Callbacks.hs @@ -15,7 +15,7 @@ -- | Sets the callback for delivery errors. -- The callback is only called in case of errors. deliveryErrorsCallback :: (KafkaError -> IO ()) -> KafkaConf -> IO () -deliveryErrorsCallback callback (KafkaConf conf _) = rdKafkaConfSetDrMsgCb conf realCb +deliveryErrorsCallback callback kc = rdKafkaConfSetDrMsgCb (getRdKafkaConf kc) realCb where realCb :: t -> Ptr RdKafkaMessageT -> IO () realCb _ mptr = 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 @@ -30,7 +30,7 @@ } {-# INLINE mempty #-} mappend (ProducerProperties k1 t1 ll1 cb1) (ProducerProperties k2 t2 ll2 cb2) = - ProducerProperties (M.union k2 k1) (M.union t2 t1) (ll2 `mplus` ll1) (cb2 `mplus` cb1) + ProducerProperties (M.union k2 k1) (M.union t2 t1) (ll2 `mplus` ll1) (cb1 `mplus` cb2) {-# INLINE mappend #-} brokersList :: [BrokerAddress] -> ProducerProperties diff --git a/tests-it/Kafka/IntegrationSpec.hs b/tests-it/Kafka/IntegrationSpec.hs --- a/tests-it/Kafka/IntegrationSpec.hs +++ b/tests-it/Kafka/IntegrationSpec.hs @@ -107,7 +107,9 @@ res <- seek k (Timeout 1000) [TopicPartition testTopic (PartitionId 0) PartitionOffsetEnd] res `shouldBe` Nothing msg <- pollMessage k (Timeout 1000) - crOffset <$> msg `shouldBe` Left (KafkaResponseError RdKafkaRespErrPartitionEof) + crOffset <$> msg `shouldSatisfy` (\x -> + x == Left (KafkaResponseError RdKafkaRespErrPartitionEof) + || x == Left (KafkaResponseError RdKafkaRespErrTimedOut)) it "should respect out-of-bound offsets (invalid offset)" $ \k -> do res <- seek k (Timeout 1000) [TopicPartition testTopic (PartitionId 0) PartitionOffsetInvalid]