diff --git a/hw-kafka-client.cabal b/hw-kafka-client.cabal
--- a/hw-kafka-client.cabal
+++ b/hw-kafka-client.cabal
@@ -1,5 +1,5 @@
 name:           hw-kafka-client
-version:        2.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:
diff --git a/src/Kafka/Consumer.hs b/src/Kafka/Consumer.hs
--- a/src/Kafka/Consumer.hs
+++ b/src/Kafka/Consumer.hs
@@ -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
diff --git a/src/Kafka/Consumer/Callbacks.hs b/src/Kafka/Consumer/Callbacks.hs
--- a/src/Kafka/Consumer/Callbacks.hs
+++ b/src/Kafka/Consumer/Callbacks.hs
@@ -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
-
diff --git a/src/Kafka/Consumer/ConsumerProperties.hs b/src/Kafka/Consumer/ConsumerProperties.hs
--- a/src/Kafka/Consumer/ConsumerProperties.hs
+++ b/src/Kafka/Consumer/ConsumerProperties.hs
@@ -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 }
diff --git a/src/Kafka/Consumer/Subscription.hs b/src/Kafka/Consumer/Subscription.hs
--- a/src/Kafka/Consumer/Subscription.hs
+++ b/src/Kafka/Consumer/Subscription.hs
@@ -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
diff --git a/src/Kafka/Internal/CancellationToken.hs b/src/Kafka/Internal/CancellationToken.hs
deleted file mode 100644
--- a/src/Kafka/Internal/CancellationToken.hs
+++ /dev/null
@@ -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
-
diff --git a/src/Kafka/Internal/RdKafka.chs b/src/Kafka/Internal/RdKafka.chs
--- a/src/Kafka/Internal/RdKafka.chs
+++ b/src/Kafka/Internal/RdKafka.chs
@@ -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
 
diff --git a/src/Kafka/Internal/Setup.hs b/src/Kafka/Internal/Setup.hs
--- a/src/Kafka/Internal/Setup.hs
+++ b/src/Kafka/Internal/Setup.hs
@@ -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
diff --git a/src/Kafka/Internal/Shared.hs b/src/Kafka/Internal/Shared.hs
--- a/src/Kafka/Internal/Shared.hs
+++ b/src/Kafka/Internal/Shared.hs
@@ -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 =
diff --git a/src/Kafka/Producer.hs b/src/Kafka/Producer.hs
--- a/src/Kafka/Producer.hs
+++ b/src/Kafka/Producer.hs
@@ -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
diff --git a/tests-it/Kafka/IntegrationSpec.hs b/tests-it/Kafka/IntegrationSpec.hs
--- a/tests-it/Kafka/IntegrationSpec.hs
+++ b/tests-it/Kafka/IntegrationSpec.hs
@@ -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)
diff --git a/tests-it/Kafka/TestEnv.hs b/tests-it/Kafka/TestEnv.hs
--- a/tests-it/Kafka/TestEnv.hs
+++ b/tests-it/Kafka/TestEnv.hs
@@ -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