packages feed

hw-kafka-client 2.2.0 → 2.3.0

raw patch · 14 files changed

+217/−76 lines, 14 files

Files

README.md view
@@ -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
example/ConsumerExample.hs view
@@ -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
hw-kafka-client.cabal view
@@ -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
src/Kafka/Callbacks.hs view
@@ -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
src/Kafka/Consumer.hs view
@@ -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 ()+
src/Kafka/Consumer/Callbacks.hs view
@@ -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
src/Kafka/Consumer/ConsumerProperties.hs view
@@ -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
src/Kafka/Consumer/Types.hs view
@@ -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
src/Kafka/Internal/RdKafka.chs view
@@ -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
src/Kafka/Internal/Setup.hs view
@@ -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
src/Kafka/Producer.hs view
@@ -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.
src/Kafka/Producer/Callbacks.hs view
@@ -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 =
src/Kafka/Producer/ProducerProperties.hs view
@@ -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
tests-it/Kafka/IntegrationSpec.hs view
@@ -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]