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:             1.1.1
+version:             1.1.2
 synopsis:            Conduit bindings for kafka-client
 homepage:            https://github.com/haskell-works/hw-kafka-client-conduit
 bug-reports:         https://github.com/haskell-works/hw-kafka-client-conduit/issues
@@ -19,53 +19,54 @@
   location: https://github.com/haskell-works/hw-kafka-conduit
 
 executable kafka-conduit-example
-  main-is:             Main.hs
-  hs-source-dirs:      example
-  default-language:    Haskell2010
-  build-depends:       base >=4.6 && < 5
-                     , bifunctors
-                     , bytestring
-                     , conduit
-                     , containers
-                     , resourcet
-                     , hw-kafka-conduit
+  main-is:              Main.hs
+  hs-source-dirs:       example
+  default-language:     Haskell2010
+  build-depends:        base >=4.6 && < 5
+                      , bifunctors
+                      , bytestring
+                      , conduit
+                      , containers
+                      , resourcet
+                      , hw-kafka-conduit
 
 library
-  hs-source-dirs:      src
-  ghc-options:         -Wall
-  exposed-modules:     Kafka.Conduit
-                     , Kafka.Conduit.Source
-                     , Kafka.Conduit.Sink
-                     , Kafka.Conduit.Utils
-  default-language:    Haskell2010
-  build-depends:       base >= 4.7 && < 5
-                     , bifunctors
-                     , bytestring
-                     , containers
-                     , conduit
-                     , conduit-extra
-                     , exceptions
-                     , hw-kafka-client
-                     , mtl
-                     , resourcet
-                     , transformers
-
+  hs-source-dirs:       src
+  ghc-options:          -Wall
+  exposed-modules:      Kafka.Conduit
+                      , Kafka.Conduit.Combinators
+                      , Kafka.Conduit.Source
+                      , Kafka.Conduit.Sink
+  default-language:     Haskell2010
+  build-depends:        base >= 4.7 && < 5
+                      , bifunctors
+                      , bytestring
+                      , containers
+                      , conduit
+                      , conduit-extra
+                      , exceptions
+                      , hw-kafka-client
+                      , mtl
+                      , resourcet
+                      , transformers
 
 test-suite kafka-client-conduit-test
