packages feed

hw-kafka-client 5.0.0 → 5.3.0

raw patch · 12 files changed

+285/−26 lines, 12 files

Files

example/ProducerExample.hs view
@@ -50,7 +50,7 @@   putStrLn "Producer is ready, send your messages!"   msg1 <- getLine -  err1 <- produceMessage prod (mkMessage Nothing (Just $ pack msg1))+  err1 <- produceMessage prod (mkMessage (Just "zero") (Just $ pack msg1))   forM_ err1 print    putStrLn "One more time!"@@ -62,7 +62,7 @@   putStrLn "And the last one..."   msg3 <- getLine   err3 <- produceMessage prod (mkMessage (Just "key3") (Just $ pack msg3))-  +   err4 <- produceMessage prod ((mkMessage (Just "key4") (Just $ pack msg3)) { prHeaders = headersFromList [("fancy", "header")]})    -- forM_ errs (print . snd)
hw-kafka-client.cabal view
@@ -1,7 +1,7 @@ cabal-version: 2.2  name:                   hw-kafka-client-version:                5.0.0+version:                5.3.0 synopsis:               Kafka bindings for Haskell description:            Apache Kafka bindings backed by the librdkafka C library.                         .@@ -64,6 +64,7 @@                         Kafka.Producer                         Kafka.Producer.ProducerProperties                         Kafka.Producer.Types+                        Kafka.Transaction                         Kafka.Types   other-modules:        Kafka.Callbacks                         Kafka.Consumer.Callbacks
src/Kafka/Consumer.hs view
@@ -55,11 +55,12 @@ , newConsumer , assign, assignment, subscription , pausePartitions, resumePartitions-, committed, position, seek+, committed, position, seek, seekPartitions , pollMessage, pollConsumerEvents , pollMessageBatch , commitOffsetMessage, commitAllOffsets, commitPartitionsOffsets , storeOffsets, storeOffsetMessage+, rewindConsumer , closeConsumer -- ReExport Types , RdKafkaRespErrT (..)@@ -75,16 +76,15 @@ import           Data.Bifunctor             (bimap, first) import qualified Data.ByteString            as BS import           Data.IORef                 (readIORef, writeIORef)-import qualified Data.Map                   as M+import qualified Data.Map                   as M hiding (map, foldr) import           Data.Maybe                 (fromMaybe)-import           Data.Monoid                ((<>)) import           Data.Set                   (Set) import qualified Data.Set                   as Set import qualified Data.Text                  as Text import           Foreign                    hiding (void) import           Kafka.Consumer.Convert     (fromMessagePtr, fromNativeTopicPartitionList'', offsetCommitToBool, offsetToInt64, toMap, toNativeTopicPartitionList, toNativeTopicPartitionList', toNativeTopicPartitionListNoDispose, topicPartitionFromMessageForCommit) import           Kafka.Consumer.Types       (KafkaConsumer (..))-import           Kafka.Internal.RdKafka     (RdKafkaRespErrT (..), RdKafkaTopicPartitionListTPtr, RdKafkaTypeT (..), newRdKafkaT, newRdKafkaTopicPartitionListT, newRdKafkaTopicT, rdKafkaAssign, rdKafkaAssignment, rdKafkaCommit, rdKafkaCommitted, rdKafkaConfSetDefaultTopicConf, rdKafkaConsumeBatchQueue, rdKafkaConsumeQueue, rdKafkaConsumerClose, rdKafkaConsumerPoll, rdKafkaOffsetsStore, rdKafkaPausePartitions, rdKafkaPollSetConsumer, rdKafkaPosition, rdKafkaQueueDestroy, rdKafkaQueueNew, rdKafkaResumePartitions, rdKafkaSeek, rdKafkaSetLogLevel, rdKafkaSubscribe, rdKafkaSubscription, rdKafkaTopicConfDup, rdKafkaTopicPartitionListAdd)+import           Kafka.Internal.RdKafka     (RdKafkaRespErrT (..), RdKafkaTopicPartitionListTPtr, RdKafkaTypeT (..), rdKafkaSeekPartitions, rdKafkaErrorDestroy, rdKafkaErrorCode, newRdKafkaT, newRdKafkaTopicPartitionListT, newRdKafkaTopicT, rdKafkaAssign, rdKafkaAssignment, rdKafkaCommit, rdKafkaCommitted, rdKafkaConfSetDefaultTopicConf, rdKafkaConsumeBatchQueue, rdKafkaConsumeQueue, rdKafkaConsumerClose, rdKafkaConsumerPoll, rdKafkaOffsetsStore, rdKafkaPausePartitions, rdKafkaPollSetConsumer, rdKafkaPosition, rdKafkaQueueDestroy, rdKafkaQueueNew, rdKafkaResumePartitions, rdKafkaSeek, rdKafkaSetLogLevel, rdKafkaSubscribe, rdKafkaSubscription, rdKafkaTopicConfDup, rdKafkaTopicPartitionListAdd) import           Kafka.Internal.Setup       (CallbackPollStatus (..), Kafka (..), KafkaConf (..), KafkaProps (..), TopicConf (..), TopicProps (..), getKafkaConf, getRdKafka, kafkaConf, topicConf, Callback(..)) import           Kafka.Internal.Shared      (kafkaErrorToMaybe, maybeToLeft, rdKafkaErrorToEither) @@ -255,6 +255,7 @@   KafkaResponseError <$> rdKafkaResumePartitions k pl  -- | Seek a particular offset for each provided 'TopicPartition'+{-# DEPRECATED seek "Use seekPartitions instead" #-} seek :: MonadIO m => KafkaConsumer -> Timeout -> [TopicPartition] -> m (Maybe KafkaError) seek (KafkaConsumer (Kafka k) _) (Timeout timeout) tps = liftIO $   either Just (const Nothing) <$> seekAll@@ -271,6 +272,14 @@       nt <- newRdKafkaTopicT k (Text.unpack tn) Nothing       return $ bimap KafkaError (,tpPartition tp, tpOffset tp) (first Text.pack nt) +-- | Seek consumer for partitions in partitions to the per-partition+--    offset in the offset field of partitions.+seekPartitions :: MonadIO m => KafkaConsumer -> [TopicPartition] -> Timeout -> m (Maybe KafkaError)+seekPartitions (KafkaConsumer (Kafka k) _) ps (Timeout timeout) = liftIO $ do+  tps <- toNativeTopicPartitionList ps+  err <- bracket (rdKafkaSeekPartitions k tps timeout) rdKafkaErrorDestroy rdKafkaErrorCode+  pure $ either Just (const Nothing) $ rdKafkaErrorToEither err+ -- | Retrieve committed offsets for topics+partitions. committed :: MonadIO m => KafkaConsumer -> Timeout -> [(TopicName, PartitionId)] -> m (Either KafkaError [TopicPartition]) committed (KafkaConsumer (Kafka k) _) (Timeout timeout) tps = liftIO $ do@@ -323,6 +332,41 @@     readIORef qr >>= mapM_ rdKafkaQueueDestroy     res <- kafkaErrorToMaybe . KafkaResponseError <$> rdKafkaConsumerClose k     pure (CallbackPollDisabled, res)++-- | Rewind consumer's consume position to the last committed offsets for the current assignment.+-- NOTE: follows https://github.com/edenhill/librdkafka/blob/master/examples/transactions.c#L166+rewindConsumer :: MonadIO m +               => KafkaConsumer +               -> Timeout+               -> m (Maybe KafkaError)+rewindConsumer c to = liftIO $ do+  ret <- assignment c+  case ret of+    Left err -> pure $ Just err+    Right os -> do+      if M.size os == 0+        -- No current assignment to rewind+        then pure Nothing+        else do+          let tps = foldr (\(t, ps) acc -> map (t,) ps ++ acc) [] $ M.toList os+          ret' <- committed c to tps+          case ret' of+            Left err -> pure $ Just err+            Right ps -> do+              -- Seek to committed offset, or start of partition if no+              -- committed offset is available.+              let ps' = map checkOffsets ps+              seekPartitions c ps' to+  where+    checkOffsets :: TopicPartition -> TopicPartition+    checkOffsets tp +      | isUncommitedOffset $ tpOffset tp +        = tp { tpOffset = PartitionOffsetBeginning }+      | otherwise = tp ++    isUncommitedOffset :: PartitionOffset -> Bool+    isUncommitedOffset (PartitionOffset _) = False+    isUncommitedOffset _ = True ----------------------------------------------------------------------------- newConsumerConf :: ConsumerProperties -> IO KafkaConf newConsumerConf ConsumerProperties {cpProps = m, cpCallbacks = cbs} = do
src/Kafka/Consumer/Callbacks.hs view
@@ -66,7 +66,10 @@   case e of     KafkaResponseError RdKafkaRespErrAssignPartitions -> do         f k (RebalanceBeforeAssign assignment)-        void $ rdKafkaAssign kptr pls+        protocol <- rdKafkaRebalanceProtocol kptr+        if protocol == "COOPERATIVE" +          then void $ rdKafkaIncrementalAssign kptr pls+          else void $ rdKafkaAssign kptr pls          mbq <- getRdMsgQueue $ getKafkaConf k         case mbq of@@ -84,6 +87,9 @@      KafkaResponseError RdKafkaRespErrRevokePartitions -> do         f k (RebalanceBeforeRevoke assignment)-        void $ newForeignPtr_ nullPtr >>= rdKafkaAssign kptr+        protocol <- rdKafkaRebalanceProtocol kptr+        if protocol == "COOPERATIVE" +          then void $ rdKafkaIncrementalUnassign kptr pls+          else void $ newForeignPtr_ nullPtr >>= rdKafkaAssign kptr         f k (RebalanceRevoke assignment)     x -> error $ "Rebalance: UNKNOWN response: " <> show x
src/Kafka/Consumer/ConsumerProperties.hs view
@@ -34,7 +34,6 @@ import           Data.Text            (Text) import qualified Data.Text            as Text import           Kafka.Consumer.Types (ConsumerGroupId (..))-import           Kafka.Internal.Setup (KafkaConf (..), Callback(..)) import           Kafka.Types          (BrokerAddress (..), ClientId (..), KafkaCompressionCodec (..), KafkaDebug (..), KafkaLogLevel (..), Millis (..), kafkaCompressionCodecToText, kafkaDebugToText)  import Kafka.Consumer.Callbacks as X
src/Kafka/Dump.hs view
@@ -1,6 +1,6 @@ ----------------------------------------------------------------------------- -- |--- Module providing various functions to dump information. These may be useful for +-- 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
src/Kafka/Internal/RdKafka.chs view
@@ -443,7 +443,7 @@ rdKafkaConfSetSocketCb :: RdKafkaConfTPtr -> SocketCallback -> IO () rdKafkaConfSetSocketCb conf cb = do     cb' <- mkSocketCallback cb-    withForeignPtr conf $ \c -> rdKafkaConfSetSocketCb' c cb' >> pure 0+    _ <- withForeignPtr conf $ \c -> rdKafkaConfSetSocketCb' c cb' >> pure (0 :: Int)     return ()  {#fun rd_kafka_conf_set_opaque as ^@@ -611,6 +611,9 @@     {`RdKafkaTPtr', `RdKafkaTopicPartitionListTPtr'}     -> `RdKafkaRespErrT' cIntToEnum #} +{#fun rd_kafka_rebalance_protocol as ^+    {`RdKafkaTPtr'} -> `String' #}+ {#fun rd_kafka_assignment as rdKafkaAssignment'     {`RdKafkaTPtr', alloca- `Ptr RdKafkaTopicPartitionListT' peekPtr* }     -> `RdKafkaRespErrT' cIntToEnum #}@@ -1084,6 +1087,66 @@     fptr <- newForeignPtr_ arrPtr     rdKafkaMessageProduceVa' kafkaPtr fptr (cIntConv i) +--- Transactional api++{#fun rd_kafka_init_transactions as rdKafkaInitTransactions+    {`RdKafkaTPtr', `Int'} -> `RdKafkaErrorTPtr' #}++{#fun rd_kafka_begin_transaction as rdKafkaBeginTransaction+    {`RdKafkaTPtr'} -> `RdKafkaErrorTPtr' #}++{#fun rd_kafka_send_offsets_to_transaction as rdKafkaSendOffsetsToTransaction'+    {`RdKafkaTPtr', `RdKafkaTopicPartitionListTPtr', `Ptr ()', `Int' } -> `RdKafkaErrorTPtr' #}++{#fun rd_kafka_commit_transaction as rdKafkaCommitTransaction+    {`RdKafkaTPtr', `Int'} -> `RdKafkaErrorTPtr' #}++{#fun rd_kafka_abort_transaction as rdKafkaAbortTransaction+    {`RdKafkaTPtr', `Int'} -> `RdKafkaErrorTPtr' #}++{#fun rd_kafka_incremental_assign as ^+    {`RdKafkaTPtr', `RdKafkaTopicPartitionListTPtr'} -> `RdKafkaErrorTPtr' #}++{#fun rd_kafka_incremental_unassign as ^+    {`RdKafkaTPtr', `RdKafkaTopicPartitionListTPtr'}+    -> `RdKafkaErrorTPtr' #}++{#fun rd_kafka_consumer_group_metadata as rdKafkaConsumerGroupMetadata+    {`RdKafkaTPtr'} -> `Ptr ()' #}++{#fun rd_kafka_consumer_group_metadata_destroy as rdKafkaConsumerGroupMetadataDestroy+    {`Ptr ()'} -> `()' #}++{#fun rd_kafka_seek_partitions as rdKafkaSeekPartitions+    {`RdKafkaTPtr', `RdKafkaTopicPartitionListTPtr', `Int'} -> `RdKafkaErrorTPtr' #}++rdKafkaSendOffsetsToTransaction :: RdKafkaTPtr -> RdKafkaTPtr -> RdKafkaTopicPartitionListTPtr -> Int -> IO RdKafkaErrorTPtr+rdKafkaSendOffsetsToTransaction p c topicPartition timeOut = do+    metaData <- rdKafkaConsumerGroupMetadata c+    errPtr   <- rdKafkaSendOffsetsToTransaction' p topicPartition metaData timeOut+    -- NOTE:need to destroy the metadata otherwise leaking memory+    rdKafkaConsumerGroupMetadataDestroy metaData+    pure errPtr++{#fun rd_kafka_error_is_fatal as rdKafkaErrorIsFatal'+    {`RdKafkaErrorTPtr'} -> `CInt' #}++{#fun rd_kafka_error_is_retriable as rdKafkaErrorIsRetriable'+    {`RdKafkaErrorTPtr'} -> `CInt' #}++{#fun rd_kafka_error_txn_requires_abort as rdKafkaErrorTxnRequiresAbort'+    {`RdKafkaErrorTPtr'} -> `CInt' #}++rdKafkaErrorIsFatal :: RdKafkaErrorTPtr -> IO Bool+rdKafkaErrorIsFatal ptr = boolFromCInt <$> rdKafkaErrorIsFatal' ptr++rdKafkaErrorIsRetriable :: RdKafkaErrorTPtr -> IO Bool+rdKafkaErrorIsRetriable ptr = boolFromCInt <$> rdKafkaErrorIsRetriable' ptr++rdKafkaErrorTxnRequiresAbort :: RdKafkaErrorTPtr -> IO Bool+rdKafkaErrorTxnRequiresAbort ptr = boolFromCInt <$> rdKafkaErrorTxnRequiresAbort' ptr++ -- Marshall / Unmarshall enumToCInt :: Enum a => a -> CInt enumToCInt = fromIntegral . fromEnum@@ -1101,6 +1164,11 @@ boolToCInt True = CInt 1 boolToCInt False = CInt 0 {-# INLINE boolToCInt #-}++boolFromCInt :: CInt -> Bool+boolFromCInt (CInt 0) = False+boolFromCInt (CInt _) = True+{-# INLINE boolFromCInt #-}  peekInt64Conv :: (Storable a, Integral a) =>  Ptr a -> IO Int64 peekInt64Conv  = liftM cIntConv . peek
src/Kafka/Producer.hs view
@@ -83,6 +83,7 @@ import           Kafka.Producer.Convert   (copyMsgFlags, handleProduceErrT, producePartitionCInt) import           Kafka.Producer.Types     (KafkaProducer (..)) + import Kafka.Producer.ProducerProperties as X import Kafka.Producer.Types              as X hiding (KafkaProducer) import Kafka.Types                       as X@@ -191,7 +192,6 @@     if (l == 0)       then pollEvents kp (Just $ Timeout 0) -- to be sure that all the delivery reports are fired       else flushProducer kp- ------------------------------------------------------------------------------------  withHeaders :: Headers -> ([RdKafkaVuT] -> IO a) -> IO a
src/Kafka/Producer/ProducerProperties.hs view
@@ -14,6 +14,7 @@ , sendTimeout , statisticsInterval , extraProps+, extraProp , suppressDisconnectLogs , extraTopicProps , debugOptions@@ -27,7 +28,6 @@ import           Data.Map                 (Map) import qualified Data.Map                 as M import           Data.Semigroup           as Sem-import           Kafka.Internal.Setup     (KafkaConf(..), Callback(..)) import           Kafka.Types              (KafkaDebug(..), Timeout(..), KafkaCompressionCodec(..), KafkaLogLevel(..), BrokerAddress(..), kafkaDebugToText, kafkaCompressionCodecToText, Millis(..))  import           Kafka.Producer.Callbacks@@ -102,6 +102,12 @@ -- The full list can be found <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here> extraProps :: Map Text Text -> ProducerProperties extraProps m = mempty { ppKafkaProps = m }++-- | Any configuration options that are supported by /librdkafka/.+-- The full list can be found <https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md here>+extraProp :: Text -> Text -> ProducerProperties+extraProp k v = mempty { ppKafkaProps = M.singleton k v }+{-# INLINE extraProp #-}  -- | Suppresses producer disconnects logs. --
+ src/Kafka/Transaction.hs view
@@ -0,0 +1,132 @@+-----------------------------------------------------------------------------+-- |+-- Module to work wih Kafkas transactional producers.+-- +-----------------------------------------------------------------------------+module Kafka.Transaction+( initTransactions+, beginTransaction+, commitTransaction+, abortTransaction++, commitOffsetMessageTransaction+-- , commitTransactionWithOffsets++, TxError+, getKafkaError+, kafkaErrorIsFatal+, kafkaErrorIsRetriable+, kafkaErrorTxnRequiresAbort+)+where++import           Control.Monad.IO.Class   (MonadIO (liftIO))+import           Kafka.Internal.RdKafka   (RdKafkaErrorTPtr, rdKafkaErrorDestroy, rdKafkaErrorIsFatal, rdKafkaErrorIsRetriable, rdKafkaErrorTxnRequiresAbort, rdKafkaErrorCode, rdKafkaInitTransactions, rdKafkaBeginTransaction, rdKafkaCommitTransaction, rdKafkaAbortTransaction, rdKafkaSendOffsetsToTransaction)+import           Kafka.Internal.Setup     (getRdKafka)+import           Kafka.Producer.Convert   (handleProduceErrT)+import           Kafka.Producer+import           Kafka.Consumer.Convert   (toNativeTopicPartitionList, topicPartitionFromMessageForCommit)+import           Kafka.Consumer++-------------------------------------------------------------------------------------+-- Tx API++data TxError = TxError +  { txErrorKafka       :: !KafkaError +  , txErrorFatal       :: !Bool+  , txErrorRetriable   :: !Bool+  , txErrorTxnReqAbort :: !Bool+  } ++-- | Initialises Kafka for transactions +initTransactions :: MonadIO m +                 => KafkaProducer +                 -> Timeout +                 -> m (Maybe KafkaError)+initTransactions p (Timeout to) +  = liftIO $ rdKafkaInitTransactions (getRdKafka p) to >>= rdKafkaErrorCode >>= handleProduceErrT++-- | Begins a new transaction+beginTransaction :: MonadIO m +                 => KafkaProducer +                 -> m (Maybe KafkaError)+beginTransaction p +  = liftIO $ rdKafkaBeginTransaction (getRdKafka p) >>= rdKafkaErrorCode >>= handleProduceErrT++-- | Commits an existing transaction+-- Pre-condition: there exists an open transaction, created with beginTransaction+commitTransaction :: MonadIO m +                  => KafkaProducer +                  -> Timeout +                  -> m (Maybe TxError)+commitTransaction p (Timeout to) = liftIO $ rdKafkaCommitTransaction (getRdKafka p) to >>= toTxError++-- | Aborts an existing transaction+-- Pre-condition: there exists an open transaction, created with beginTransaction+abortTransaction :: MonadIO m +                 => KafkaProducer +                 -> Timeout +                 -> m (Maybe KafkaError)+abortTransaction p (Timeout to) +  = liftIO $ do rdKafkaAbortTransaction (getRdKafka p) to >>= rdKafkaErrorCode >>= handleProduceErrT++-- | Commits the message's offset in the current transaction+--    Similar to Kafka.Consumer.commitOffsetMessage but within a transactional context+-- Pre-condition: there exists an open transaction, created with beginTransaction+commitOffsetMessageTransaction :: MonadIO m +                               => KafkaProducer +                               -> KafkaConsumer +                               -> ConsumerRecord k v+                               -> Timeout +                               -> m (Maybe TxError)+commitOffsetMessageTransaction p c m (Timeout to) = liftIO $ do+  tps <- toNativeTopicPartitionList [topicPartitionFromMessageForCommit m]+  rdKafkaSendOffsetsToTransaction (getRdKafka p) (getRdKafka c) tps to >>= toTxError++-- -- | Commit offsets for all currently assigned partitions in the current transaction+-- --    Similar to Kafka.Consumer.commitAllOffsets but within a transactional context+-- -- Pre-condition: there exists an open transaction, created with beginTransaction+-- commitAllOffsetsTransaction :: MonadIO m +--                             => KafkaProducer +--                             -> KafkaConsumer +--                             -> Timeout +--                             -> m (Maybe TxError)+-- commitAllOffsetsTransaction p c (Timeout to) = liftIO $ do+--   -- TODO: this can't be right...+--   tps <- newForeignPtr_ nullPtr+--   rdKafkaSendOffsetsToTransaction (getRdKafka p) (getRdKafka c) tps to >>= toTxError+    +getKafkaError :: TxError -> KafkaError+getKafkaError = txErrorKafka++kafkaErrorIsFatal :: TxError -> Bool+kafkaErrorIsFatal = txErrorFatal++kafkaErrorIsRetriable :: TxError -> Bool+kafkaErrorIsRetriable = txErrorRetriable++kafkaErrorTxnRequiresAbort :: TxError -> Bool+kafkaErrorTxnRequiresAbort = txErrorTxnReqAbort++----------------------------------------------------------------------------------------------------+-- Implementation detail, used internally+toTxError :: RdKafkaErrorTPtr -> IO (Maybe TxError)+toTxError errPtr = do+  ret <- rdKafkaErrorCode errPtr >>= handleProduceErrT+  case ret of+    Nothing -> do+      -- NOTE: don't forget to free error structure, otherwise we are leaking memory!+      rdKafkaErrorDestroy errPtr+      pure Nothing+    Just ke -> do+      fatal     <- rdKafkaErrorIsFatal errPtr+      retriable <- rdKafkaErrorIsRetriable errPtr+      reqAbort  <- rdKafkaErrorTxnRequiresAbort errPtr+      -- NOTE: don't forget to free error structure, otherwise we are leaking memory!+      rdKafkaErrorDestroy errPtr+      pure $ Just $ TxError +        { txErrorKafka       = ke+        , txErrorFatal       = fatal+        , txErrorRetriable   = retriable+        , txErrorTxnReqAbort = reqAbort+        }
src/Kafka/Types.hs view
@@ -150,6 +150,7 @@   | Gzip   | Snappy   | Lz4+  | Zstd   deriving (Eq, Show, Typeable, Generic)  -- | Convert a 'KafkaCompressionCodec' into its /librdkafka/ string equivalent.@@ -161,6 +162,7 @@   Gzip          -> "gzip"   Snappy        -> "snappy"   Lz4           -> "lz4" +  Zstd          -> "zstd"  -- | Headers that might be passed along with a record newtype Headers = Headers { unHeaders :: [(BS.ByteString, BS.ByteString)] }
tests-it/Kafka/IntegrationSpec.hs view
@@ -153,21 +153,21 @@             it "should consume empty batch when there are no messages" $ \k -> do                 res <- pollMessageBatch k (Timeout 1000) (BatchSize 50)                 length res `shouldBe` 0-    +     describe "Kafka.Headers.Spec" $ do         let testHeaders = headersFromList [("a-header-name", "a-header-value"), ("b-header-name", "b-header-value")]          specWithKafka "Headers consumer/producer" consumerProps $ do               it "1. sends 2 messages to test topic enriched with headers" $ \(k, prod) -> do                   void $ receiveMessages k-                  +                   res  <- sendMessagesWithHeaders (testMessages testTopic) testHeaders prod                   res `shouldBe` Right ()               it "2. should receive 2 messages enriched with headers" $ \(k, _) -> do                   res <- receiveMessages k                   (length <$> res) `shouldBe` Right 2-                  -                  forM_ res $ \rcs -> ++                  forM_ res $ \rcs ->                     forM_ rcs ((`shouldBe` Set.fromList (headersToList testHeaders)) . Set.fromList . headersToList . crHeaders)  ----------------------------------------------------------------------------------------------------------------@@ -253,13 +253,14 @@     hasTopic `shouldBe` True    it "should return topic metadata" $ \k -> do-    res <- topicMetadata k (Timeout 1000) testTopic+    res <- topicMetadata k (Timeout 2000) testTopic     res `shouldSatisfy` isRight     length . kmBrokers <$> res `shouldBe` Right 1     length . kmTopics <$> res `shouldBe` Right 1    it "should describe all consumer groups" $ \k -> do-    res <- allConsumerGroupsInfo k (Timeout 1000)+    res <- allConsumerGroupsInfo k (Timeout 2000)+    res `shouldSatisfy` isRight     let groups = either (const []) (fmap giGroup) res     let prefixedGroups = filter isTestGroupId groups     let resLen = length prefixedGroups@@ -267,15 +268,15 @@     -- fmap giGroup <$> res `shouldBe` Right [testGroupId]    it "should describe a given consumer group" $ \k -> do-    res <- consumerGroupInfo k (Timeout 1000) testGroupId+    res <- consumerGroupInfo k (Timeout 2000) testGroupId     fmap giGroup <$> res `shouldBe` Right [testGroupId]    it "should describe non-existent consumer group" $ \k -> do-    res <- consumerGroupInfo k (Timeout 1000) "does-not-exist"+    res <- consumerGroupInfo k (Timeout 2000) "does-not-exist"     res `shouldBe` Right []    it "should read topic offsets for time" $ \k -> do-    res <- topicOffsetsForTime k (Timeout 1000) (Millis 1904057189508) testTopic+    res <- topicOffsetsForTime k (Timeout 2000) (Millis 1904057189508) testTopic     res `shouldSatisfy` isRight     fmap tpOffset <$> res `shouldBe` Right [PartitionOffsetEnd] @@ -300,13 +301,13 @@         ||  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]+    res <- seek k (Timeout 2000) [TopicPartition testTopic (PartitionId 0) PartitionOffsetInvalid]     res `shouldBe` Nothing     msg <- pollMessage k (Timeout 1000)     crOffset <$> msg `shouldBe` Right (Offset 0)    it "should respect out-of-bound offsets (huge offset)" $ \k -> do-    res <- seek k (Timeout 1000) [TopicPartition testTopic (PartitionId 0) (PartitionOffset 123456)]+    res <- seek k (Timeout 3000) [TopicPartition testTopic (PartitionId 0) (PartitionOffset 123456)]     res `shouldBe` Nothing-    msg <- pollMessage k (Timeout 1000)+    msg <- pollMessage k (Timeout 2000)     crOffset <$> msg `shouldBe` Right (Offset 0)