packages feed

hw-kafka-client 2.6.1 → 3.0.0

raw patch · 12 files changed

+349/−388 lines, 12 filesdep +random

Dependencies added: random

Files

hw-kafka-client.cabal view
@@ -1,5 +1,5 @@ name:           hw-kafka-client-version:        2.6.1+version:        3.0.0 synopsis:       Kafka bindings for Haskell description:    Apache Kafka bindings backed by the librdkafka C library.                 .@@ -52,8 +52,6 @@   if impl(ghc <8.0)     build-depends:         semigroups-  if os(osx)-    cpp-options: -D__attribute__(A)= -D_Nullable= -D_Nonnull=   exposed-modules:       Kafka.Consumer       Kafka.Consumer.ConsumerProperties@@ -69,7 +67,6 @@       Kafka.Callbacks       Kafka.Consumer.Callbacks       Kafka.Consumer.Convert-      Kafka.Internal.CancellationToken       Kafka.Internal.RdKafka       Kafka.Internal.Setup       Kafka.Internal.Shared@@ -107,6 +104,7 @@     , hspec     , hw-kafka-client     , monad-loops+    , random     , text     , transformers   other-modules:
src/Kafka/Consumer.hs view
@@ -1,5 +1,6 @@+{-# LANGUAGE LambdaCase        #-} {-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE TupleSections #-}+{-# LANGUAGE TupleSections     #-} module Kafka.Consumer ( module X , runConsumer@@ -18,73 +19,27 @@ ) where -import           Data.Set                         (Set)-import qualified Data.Set                         as Set-import qualified Data.Text                        as Text-import           Control.Arrow                    ((&&&), left)-import           Control.Concurrent               (forkIO, rtsSupportsBoundThreads)-import           Control.Exception                (bracket)-import           Control.Monad                    (forM_, void, when)-import           Control.Monad.IO.Class           (MonadIO(liftIO))-import           Control.Monad.Trans.Except       (ExceptT(ExceptT), runExceptT)-import           Data.Bifunctor                   (first, bimap)-import qualified Data.ByteString                  as BS-import           Data.IORef                       (writeIORef, readIORef)-import qualified Data.Map                         as M-import           Data.Maybe                       (fromMaybe)-import           Data.Monoid                      ((<>))-import           Foreign                          hiding (void)-import           Kafka.Consumer.Convert-  (fromMessagePtr, toNativeTopicPartitionList, topicPartitionFromMessageForCommit-  , toNativeTopicPartitionListNoDispose, toNativeTopicPartitionList, fromNativeTopicPartitionList''-  , toMap, offsetToInt64, toNativeTopicPartitionList', offsetCommitToBool-  )-import           Kafka.Consumer.Types (KafkaConsumer(..))-import           Kafka.Internal.CancellationToken as CToken-import           Kafka.Internal.RdKafka-  ( RdKafkaRespErrT(..)-  , RdKafkaTopicPartitionListTPtr-  , RdKafkaTypeT(..)-  , newRdKafkaT-  , rdKafkaQueueNew-  , rdKafkaConsumeQueue-  , rdKafkaPollSetConsumer-  , rdKafkaSetLogLevel-  , rdKafkaOffsetsStore-  , rdKafkaCommit-  , rdKafkaConfSetDefaultTopicConf-  , rdKafkaTopicConfDup-  , rdKafkaSubscribe-  , rdKafkaTopicPartitionListAdd-  , newRdKafkaTopicPartitionListT-  , rdKafkaConsumerClose-  , rdKafkaQueueDestroy-  , rdKafkaConsumerPoll-  , rdKafkaPosition-  , rdKafkaCommitted-  , rdKafkaSeek-  , rdKafkaResumePartitions-  , rdKafkaPausePartitions-  , rdKafkaSubscription-  , rdKafkaAssignment-  , rdKafkaConsumeBatchQueue-  , newRdKafkaTopicT-  )-import           Kafka.Internal.Setup-  ( Kafka(..)-  , KafkaConf(..)-  , TopicConf(..)-  , KafkaProps(..)-  , TopicProps(..)-  , kafkaConf-  , topicConf-  , getRdKafka-  )-import           Kafka.Internal.Shared-  ( kafkaErrorToMaybe-  , maybeToLeft-  , rdKafkaErrorToEither-  )+import           Control.Arrow              (left, (&&&))+import           Control.Concurrent         (forkIO, modifyMVar, rtsSupportsBoundThreads, withMVar)+import           Control.Exception          (bracket)+import           Control.Monad              (forM_, mapM_, void, when)+import           Control.Monad.IO.Class     (MonadIO (liftIO))+import           Control.Monad.Trans.Except (ExceptT (ExceptT), runExceptT)+import           Data.Bifunctor             (bimap, first)+import qualified Data.ByteString            as BS+import           Data.IORef                 (readIORef, writeIORef)+import qualified Data.Map                   as M+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, 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)+import           Kafka.Internal.Shared      (kafkaErrorToMaybe, maybeToLeft, rdKafkaErrorToEither)  import Kafka.Consumer.ConsumerProperties as X import Kafka.Consumer.Subscription       as X@@ -114,16 +69,19 @@             -> Subscription             -> m (Either KafkaError KafkaConsumer) newConsumer props (Subscription ts tp) = liftIO $ do-  let cp = setCallback (rebalanceCallback (\_ _ -> return ())) <> props-  kc@(KafkaConf kc' qref ct) <- newConsumerConf cp+  let cp = case cpCallbackPollMode props of+            CallbackPollModeAsync -> setCallback (rebalanceCallback (\_ _ -> return ())) <> props+            CallbackPollModeSync  -> props+  kc@(KafkaConf kc' qref _) <- newConsumerConf cp   tp' <- topicConf (TopicProps tp)   _   <- setDefaultTopicConf kc tp'   rdk <- newRdKafkaT RdKafkaConsumer kc'   case rdk of     Left err   -> return . Left $ KafkaError err     Right rdk' -> do-      msgq <- rdKafkaQueueNew rdk'-      writeIORef qref (Just msgq)+      when (cpCallbackPollMode props == CallbackPollModeAsync) $ do+        msgq <- rdKafkaQueueNew rdk'+        writeIORef qref (Just msgq)       let kafka = KafkaConsumer (Kafka rdk') kc       redErr <- redirectCallbacksPoll kafka       case redErr of@@ -132,7 +90,8 @@           forM_ (cpLogLevel cp) (setConsumerLogLevel kafka)           sub <- subscribe kafka ts           case sub of-            Nothing  -> runConsumerLoop kafka ct (Just $ Timeout 100) >> return (Right kafka)+            Nothing  -> (when (cpCallbackPollMode props == CallbackPollModeAsync) $+              runConsumerLoop kafka (Just $ Timeout 100)) >> return (Right kafka)             Just err -> closeConsumer kafka >> return (Left err)  pollMessage :: MonadIO m@@ -140,15 +99,16 @@             -> 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 _ (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."+    Nothing -> rdKafkaConsumerPoll (getRdKafka c) ms >>= fromMessagePtr     Just q  -> rdKafkaConsumeQueue q (fromIntegral ms) >>= fromMessagePtr  -- | Polls up to BatchSize messages. -- Unlike 'pollMessage' this function does not return usual "timeout" errors. -- An empty batch is returned when there are no messages available.+--+-- This API is not available when 'userPolls' is set. pollMessageBatch :: MonadIO m                  => KafkaConsumer                  -> Timeout@@ -158,7 +118,7 @@   pollConsumerEvents c Nothing   mbq <- readIORef qr   case mbq of-    Nothing -> return [Left $ KafkaBadSpecification "Messages queue is not configured, internal error, fatal."]+    Nothing -> return [Left $ KafkaBadSpecification "userPolls is set when calling pollMessageBatch."]     Just q  -> rdKafkaConsumeBatchQueue q ms b >>= traverse fromMessagePtr  -- | Commit message's offset on broker for the message's partition.@@ -244,8 +204,7 @@   where     seekAll = runExceptT $ do       tr <- traverse (ExceptT . topicPair) tps-      void $ traverse (\(kt, p, o) -> ExceptT (rdSeek kt p o)) tr-      return ()+      mapM_ (\(kt, p, o) -> ExceptT (rdSeek kt p o)) tr      rdSeek kt (PartitionId p) o =       rdKafkaErrorToEither <$> rdKafkaSeek kt (fromIntegral p) (offsetToInt64 o) timeout@@ -291,17 +250,20 @@ -- frequent enough. pollConsumerEvents :: KafkaConsumer -> Maybe Timeout -> IO () pollConsumerEvents k timeout =-  let (Timeout tm) = fromMaybe (Timeout 0) timeout-  in void $ rdKafkaConsumerPoll (getRdKafka k) tm+  void . withCallbackPollEnabled k $ pollConsumerEvents' k timeout  -- | Closes the consumer. closeConsumer :: MonadIO m => KafkaConsumer -> m (Maybe KafkaError)-closeConsumer (KafkaConsumer (Kafka k) (KafkaConf _ qr ct)) = liftIO $ do-  CToken.cancel ct-  mbq <- readIORef qr-  void $ traverse rdKafkaQueueDestroy mbq-  kafkaErrorToMaybe . KafkaResponseError <$> rdKafkaConsumerClose k-+closeConsumer (KafkaConsumer (Kafka k) (KafkaConf _ qr statusVar)) = liftIO $+  -- because closing the consumer will raise callbacks,+  -- prevent the async loop from doing it at the same time+  modifyMVar statusVar $ \_ -> do+    -- librdkafka says:+    --   Prior to destroying the client instance, loose your reference to the+    --   background queue by calling rd_kafka_queue_destroy()+    readIORef qr >>= mapM_ rdKafkaQueueDestroy+    res <- kafkaErrorToMaybe . KafkaResponseError <$> rdKafkaConsumerClose k+    pure (CallbackPollDisabled, res) ----------------------------------------------------------------------------- newConsumerConf :: ConsumerProperties -> IO KafkaConf newConsumerConf ConsumerProperties {cpProps = m, cpCallbacks = cbs} = do@@ -342,13 +304,24 @@ redirectCallbacksPoll (KafkaConsumer (Kafka k) _) =   kafkaErrorToMaybe . KafkaResponseError <$> rdKafkaPollSetConsumer k -runConsumerLoop :: KafkaConsumer -> CancellationToken -> Maybe Timeout -> IO ()-runConsumerLoop k ct timeout =+runConsumerLoop :: KafkaConsumer -> Maybe Timeout -> IO ()+runConsumerLoop k timeout =   when rtsSupportsBoundThreads $ void $ forkIO go   where     go = do-      token <- CToken.status ct-      case token of-        Running   -> pollConsumerEvents k timeout >> go-        Cancelled -> return ()+      st <- withCallbackPollEnabled k (pollConsumerEvents' k timeout)+      case st of+        CallbackPollEnabled  -> go+        CallbackPollDisabled -> pure () +withCallbackPollEnabled :: KafkaConsumer -> IO () -> IO CallbackPollStatus+withCallbackPollEnabled k f = do+  let statusVar = kcfgCallbackPollStatus (getKafkaConf k)+  withMVar statusVar $ \case+   CallbackPollEnabled  -> f >> pure CallbackPollEnabled+   CallbackPollDisabled -> pure CallbackPollDisabled++pollConsumerEvents' :: KafkaConsumer -> Maybe Timeout -> IO ()+pollConsumerEvents' k timeout =+  let (Timeout tm) = fromMaybe (Timeout 0) timeout+  in void $ rdKafkaConsumerPoll (getRdKafka k) tm
src/Kafka/Consumer/Callbacks.hs view
@@ -6,34 +6,21 @@ ) where -import           Control.Arrow          ((&&&))-import           Control.Concurrent     (threadDelay)-import           Control.Monad          (forM_, void)-import           Data.Monoid            ((<>))-import qualified Data.Text              as Text-import           Foreign.ForeignPtr     (newForeignPtr_)-import           Foreign.Ptr            (nullPtr)-import           Kafka.Callbacks        as X-import           Kafka.Consumer.Convert (fromNativeTopicPartitionList', fromNativeTopicPartitionList'', toNativeTopicPartitionList)-import           Kafka.Consumer.Types   (KafkaConsumer(..), TopicPartition(..), RebalanceEvent(..))-import           Kafka.Internal.RdKafka-import           Kafka.Internal.Setup   (Kafka(..), KafkaConf(..), HasKafka(..), HasKafkaConf(..), getRdMsgQueue)-import           Kafka.Internal.Shared  (kafkaErrorToMaybe)-import           Kafka.Types            (KafkaError(..), PartitionId(..), TopicName(..))+import Control.Arrow          ((&&&))+import Control.Monad          (forM_, void)+import Data.Monoid            ((<>))+import Foreign.ForeignPtr     (newForeignPtr_)+import Foreign.Ptr            (nullPtr)+import Kafka.Callbacks        as X+import Kafka.Consumer.Convert (fromNativeTopicPartitionList', fromNativeTopicPartitionList'')+import Kafka.Consumer.Types   (KafkaConsumer (..), RebalanceEvent (..), TopicPartition (..))+import Kafka.Internal.RdKafka+import Kafka.Internal.Setup   (HasKafka (..), HasKafkaConf (..), Kafka (..), KafkaConf (..), getRdMsgQueue)+import Kafka.Types            (KafkaError (..), PartitionId (..), TopicName (..)) +import qualified Data.Text as Text+ -- | Sets a callback that is called when rebalance is needed.------ Callback implementations suppose to watch for 'KafkaResponseError' 'RdKafkaRespErrAssignPartitions' and--- for 'KafkaResponseError' 'RdKafkaRespErrRevokePartitions'. Other error codes are not expected and would indicate--- something really bad happening in a system, or bugs in @librdkafka@ itself.------ A callback is expected to call 'assign' according to the error code it receives.------     * When 'RdKafkaRespErrAssignPartitions' happens 'assign' should be called with all the partitions it was called with.---       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 :: (KafkaConsumer -> RebalanceEvent -> IO ()) -> KafkaConf -> IO () rebalanceCallback callback kc@(KafkaConf conf _ _) = rdKafkaConfSetRebalanceCb conf realCb   where@@ -47,8 +34,6 @@ -- The results of automatic or manual offset commits will be scheduled -- for this callback and is served by `pollMessage`. ----- A callback is expected to call 'assign' according to the error code it receives.--- -- If no partitions had valid offsets to commit this callback will be called -- with `KafkaError` == `KafkaResponseError` `RdKafkaRespErrNoOffset` which is not to be considered -- an error.@@ -75,41 +60,29 @@ setRebalanceCallback f k e pls = do   ps <- fromNativeTopicPartitionList'' pls   let assignment = (tpTopicName &&& tpPartition) <$> ps+  let (Kafka kptr) = getKafka k+   case e of     KafkaResponseError RdKafkaRespErrAssignPartitions -> do+        f k (RebalanceBeforeAssign assignment)+        void $ rdKafkaAssign kptr pls+         mbq <- getRdMsgQueue $ getKafkaConf k         case mbq of           Nothing -> pure ()           Just mq -> do+            {- Magnus Edenhill:+                If you redirect after assign() it means some messages may be forwarded to the single consumer queue,+                so either do it before assign() or do: assign(); pause(); redirect; resume()+            -}+            void $ rdKafkaPausePartitions kptr pls             forM_ ps (\tp -> redirectPartitionQueue (getKafka k) (tpTopicName tp) (tpPartition tp) mq)-            -- sleep for 1 second.-            -- it looks like without it there is not enough time for redirect to happen-            -- or something similarly strange. I don't understand it.-            -- If you know WTH is going on PLEASE let me know because the current "fix" is ugly-            -- and is completely unreasonable :(-            threadDelay 1000000-        f k (RebalanceBeforeAssign assignment)-        void $ assign' k pls -- pass as pointer to avoid possible serialisation issues+            void $ rdKafkaResumePartitions kptr pls+         f k (RebalanceAssign assignment)+     KafkaResponseError RdKafkaRespErrRevokePartitions -> do         f k (RebalanceBeforeRevoke assignment)-        void $ assign k []+        void $ newForeignPtr_ nullPtr >>= rdKafkaAssign kptr         f k (RebalanceRevoke assignment)     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---- | Assigns specified partitions to a current consumer.--- Assigning an empty list means unassigning from all partitions that are currently assigned.-assign' :: KafkaConsumer -> RdKafkaTopicPartitionListTPtr -> IO (Maybe KafkaError)-assign' (KafkaConsumer (Kafka k) _) pls =-    (kafkaErrorToMaybe . KafkaResponseError) <$> rdKafkaAssign k pls-
src/Kafka/Consumer/ConsumerProperties.hs view
@@ -2,7 +2,9 @@  module Kafka.Consumer.ConsumerProperties ( ConsumerProperties(..)+, CallbackPollMode(..) , brokersList+, autoCommit , noAutoCommit , noAutoOffsetStore , groupId@@ -15,40 +17,45 @@ , extraProp , debugOptions , queuedMaxMessagesKBytes+, callbackPollMode , module X ) where -import           Control.Monad        (MonadPlus(mplus))+import           Control.Monad        (MonadPlus (mplus)) import           Data.Map             (Map) import qualified Data.Map             as M import           Data.Semigroup       as Sem import           Data.Text            (Text) import qualified Data.Text            as Text-import           Kafka.Consumer.Types (ConsumerGroupId(..))-import           Kafka.Internal.Setup (KafkaConf(..))-import           Kafka.Types          (KafkaDebug(..), KafkaCompressionCodec(..), KafkaLogLevel(..), ClientId(..), BrokerAddress(..), kafkaDebugToText, kafkaCompressionCodecToText)+import           Kafka.Consumer.Types (ConsumerGroupId (..))+import           Kafka.Internal.Setup (KafkaConf (..))+import           Kafka.Types          (BrokerAddress (..), ClientId (..), KafkaCompressionCodec (..), KafkaDebug (..), KafkaLogLevel (..), Millis (..), kafkaCompressionCodecToText, kafkaDebugToText)  import Kafka.Consumer.Callbacks as X +data CallbackPollMode = CallbackPollModeSync | CallbackPollModeAsync deriving (Show, Eq)+ -- | Properties to create 'KafkaConsumer'. data ConsumerProperties = ConsumerProperties-  { cpProps     :: Map Text Text-  , cpLogLevel  :: Maybe KafkaLogLevel-  , cpCallbacks :: [KafkaConf -> IO ()]+  { cpProps            :: Map Text Text+  , cpLogLevel         :: Maybe KafkaLogLevel+  , cpCallbacks        :: [KafkaConf -> IO ()]+  , cpCallbackPollMode :: CallbackPollMode   }  instance Sem.Semigroup ConsumerProperties where-  (ConsumerProperties m1 ll1 cb1) <> (ConsumerProperties m2 ll2 cb2) =-    ConsumerProperties (M.union m2 m1) (ll2 `mplus` ll1) (cb1 `mplus` cb2)+  (ConsumerProperties m1 ll1 cb1 _) <> (ConsumerProperties m2 ll2 cb2 cup2) =+    ConsumerProperties (M.union m2 m1) (ll2 `mplus` ll1) (cb1 `mplus` cb2) cup2   {-# INLINE (<>) #-}  -- | /Right biased/ so we prefer newer properties over older ones. instance Monoid ConsumerProperties where   mempty = ConsumerProperties-    { cpProps          = M.empty-    , cpLogLevel       = Nothing-    , cpCallbacks      = []+    { cpProps             = M.empty+    , cpLogLevel          = Nothing+    , cpCallbacks         = []+    , cpCallbackPollMode  = CallbackPollModeAsync     }   {-# INLINE mempty #-}   mappend = (Sem.<>)@@ -59,6 +66,13 @@   let bs' = Text.intercalate "," ((\(BrokerAddress x) -> x) <$> bs)    in extraProps $ M.fromList [("bootstrap.servers", bs')] +autoCommit :: Millis -> ConsumerProperties+autoCommit (Millis ms) = extraProps $+  M.fromList+    [ ("enable.auto.commit", "true")+    , ("auto.commit.interval.ms", Text.pack $ show ms)+    ]+ -- | Disables auto commit for the consumer noAutoCommit :: ConsumerProperties noAutoCommit =@@ -123,3 +137,17 @@ queuedMaxMessagesKBytes kBytes =   extraProp "queued.max.messages.kbytes" (Text.pack $ show kBytes) {-# INLINE queuedMaxMessagesKBytes #-}++-- | Sets the callback poll mode.+--+-- The default 'CallbackPollModeAsync' mode handles polling rebalance+-- and keep alive events for you+-- in a background thread.+--+-- With 'CallbacPollModeSync' the user will poll the consumer+-- frequently to handle new messages as well as rebalance and keep alive events.+-- 'CallbacPollModeSync' lets you can simplify+-- hw-kafka-client's footprint and have full control over when polling+-- happens at the cost of having to manage this yourself.+callbackPollMode :: CallbackPollMode -> ConsumerProperties+callbackPollMode mode = mempty { cpCallbackPollMode = mode }
src/Kafka/Consumer/Subscription.hs view
@@ -4,20 +4,18 @@ ( Subscription(..) , topics , offsetReset-, autoCommit , extraSubscriptionProps ) where -import qualified Data.Text            as Text-import           Data.Text            (Text) import           Data.Map             (Map) import qualified Data.Map             as M import           Data.Semigroup       as Sem-import           Kafka.Consumer.Types (OffsetReset(..))-import           Kafka.Types          (TopicName(..), Millis(..)) import           Data.Set             (Set) import qualified Data.Set             as Set+import           Data.Text            (Text)+import           Kafka.Consumer.Types (OffsetReset (..))+import           Kafka.Types          (TopicName (..))  data Subscription = Subscription (Set TopicName) (Map Text Text) @@ -42,14 +40,7 @@   let o' = case o of              Earliest -> "earliest"              Latest   -> "latest"-   in Subscription (Set.empty) (M.fromList [("auto.offset.reset", o')])--autoCommit :: Millis -> Subscription-autoCommit (Millis ms) = Subscription (Set.empty) $-  M.fromList-    [ ("enable.auto.commit", "true")-    , ("auto.commit.interval.ms", Text.pack $ show ms)-    ]+   in Subscription Set.empty (M.fromList [("auto.offset.reset", o')])  extraSubscriptionProps :: Map Text Text -> Subscription-extraSubscriptionProps = Subscription (Set.empty)+extraSubscriptionProps = Subscription Set.empty
− src/Kafka/Internal/CancellationToken.hs 
@@ -1,23 +0,0 @@-module Kafka.Internal.CancellationToken-( CancellationStatus(..)-, CancellationToken(..)-, newCancellationToken-, status-, cancel-)-where--import Data.IORef (IORef, newIORef, readIORef, atomicWriteIORef)--data CancellationStatus = Cancelled | Running deriving (Show, Eq)-newtype CancellationToken = CancellationToken (IORef CancellationStatus)--newCancellationToken :: IO CancellationToken-newCancellationToken = CancellationToken <$> newIORef Running--status :: CancellationToken -> IO CancellationStatus-status (CancellationToken ref) = readIORef ref--cancel :: CancellationToken -> IO ()-cancel (CancellationToken ref) = atomicWriteIORef ref Cancelled-
src/Kafka/Internal/RdKafka.chs view
@@ -838,6 +838,9 @@         withForeignPtr ret $ \realPtr -> do             if realPtr == nullPtr then peekCText charPtr >>= return . Left             else do+                -- do not call 'rd_kafka_close_consumer' on destroying all Kafka.+                -- when needed, applications should do it explicitly.+                -- {# call rd_kafka_destroy_flags #} realPtr 0x8                 addForeignPtrFinalizer rdKafkaDestroy ret                 return $ Right ret
src/Kafka/Internal/Setup.hs view
@@ -7,6 +7,7 @@ , HasKafka(..) , HasKafkaConf(..) , HasTopicConf(..)+, CallbackPollStatus(..) , getRdKafka , getRdKafkaConf , getRdMsgQueue@@ -23,18 +24,19 @@ ) where -import Kafka.Internal.RdKafka (RdKafkaConfResT(..), CCharBufPointer, RdKafkaQueueTPtr, RdKafkaTPtr, RdKafkaConfTPtr, RdKafkaTopicConfTPtr, nErrorBytes, rdKafkaTopicConfSet, newRdKafkaTopicConfT, newRdKafkaConfT, rdKafkaConfSet)-import Kafka.Types (KafkaError(..))+import Kafka.Internal.RdKafka (CCharBufPointer, RdKafkaConfResT (..), RdKafkaConfTPtr, RdKafkaQueueTPtr, RdKafkaTPtr, RdKafkaTopicConfTPtr, nErrorBytes, newRdKafkaConfT, newRdKafkaTopicConfT, rdKafkaConfSet, rdKafkaTopicConfSet)+import Kafka.Types            (KafkaError (..)) -import Control.Exception (throw)-import Data.IORef (IORef, newIORef, readIORef)-import Foreign.Marshal.Alloc (allocaBytes)-import Foreign.C.String (peekCString)-import Kafka.Internal.CancellationToken (CancellationToken(..), newCancellationToken)+import Control.Concurrent.MVar (MVar, newMVar)+import Control.Exception       (throw)+import Data.IORef              (IORef, newIORef, readIORef)+import Data.Map                (Map)+import Data.Text               (Text)+import Foreign.C.String        (peekCString)+import Foreign.Marshal.Alloc   (allocaBytes)++import qualified Data.Map  as Map import qualified Data.Text as Text-import Data.Map (Map)-import Data.Text (Text)-import qualified Data.Map as Map  -- -- Configuration@@ -42,9 +44,22 @@ newtype KafkaProps = KafkaProps (Map Text Text) deriving (Show, Eq) newtype TopicProps = TopicProps (Map Text Text) deriving (Show, Eq) newtype Kafka      = Kafka RdKafkaTPtr deriving Show-data KafkaConf     = KafkaConf RdKafkaConfTPtr (IORef (Maybe RdKafkaQueueTPtr)) CancellationToken newtype TopicConf  = TopicConf RdKafkaTopicConfTPtr deriving Show +data CallbackPollStatus = CallbackPollEnabled | CallbackPollDisabled deriving (Show, Eq)++data KafkaConf         = KafkaConf+  { kcfgKafkaConfPtr       :: RdKafkaConfTPtr+    -- ^ A pointer to a native Kafka configuration++  , kcfgMessagesQueue      :: IORef (Maybe RdKafkaQueueTPtr)+    -- ^ A queue for messages++  , kcfgCallbackPollStatus :: MVar CallbackPollStatus+    -- ^ A mutex to prevent handling callbacks from multiple threads+    -- which can be dangerous in some cases.+  }+ class HasKafka a where   getKafka :: a -> Kafka @@ -87,7 +102,7 @@ newTopicConf = TopicConf <$> newRdKafkaTopicConfT  newKafkaConf :: IO KafkaConf-newKafkaConf = KafkaConf <$> newRdKafkaConfT <*> newIORef Nothing <*> newCancellationToken+newKafkaConf = KafkaConf <$> newRdKafkaConfT <*> newIORef Nothing <*> newMVar CallbackPollEnabled  kafkaConf :: KafkaProps -> IO KafkaConf kafkaConf overrides = do
src/Kafka/Internal/Shared.hs view
@@ -1,6 +1,5 @@ module Kafka.Internal.Shared-( runEventLoop-, pollEvents+( pollEvents , word8PtrToBS , kafkaRespErr , throwOnError@@ -17,34 +16,22 @@ ) where -import           Data.Text                        (Text)-import qualified Data.Text                        as Text-import           Control.Concurrent               (forkIO, rtsSupportsBoundThreads)-import           Control.Exception                (throw)-import           Control.Monad                    (void, when)-import qualified Data.ByteString                  as BS-import qualified Data.ByteString.Internal         as BSI-import           Data.Word                        (Word8)-import           Foreign.Ptr                      (Ptr, nullPtr)-import           Foreign.Marshal.Alloc            (alloca)-import           Foreign.ForeignPtr               (newForeignPtr_)-import           Foreign.Storable                 (Storable(peek))-import           Foreign.C.Error                  (Errno(..))-import           Kafka.Consumer.Types             (Timestamp(..))-import           Kafka.Internal.CancellationToken as CToken-import           Kafka.Internal.RdKafka           (RdKafkaTimestampTypeT(..), RdKafkaMessageTPtr, RdKafkaMessageT(..), RdKafkaRespErrT(..), Word8Ptr, rdKafkaPoll, rdKafkaErrno2err, rdKafkaTopicName, rdKafkaMessageTimestamp)-import           Kafka.Internal.Setup             (HasKafka(..), Kafka(..))-import           Kafka.Types                      (KafkaError(..), Timeout(..), Millis(..))--runEventLoop :: HasKafka a => a -> CancellationToken -> Maybe Timeout -> IO ()-runEventLoop k ct timeout =-    when rtsSupportsBoundThreads $ void $ forkIO go-    where-        go = do-            token <- CToken.status ct-            case token of-                Running   -> pollEvents k timeout >> go-                Cancelled -> return ()+import           Control.Exception        (throw)+import           Control.Monad            (void)+import qualified Data.ByteString          as BS+import qualified Data.ByteString.Internal as BSI+import           Data.Text                (Text)+import qualified Data.Text                as Text+import           Data.Word                (Word8)+import           Foreign.C.Error          (Errno (..))+import           Foreign.ForeignPtr       (newForeignPtr_)+import           Foreign.Marshal.Alloc    (alloca)+import           Foreign.Ptr              (Ptr, nullPtr)+import           Foreign.Storable         (Storable (peek))+import           Kafka.Consumer.Types     (Timestamp (..))+import           Kafka.Internal.RdKafka   (RdKafkaMessageT (..), RdKafkaMessageTPtr, RdKafkaRespErrT (..), RdKafkaTimestampTypeT (..), Word8Ptr, rdKafkaErrno2err, rdKafkaMessageTimestamp, rdKafkaPoll, rdKafkaTopicName)+import           Kafka.Internal.Setup     (HasKafka (..), Kafka (..))+import           Kafka.Types              (KafkaError (..), Millis (..), Timeout (..))  pollEvents :: HasKafka a => a -> Maybe Timeout -> IO () pollEvents a tm =
src/Kafka/Producer.hs view
@@ -11,37 +11,25 @@ ) where -import           Control.Arrow                    ((&&&))-import           Control.Exception                (bracket)-import           Control.Monad                    (forM, forM_, (<=<))-import           Control.Monad.IO.Class           (MonadIO(liftIO))-import qualified Data.ByteString                  as BS-import qualified Data.ByteString.Internal         as BSI-import           Data.Function                    (on)-import           Data.List                        (groupBy, sortBy)-import           Data.Ord                         (comparing)-import qualified Data.Text                        as Text-import           Foreign.ForeignPtr               (withForeignPtr, newForeignPtr_)-import           Foreign.Marshal.Array            (withArrayLen)-import           Foreign.Ptr                      (Ptr, nullPtr, plusPtr)-import           Foreign.Storable                 (Storable(..))-import           Kafka.Internal.CancellationToken as CToken-import           Kafka.Internal.RdKafka-  ( RdKafkaMessageT(..)-  , RdKafkaTypeT(..)-  , RdKafkaRespErrT(..)-  , newRdKafkaT-  , newUnmanagedRdKafkaTopicT-  , destroyUnmanagedRdKafkaTopic-  , rdKafkaProduce-  , rdKafkaSetLogLevel-  , rdKafkaProduceBatch-  , rdKafkaOutqLen-  )-import           Kafka.Internal.Setup             (KafkaConf(..), Kafka(..), TopicConf(..), kafkaConf, KafkaProps(..), topicConf, TopicProps(..))-import           Kafka.Internal.Shared            (pollEvents)-import           Kafka.Producer.Convert           (copyMsgFlags, producePartitionCInt, producePartitionInt, handleProduceErr)-import           Kafka.Producer.Types             (KafkaProducer(..))+import           Control.Arrow            ((&&&))+import           Control.Exception        (bracket)+import           Control.Monad            (forM, forM_, (<=<))+import           Control.Monad.IO.Class   (MonadIO (liftIO))+import qualified Data.ByteString          as BS+import qualified Data.ByteString.Internal as BSI+import           Data.Function            (on)+import           Data.List                (groupBy, sortBy)+import           Data.Ord                 (comparing)+import qualified Data.Text                as Text+import           Foreign.ForeignPtr       (newForeignPtr_, withForeignPtr)+import           Foreign.Marshal.Array    (withArrayLen)+import           Foreign.Ptr              (Ptr, nullPtr, plusPtr)+import           Foreign.Storable         (Storable (..))+import           Kafka.Internal.RdKafka   (RdKafkaMessageT (..), RdKafkaRespErrT (..), RdKafkaTypeT (..), destroyUnmanagedRdKafkaTopic, newRdKafkaT, newUnmanagedRdKafkaTopicT, rdKafkaOutqLen, rdKafkaProduce, rdKafkaProduceBatch, rdKafkaSetLogLevel)+import           Kafka.Internal.Setup     (Kafka (..), KafkaConf (..), KafkaProps (..), TopicConf (..), TopicProps (..), kafkaConf, topicConf)+import           Kafka.Internal.Shared    (pollEvents)+import           Kafka.Producer.Convert   (copyMsgFlags, handleProduceErr, producePartitionCInt, producePartitionInt)+import           Kafka.Producer.Types     (KafkaProducer (..))  import Kafka.Producer.ProducerProperties as X import Kafka.Producer.Types              as X hiding (KafkaProducer)@@ -163,9 +151,7 @@ -- | Closes the producer. -- Will wait until the outbound queue is drained before returning the control. closeProducer :: MonadIO m => KafkaProducer -> m ()-closeProducer p =-  let (KafkaConf _ _ ct) = kpKafkaConf p-  in liftIO (CToken.cancel ct) >> flushProducer p+closeProducer = flushProducer  -- | Drains the outbound queue for a producer. --  This function is also called automatically when the producer is closed
tests-it/Kafka/IntegrationSpec.hs view
@@ -113,103 +113,8 @@                 res    <- sendMessages (testMessages testTopic) prod                 res `shouldBe` Right () -        specWithConsumer "Run consumer" consumerProps $ do-            it "should get committed" $ \k -> do-                res <- committed k (Timeout 1000) [(testTopic, PartitionId 0)]-                res `shouldSatisfy` isRight--            it "should get position" $ \k -> do-                res <- position k [(testTopic, PartitionId 0)]-                res `shouldSatisfy` isRight--            it "should receive messages" $ \k -> do-                res <- receiveMessages k-                length <$> res `shouldBe` Right 2--                let timestamps = crTimestamp <$> either (const []) id res-                forM_ timestamps $ \ts ->-                    ts `shouldNotBe` NoTimestamp--                comRes <- commitAllOffsets OffsetCommit k-                comRes `shouldBe` Nothing--            it "should get watermark offsets" $ \k -> do-                res <- sequence <$> watermarkOffsets k (Timeout 1000) testTopic-                res `shouldSatisfy` isRight-                length <$> res `shouldBe` (Right 1)--            it "should return subscription" $ \k -> do-                res <- subscription k-                res `shouldSatisfy` isRight-                length <$> res `shouldBe` Right 1--            it "should return assignment" $ \k -> do-                res <- assignment k-                res `shouldSatisfy` isRight-                res `shouldBe` Right (fromList [(testTopic, [PartitionId 0])])--            it "should return all topics metadata" $ \k -> do-                res <- allTopicsMetadata k (Timeout 1000)-                res `shouldSatisfy` isRight-                let filterUserTopics m = m { kmTopics = filter (\t -> topicType (tmTopicName t) == User) (kmTopics m) }-                let res' = fmap filterUserTopics res-                length . kmBrokers <$> res' `shouldBe` Right 1-                length . kmTopics  <$> res' `shouldBe` Right 1--            it "should return topic metadata" $ \k -> do-                res <- topicMetadata k (Timeout 1000) 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)-                fmap giGroup <$> res `shouldBe` Right [testGroupId]--            it "should describe a given consumer group" $ \k -> do-                res <- consumerGroupInfo k (Timeout 1000) testGroupId-                fmap giGroup <$> res `shouldBe` Right [testGroupId]--            it "should describe non-existent consumer group" $ \k -> do-                res <- consumerGroupInfo k (Timeout 1000) (ConsumerGroupId "does-not-exist")-                res `shouldBe` Right []--            it "should read topic offsets for time" $ \k -> do-                res <- topicOffsetsForTime k (Timeout 1000) (Millis 1904057189508) testTopic-                res `shouldSatisfy` isRight-                fmap tpOffset <$> res `shouldBe` Right [PartitionOffsetEnd]--            it "should seek and return no error" $ \k -> do-                res <- seek k (Timeout 1000) [TopicPartition testTopic (PartitionId 0) (PartitionOffset 1)]-                res `shouldBe` Nothing-                msg <- pollMessage k (Timeout 1000)-                crOffset <$> msg `shouldBe` Right (Offset 1)--            it "should seek to the beginning" $ \k -> do-                res <- seek k (Timeout 1000) [TopicPartition testTopic (PartitionId 0) PartitionOffsetBeginning]-                res `shouldBe` Nothing-                msg <- pollMessage k (Timeout 1000)-                crOffset <$> msg `shouldBe` Right (Offset 0)--            it "should seek to the end" $ \k -> do-                res <- seek k (Timeout 1000) [TopicPartition testTopic (PartitionId 0) PartitionOffsetEnd]-                res `shouldBe` Nothing-                msg <- pollMessage k (Timeout 1000)-                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]-                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 `shouldBe` Nothing-                msg <- pollMessage k (Timeout 1000)-                crOffset <$> msg `shouldBe` Right (Offset 0)+        specWithConsumer "Run consumer with async polling" (consumerProps <> groupId (makeGroupId "async")) runConsumerSpec+        specWithConsumer "Run consumer with sync polling" (consumerProps <> groupId (makeGroupId "sync") <> callbackPollMode CallbackPollModeSync) runConsumerSpec      describe "Kafka.Consumer.BatchSpec" $ do         specWithConsumer "Batch consumer" (consumerProps <> groupId (ConsumerGroupId "batch-consumer")) $ do@@ -254,3 +159,112 @@ sendMessages :: [ProducerRecord] -> KafkaProducer -> IO (Either KafkaError ()) sendMessages msgs prod =   Right <$> (forM_ msgs (produceMessage prod) >> flushProducer prod)++runConsumerSpec :: SpecWith KafkaConsumer+runConsumerSpec = do+  it "should receive messages" $ \k -> do+    res <- receiveMessages k+    let msgsLen = either (const 0) length res+    msgsLen `shouldSatisfy` (> 0)++    let timestamps = crTimestamp <$> either (const []) id res+    forM_ timestamps $ \ts ->+      ts `shouldNotBe` NoTimestamp++    comRes <- commitAllOffsets OffsetCommit k+    comRes `shouldBe` Nothing++  it "should get committed" $ \k -> do+    res <- committed k (Timeout 1000) [(testTopic, PartitionId 0)]+    res `shouldSatisfy` isRight++  it "should get position" $ \k -> do+    res <- position k [(testTopic, PartitionId 0)]+    res `shouldSatisfy` isRight++  it "should get watermark offsets" $ \k -> do+    res <- sequence <$> watermarkOffsets k (Timeout 1000) testTopic+    res `shouldSatisfy` isRight+    length <$> res `shouldBe` (Right 1)++  it "should return subscription" $ \k -> do+    res <- subscription k+    res `shouldSatisfy` isRight+    length <$> res `shouldBe` Right 1++  it "should return assignment" $ \k -> do+    res <- assignment k+    res `shouldSatisfy` isRight+    res `shouldBe` Right (fromList [(testTopic, [PartitionId 0])])++  it "should return all topics metadata" $ \k -> do+    res <- allTopicsMetadata k (Timeout 1000)+    res `shouldSatisfy` isRight+    let filterUserTopics m = m { kmTopics = filter (\t -> topicType (tmTopicName t) == User) (kmTopics m) }+    let res' = fmap filterUserTopics res+    length . kmBrokers <$> res' `shouldBe` Right 1++    let topicsLen = either (const 0) (length . kmTopics) res'+    let hasTopic = either (const False) (any (\t -> tmTopicName t == testTopic) . kmTopics) res'++    topicsLen `shouldSatisfy` (>0)+    hasTopic `shouldBe` True++  it "should return topic metadata" $ \k -> do+    res <- topicMetadata k (Timeout 1000) 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)+    let groups = either (const []) (fmap giGroup) res+    let prefixedGroups = filter isTestGroupId groups+    let resLen = length prefixedGroups+    resLen `shouldSatisfy` (>0)+    -- fmap giGroup <$> res `shouldBe` Right [testGroupId]++  it "should describe a given consumer group" $ \k -> do+    res <- consumerGroupInfo k (Timeout 1000) testGroupId+    fmap giGroup <$> res `shouldBe` Right [testGroupId]++  it "should describe non-existent consumer group" $ \k -> do+    res <- consumerGroupInfo k (Timeout 1000) (ConsumerGroupId "does-not-exist")+    res `shouldBe` Right []++  it "should read topic offsets for time" $ \k -> do+    res <- topicOffsetsForTime k (Timeout 1000) (Millis 1904057189508) testTopic+    res `shouldSatisfy` isRight+    fmap tpOffset <$> res `shouldBe` Right [PartitionOffsetEnd]++  it "should seek and return no error" $ \k -> do+    res <- seek k (Timeout 1000) [TopicPartition testTopic (PartitionId 0) (PartitionOffset 1)]+    res `shouldBe` Nothing+    msg <- pollMessage k (Timeout 1000)+    crOffset <$> msg `shouldBe` Right (Offset 1)++  it "should seek to the beginning" $ \k -> do+    res <- seek k (Timeout 1000) [TopicPartition testTopic (PartitionId 0) PartitionOffsetBeginning]+    res `shouldBe` Nothing+    msg <- pollMessage k (Timeout 1000)+    crOffset <$> msg `shouldBe` Right (Offset 0)++  it "should seek to the end" $ \k -> do+    res <- seek k (Timeout 1000) [TopicPartition testTopic (PartitionId 0) PartitionOffsetEnd]+    res `shouldBe` Nothing+    msg <- pollMessage k (Timeout 1000)+    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]+    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 `shouldBe` Nothing+    msg <- pollMessage k (Timeout 1000)+    crOffset <$> msg `shouldBe` Right (Offset 0)
tests-it/Kafka/TestEnv.hs view
@@ -3,37 +3,50 @@  module Kafka.TestEnv where -import Control.Exception-import Control.Monad      (void)-import Data.Monoid        ((<>))-import System.Environment-import System.IO.Unsafe-import qualified Data.Text as Text+import           Control.Exception+import           Control.Monad      (void)+import           Data.Monoid        ((<>))+import qualified Data.Text          as Text+import           System.Environment+import           System.IO.Unsafe +import qualified System.Random as Rnd+ import Control.Concurrent import Kafka.Consumer     as C import Kafka.Producer     as P  import Test.Hspec +testPrefix :: String+testPrefix = unsafePerformIO $ take 10 . Rnd.randomRs ('a','z') <$> Rnd.newStdGen+{-# NOINLINE testPrefix #-}+ brokerAddress :: BrokerAddress-brokerAddress = unsafePerformIO $ do-  (BrokerAddress . Text.pack) <$> getEnv "KAFKA_TEST_BROKER" `catch` \(_ :: SomeException) -> (return "localhost:9092")+brokerAddress = unsafePerformIO $+  (BrokerAddress . Text.pack) <$> getEnv "KAFKA_TEST_BROKER" `catch` \(_ :: SomeException) -> return "localhost:9092" {-# NOINLINE brokerAddress #-}  testTopic :: TopicName-testTopic = unsafePerformIO $ do-  (TopicName . Text.pack) <$> getEnv "KAFKA_TEST_TOPIC" `catch` \(_ :: SomeException) -> (return "kafka-client_tests")+testTopic = unsafePerformIO $+  (TopicName . Text.pack) <$> getEnv "KAFKA_TEST_TOPIC" `catch` \(_ :: SomeException) -> return $ testPrefix <> "-topic" {-# NOINLINE testTopic #-}  testGroupId :: ConsumerGroupId-testGroupId = ConsumerGroupId "it_spec_03"+testGroupId = ConsumerGroupId (Text.pack testPrefix) +makeGroupId :: String -> ConsumerGroupId+makeGroupId suffix =+  ConsumerGroupId . Text.pack $ testPrefix <> "-" <> suffix++isTestGroupId :: ConsumerGroupId -> Bool+isTestGroupId (ConsumerGroupId group) = Text.pack testPrefix `Text.isPrefixOf` group+ consumerProps :: ConsumerProperties consumerProps =  C.brokersList [brokerAddress]               <> groupId testGroupId-              <> C.setCallback (logCallback (\l s1 s2 -> print $ show l <> ": " <> s1 <> ", " <> s2))-              <> C.setCallback (errorCallback (\e r -> print $ show e <> ": " <> r))+              <> C.setCallback (logCallback (\l s1 s2 -> print $ "[Consumer] " <> show l <> ": " <> s1 <> ", " <> s2))+              <> C.setCallback (errorCallback (\e r -> print $ "[Consumer] " <> show e <> ": " <> r))               <> noAutoCommit  consumerPropsNoStore :: ConsumerProperties@@ -41,8 +54,8 @@  producerProps :: ProducerProperties producerProps =  P.brokersList [brokerAddress]-              <> P.setCallback (logCallback (\l s1 s2 -> print $ show l <> ": " <> s1 <> ", " <> s2))-              <> P.setCallback (errorCallback (\e r -> print $ show e <> ": " <> r))+              <> P.setCallback (logCallback (\l s1 s2 -> print $ "[Producer] " <> show l <> ": " <> s1 <> ", " <> s2))+              <> P.setCallback (errorCallback (\e r -> print $ "[Producer] " <> show e <> ": " <> r))  testSubscription :: TopicName -> Subscription testSubscription t = topics [t]@@ -65,7 +78,10 @@   specWithConsumer :: String -> ConsumerProperties -> SpecWith KafkaConsumer -> Spec-specWithConsumer s p f = beforeAll (mkConsumerWith p) $ afterAll (void . closeConsumer) $ describe s f+specWithConsumer s p f =+  beforeAll (mkConsumerWith p)+  $ afterAll (void . closeConsumer)+  $ describe s f  specWithProducer :: String -> SpecWith KafkaProducer -> Spec specWithProducer s f = beforeAll mkProducer $ afterAll (void . closeProducer) $ describe s f