-  type:                exitcode-stdio-1.0
-  hs-source-dirs:      test
-  main-is:             Spec.hs
-  ghc-options:         -Wall -threaded -rtsopts -with-rtsopts=-N
-  default-language:    Haskell2010
-  build-depends:       base
-                     , hw-kafka-conduit
-                     , bifunctors
-                     , bytestring
-                     , containers
-                     , conduit
-                     , conduit-extra
-                     , hspec
-                     , hw-kafka-client
-                     , mtl
-                     , resourcet
-                     , transformers
+  type:                 exitcode-stdio-1.0
+  hs-source-dirs:       test
+  main-is:              Spec.hs
+  ghc-options:          -Wall -threaded -rtsopts -with-rtsopts=-N
+  default-language:     Haskell2010
+  build-depends:        base
+                      , hw-kafka-conduit
+                      , bifunctors
+                      , bytestring
+                      , containers
+                      , conduit
+                      , conduit-extra
+                      , extra
+                      , hspec
+                      , hw-kafka-client
+                      , mtl
+                      , resourcet
+                      , transformers
+                      , QuickCheck
diff --git a/src/Kafka/Conduit/Combinators.hs b/src/Kafka/Conduit/Combinators.hs
new file mode 100644
--- /dev/null
+++ b/src/Kafka/Conduit/Combinators.hs
@@ -0,0 +1,43 @@
+module Kafka.Conduit.Combinators
+  ( batchByOrFlush
+  , foldYield
+  , throwLeft
+  , throwLeftSatisfy
+  ) where
+
+import Control.Exception
+import Control.Monad
+import Control.Monad.Catch
+import Data.Conduit
+
+-- | Throws the left part of a value in a 'MonadThrow' context
+throwLeft :: (MonadThrow m, Exception e) => Conduit (Either e i) m i
+throwLeft = awaitForever (either throwM yield)
+
+-- | Throws the left part of a value in a 'MonadThrow' context if the value
+-- satisfies the predicate
+throwLeftSatisfy :: (MonadThrow m, Exception e) => (e -> Bool) -> Conduit (Either e i) m (Either e i)
+throwLeftSatisfy p = awaitForever awaitHandle
+  where awaitHandle (Left e) | p e  = throwM e
+        awaitHandle v               = yield v
+
+-- | Create a conduit that folds with the function f over its input i with its
+-- internal state s and emits outputs [o], then finally emits outputs [o] from
+-- the function g applied to the final state s.
+foldYield :: Monad m => (i -> s -> (s, [o])) -> (s -> [o]) -> s -> Conduit i m o
+foldYield f g s = do
+  mi <- await
+  case mi of
+    Just i -> do
+      let (s', os) = f i s
+      forM_ os yield
+      foldYield f g s'
+    Nothing -> forM_ (g s) yield
+
+batchByOrFlush :: Monad m => Int -> Conduit (Maybe a) m [a]
+batchByOrFlush n = foldYield folder finish (0 :: Int, [])
+  where
+    folder Nothing  (_, xs)                 = ((0    ,   []), [reverse    xs ])
+    folder (Just a) (i, xs) | (i + 1) >= n  = ((0    ,   []), [reverse (a:xs)])
+    folder (Just a) (i, xs)                 = ((i + 1, a:xs),               [])
+    finish (_, xs) = [reverse xs]
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,7 +8,7 @@
 import Control.Monad.Trans.Resource
 import Data.Conduit
 import Kafka.Producer as X
-import Kafka.Conduit.Utils as X
+import Kafka.Conduit.Combinators as X
 
 -- | Creates a Sink for a given `KafkaProducer`.
 -- The producer will be closed when the Sink is closed.
diff --git a/src/Kafka/Conduit/Source.hs b/src/Kafka/Conduit/Source.hs
--- a/src/Kafka/Conduit/Source.hs
+++ b/src/Kafka/Conduit/Source.hs
@@ -23,7 +23,7 @@
 import Data.Conduit
 import qualified Data.Conduit.List as L
 import Kafka.Consumer as X
-import Kafka.Conduit.Utils as X
+import Kafka.Conduit.Combinators as X
 
 -- | Create a `Source` for a given `KafkaConsumer`.
 -- The consumer will NOT be closed automatically when the `Source` is closed.
diff --git a/src/Kafka/Conduit/Utils.hs b/src/Kafka/Conduit/Utils.hs
deleted file mode 100644
--- a/src/Kafka/Conduit/Utils.hs
+++ /dev/null
@@ -1,23 +0,0 @@
-module Kafka.Conduit.Utils where
-
-import Control.Exception
-import Control.Monad
-import Control.Monad.Catch
-import Data.Conduit
-
--- | Throws the left part of a value in a 'MonadThrow' context
-throwLeft :: (MonadThrow m, Exception e) => Conduit (Either e i) m i
-throwLeft = awaitForever (either throwM yield)
-
--- | Create a conduit that folds with the function f over its input i with its
--- internal state s and emits outputs [o], then finally emits outputs [o] from
--- the function g applied to the final state s.
-foldYield :: Monad m => (i -> s -> (s, [o])) -> (s -> [o]) -> s -> Conduit i m o
-foldYield f g s = do
-  mi <- await
-  case mi of
-    Just i -> do
-      let (s', os) = f i s
-      forM_ os yield
-      foldYield f g s'
-    Nothing -> forM_ (g s) yield
diff --git a/test/Spec.hs b/test/Spec.hs
--- a/test/Spec.hs
+++ b/test/Spec.hs
@@ -1,2 +1,1 @@
-main :: IO ()
-main = putStrLn "Test suite not yet implemented"
+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}