diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,8 @@
-## [_Unreleased_](https://github.com/freckle/freckle-app/compare/freckle-kafka-v0.0.0.1...main)
+## [_Unreleased_](freckle-kafka-v0.0.0.2...main)
+
+## [v1.21.1.0](https://github.com/freckle/freckle-app/compare/freckle-kafka-v0.0.0.1...freckle-kafka-v0.0.0.2)
+
+Add `Freckle.App.Kafka.Consumer.runConsumerBatched`
 
 ## [v0.0.0.1](https://github.com/freckle/freckle-app/compare/freckle-kafka-v0.0.0.0...freckle-kafka-v0.0.0.1)
 
diff --git a/README.md b/README.md
--- a/README.md
+++ b/README.md
@@ -1,6 +1,6 @@
 # freckle-kafka
 
-Some extensions to the `hw-kafka-client` library.
+Some extensions to the [hw-kafka-client](https://hackage.haskell.org/package/hw-kafka-client) library.
 
 ---
 
diff --git a/freckle-kafka.cabal b/freckle-kafka.cabal
--- a/freckle-kafka.cabal
+++ b/freckle-kafka.cabal
@@ -1,6 +1,6 @@
 cabal-version:      1.18
 name:               freckle-kafka
-version:            0.0.0.1
+version:            0.0.0.2
 license:            MIT
 license-file:       LICENSE
 maintainer:         Freckle Education
@@ -42,7 +42,7 @@
         -Wno-safe -Wno-unsafe
 
     build-depends:
-        Blammo >=2.0.0.0,
+        Blammo >=2.1.0.0,
         aeson >=2.0.3.0,
         annotated-exception >=0.2.0.4,
         base >=4.16.4.0 && <5,
diff --git a/library/Freckle/App/Kafka/Consumer.hs b/library/Freckle/App/Kafka/Consumer.hs
--- a/library/Freckle/App/Kafka/Consumer.hs
+++ b/library/Freckle/App/Kafka/Consumer.hs
@@ -6,6 +6,7 @@
   , KafkaConsumerConfig (..)
   , envKafkaConsumerConfig
   , runConsumer
+  , runConsumerBatched
   ) where
 
 import Prelude
@@ -18,17 +19,19 @@
   , displayException
   )
 import Control.Exception.Annotated.UnliftIO qualified as Annotated
-import Control.Lens (Lens', view)
+import Control.Lens (Lens', over, view, _Left)
 import Control.Monad (forever, (<=<))
 import Control.Monad.IO.Class (MonadIO)
 import Control.Monad.Reader (MonadReader)
 import Data.Aeson
 import Data.ByteString (ByteString)
-import Data.Foldable (for_)
-import Data.List.NonEmpty (NonEmpty)
+import Data.Either (partitionEithers)
+import Data.Foldable (for_, traverse_)
+import Data.List.NonEmpty (NonEmpty, nonEmpty)
 import Data.List.NonEmpty qualified as NE
 import Data.Map.Strict (Map)
 import Data.Map.Strict qualified as Map
+import Data.Maybe (mapMaybe)
 import Data.Text (Text)
 import Data.Text qualified as T
 import Data.Text.Encoding qualified as T
@@ -198,7 +201,7 @@
         "kafka.consumer"
         (defaultSpanArguments {Trace.kind = Consumer})
       $ do
-        mRecord <- fromKafkaError =<< pollMessage consumer kTimeout
+        mRecord <- fromKafkaError =<< pollMessage consumer (kafkaTimeout pollTimeout)
 
         for_ (crValue =<< mRecord) $ \bs -> do
           a <-
@@ -207,8 +210,6 @@
                 eitherDecodeStrict bs
           inSpan "kafka.consumer.message.handle" defaultSpanArguments $ onMessage a
  where
-  kTimeout = Kafka.Timeout $ timeoutMs pollTimeout
-
   handlers =
     [ Annotated.Handler $
         logErrorNS "kafka"
@@ -219,6 +220,63 @@
           . annotatedExceptionMessageFrom @KafkaMessageDecodeError
             (const "Could not decode message value")
     ]
+
+data BatchConsumerError
+  = BatchConsumerDecodeError KafkaMessageDecodeError
+  | BatchConsumerKafkaError KafkaError
+  deriving stock (Show)
+
+runConsumerBatched
+  :: forall a m env
+   . ( MonadUnliftIO m
+     , MonadReader env m
+     , MonadLogger m
+     , MonadTracer m
+     , HasKafkaConsumer env
+     , FromJSON a
+     , HasCallStack
+     )
+  => Timeout
+  -> BatchSize
+  -> ([a] -> m ())
+  -> m ()
+runConsumerBatched pollTimeout batchSize onBatch =
+  forever $ do
+    consumer <- view kafkaConsumerL
+
+    errors <- inSpan
+      "kafka.batchConsumer"
+      (defaultSpanArguments {Trace.kind = Consumer})
+      $ do
+        (errors, records) <- do
+          (kafkaErrors, batch) <-
+            partitionEithers
+              <$> pollMessageBatch consumer (kafkaTimeout pollTimeout) batchSize
+          (decodeErrors, records) <- fmap partitionEithers $ inSpan "kafka.batchConsumer.messages.decode" defaultSpanArguments $ do
+            let errorDecode bs = over _Left (KafkaMessageDecodeError bs) $ eitherDecodeStrict @a bs
+            pure $ mapMaybe (fmap errorDecode . crValue) batch
+          pure
+            ( fmap BatchConsumerKafkaError kafkaErrors
+                <> fmap BatchConsumerDecodeError decodeErrors
+            , records
+            )
+
+        inSpan "kafka.batchConsumer.messages.handle" defaultSpanArguments $
+          onBatch records
+        pure $ nonEmpty errors
+
+    traverse_ logErrors errors
+ where
+  displayConsumerError = \case
+    BatchConsumerDecodeError e -> displayException e
+    BatchConsumerKafkaError e -> displayException e
+  logErrors errors =
+    logErrorNS "kafka" $
+      "Batch consumer errors"
+        :# ["errors" .= fmap displayConsumerError errors]
+
+kafkaTimeout :: Timeout -> Kafka.Timeout
+kafkaTimeout = Kafka.Timeout . timeoutMs
 
 -- | Like 'annotatedExceptionMessage', but use the supplied function to
 --   construct an initial 'Message' that it will augment.
diff --git a/package.yaml b/package.yaml
--- a/package.yaml
+++ b/package.yaml
@@ -1,5 +1,5 @@
 name: freckle-kafka
-version: 0.0.0.1
+version: 0.0.0.2
 maintainer: Freckle Education
 category: Database
 github: freckle/freckle-app
@@ -54,7 +54,7 @@
 library:
   source-dirs: library
   dependencies:
-    - Blammo >= 2.0.0.0
+    - Blammo
     - aeson
     - annotated-exception
     - bytestring