diff --git a/example/Main.hs b/example/Main.hs
--- a/example/Main.hs
+++ b/example/Main.hs
@@ -1,12 +1,12 @@
 {-# LANGUAGE OverloadedStrings #-}
 module Main where
 
-import Control.Monad (forM_)
-import Data.Monoid ((<>))
-import Kafka.Conduit.Source
-import Kafka.Conduit.Sink
-import Data.Conduit
-import qualified Data.Conduit.List as L
+import           Control.Monad        (forM_)
+import           Data.Conduit
+import qualified Data.Conduit.List    as L
+import           Data.Monoid          ((<>))
+import           Kafka.Conduit.Sink   as KSnk
+import           Kafka.Conduit.Source as KSrc
 
 kafkaBroker :: BrokerAddress
 kafkaBroker = BrokerAddress "localhost:9092"
@@ -17,7 +17,7 @@
 
 -- Global consumer properties
 consumerProps :: ConsumerProperties
-consumerProps = consumerBrokersList [kafkaBroker]
+consumerProps = KSrc.brokersList [kafkaBroker]
              <> groupId (ConsumerGroupId "consumer_conduit_example_group")
              <> noAutoCommit
 
@@ -28,7 +28,7 @@
 
 -- Global producer properties
 producerProps :: ProducerProperties
-producerProps = producerBrokersList [kafkaBroker]
+producerProps = KSnk.brokersList [kafkaBroker]
 
 main :: IO ()
 main = do
diff --git a/hw-kafka-conduit.cabal b/hw-kafka-conduit.cabal
--- a/hw-kafka-conduit.cabal
+++ b/hw-kafka-conduit.cabal
@@ -1,5 +1,5 @@
 name:                hw-kafka-conduit
-version:             2.0.0
+version:             2.1.0
 synopsis:            Conduit bindings for hw-kafka-client
 homepage:            https://github.com/haskell-works/hw-kafka-conduit
 bug-reports:         https://github.com/haskell-works/hw-kafka-conduit/issues
@@ -62,6 +62,7 @@
   type:                 exitcode-stdio-1.0
   hs-source-dirs:       test
   main-is:              Spec.hs
+  other-modules:        Kafka.Conduit.CombinatorSpec
   ghc-options:          -Wall -threaded -rtsopts -with-rtsopts=-N
   default-language:     Haskell2010
   build-depends:        base
diff --git a/src/Kafka/Conduit/Sink.hs b/src/Kafka/Conduit/Sink.hs
--- a/src/Kafka/Conduit/Sink.hs
+++ b/src/Kafka/Conduit/Sink.hs
@@ -8,6 +8,7 @@
 import           Control.Monad.IO.Class
 import           Control.Monad.Trans.Resource
 import           Data.Conduit
+import qualified Data.Conduit.List            as L
 import           Kafka.Consumer
 
 import           Kafka.Conduit.Combinators    as X
@@ -92,10 +93,49 @@
           res <- produceMessage prod msg
           runHandler $ maybe (Right prod) Left res
 
+-- | Ignores incoming messages and commits offsets. Commit errors are ignored.
+-- This functionality should not exist as a Sink and will be removed in future versions.
+-- Consider having an effect instead:
+--
+-- > mapMC (\_ -> commitAllOffsets OffsetCommit consumer)
+{-# DEPRECATED commitOffsetsSink "Conceptually wrong thing to do. Does not require library support. Consider calling 'commitAllOffsets' when appropriate." #-}
 commitOffsetsSink :: MonadIO m => KafkaConsumer -> Sink i m ()
-commitOffsetsSink = awaitForever . const . commitAllOffsets OffsetCommit
+commitOffsetsSink = flip commitOffsetsSink' (const $ pure ())
 
+-- | Ignores incoming messages and commits offsets. Commit errors are handled with 'handleError' effect.
+-- This functionality should not exist as a Sink and will be removed in future versions.
+-- Consider having an effect instead:
+--
+-- > mapMC (\_ -> commitAllOffsets OffsetCommit consumer >>= handleError)
+{-# DEPRECATED commitOffsetsSink' "Conceptually wrong thing to do. Does not require library support. Consider calling 'commitAllOffsets' when appropriate." #-}
+commitOffsetsSink':: MonadIO m => KafkaConsumer -> (KafkaError -> m ()) -> Sink i m ()
+commitOffsetsSink' consumer handleError = L.mapM_ $ \_ -> do
+  res <- commitAllOffsets OffsetCommit consumer
+  case res of
+    Nothing  -> pure ()
+    Just err -> handleError err
+
+-- | Ignores incoming messages and commits offsets, but makes sure that 'producer' has an empty outgoing queue.
+-- Commit errors are ignored.
+-- This functionality should not exist as a Sink and will be removed in future versions.
+-- Consider having an effect instead:
+--
+-- > mapMC (\_ -> flushProducer producer >>= commitAllOffsets OffsetCommit consumer)
+{-# DEPRECATED flushThenCommitSink "Conceptually wrong thing to do. Does not require library support. Consider calling 'flushProducer >>= commitAllOffsets' when appropriate." #-}
 flushThenCommitSink :: MonadIO m => KafkaConsumer -> KafkaProducer -> Sink i m ()
-flushThenCommitSink c p = awaitForever . const $ do
-  flushProducer p
-  commitAllOffsets OffsetCommit c
+flushThenCommitSink consumer producer = flushThenCommitSink' consumer producer (const $ pure ())
+
+-- | Ignores incoming messages and commits offsets, but makes sure that 'producer' has an empty outgoing queue.
+-- Commit errors are handled with 'handleError' effect.
+-- This functionality should not exist as a Sink and will be removed in future versions.
+-- Consider having an effect instead:
+--
+-- > mapMC (\_ -> flushProducer producer >>= commitAllOffsets OffsetCommit consumer >>= handleError)
+{-# DEPRECATED flushThenCommitSink' "Conceptually wrong thing to do. Does not require library support. Consider calling 'flushProducer >>= commitAllOffsets' when appropriate." #-}
+flushThenCommitSink' :: MonadIO m => KafkaConsumer -> KafkaProducer -> (KafkaError -> m ()) -> Sink i m ()
+flushThenCommitSink' consumer producer handleError = L.mapM_ $ \_ -> do
+  flushProducer producer
+  res <- commitAllOffsets OffsetCommit consumer
+  case res of
+    Nothing  -> pure ()
+    Just err -> handleError err
diff --git a/test/Kafka/Conduit/CombinatorSpec.hs b/test/Kafka/Conduit/CombinatorSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/Kafka/Conduit/CombinatorSpec.hs
@@ -0,0 +1,43 @@
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module Kafka.Conduit.CombinatorSpec (spec) where
+
+import           Data.Conduit
+import qualified Data.Conduit.List         as CL
+import           Data.List.Extra
+import           Kafka.Conduit.Combinators
+import           Prelude                   as P
+import           Test.Hspec
+import           Test.QuickCheck
+
+{-# ANN module ("HLint: ignore Redundant do"        :: String) #-}
+{-# ANN module ("HLint: ignore Reduce duplication"  :: String) #-}
+
+spec :: Spec
+spec = describe "Kafka.Conduit.UtilSpec" $ do
+  describe "batchByOrFlush" $ do
+    it "Should batch by i properly when Nothing is interspersed between every j elements" $ do
+      forAll (choose (1 :: Int, 100)) $ \i ->
+        forAll (choose (1 :: Int, 100)) $ \j ->
+          testBatchBy batchByOrFlush Just [Nothing] i j
+    it "Should batch by i properly when [Nothing, Nothing] is interspersed between every j elements" $ do
+      forAll (choose (1 :: Int, 100)) $ \i ->
+        forAll (choose (1 :: Int, 100)) $ \j ->
+          testBatchBy batchByOrFlush Just [Nothing, Nothing] i j
+  describe "batchByOrFlushEither" $ do
+    it "Should batch by i properly when Left () is interspersed between every j elements" $ do
+      forAll (choose (1 :: Int, 100)) $ \i ->
+        forAll (choose (1 :: Int, 100)) $ \j ->
+          testBatchBy batchByOrFlushEither Right [Left ()] i j
+    it "Should batch by i properly when [Left (), Left ()] is interspersed between every j elements" $ do
+      forAll (choose (1 :: Int, 100)) $ \i ->
+        forAll (choose (1 :: Int, 100)) $ \j ->
+          testBatchBy batchByOrFlushEither Right [Left (), Left ()] i j
+
+testBatchBy :: (Enum a, Num a) => (BatchSize -> ConduitM b [Int] IO ()) -> (a -> b) -> [b] -> Int -> Int -> IO ()
+testBatchBy batchFn constr inter i j = do
+  let as = intercalate inter (chunksOf j (constr <$> [1..100]))
+  xs :: [[Int]] <- runConduit $ CL.sourceList as .| batchFn (BatchSize i) .| CL.consume
+  P.concat xs `shouldBe` [1..100]
+  P.filter (> i) (length <$> P.init xs) `shouldBe` []
+  P.filter (> j) (length <$> P.init xs) `shouldBe` []