diff --git a/example/ProducerExample.hs b/example/ProducerExample.hs
--- a/example/ProducerExample.hs
+++ b/example/ProducerExample.hs
@@ -13,7 +13,7 @@
 producerProps :: ProducerProperties
 producerProps = brokersList [BrokerAddress "localhost:9092"]
              <> sendTimeout (Timeout 10000)
-             <> setCallback (deliveryErrorsCallback print)
+             <> setCallback (deliveryCallback print)
              <> logLevel KafkaLogDebug
 
 -- Topic to send messages to
diff --git a/hw-kafka-client.cabal b/hw-kafka-client.cabal
--- a/hw-kafka-client.cabal
+++ b/hw-kafka-client.cabal
@@ -2,10 +2,10 @@
 --
 -- see: https://github.com/sol/hpack
 --
--- hash: 8d76dba4ea94855b431d568543c5cd5d2f86a3bab474cd1151a9f936673891fc
+-- hash: ae89ef8878b399834f3550efb6ad9b04c27fad1d06d9d22fecfc872a4f0d23c4
 
 name:           hw-kafka-client
-version:        2.4.3
+version:        2.4.4
 synopsis:       Kafka bindings for Haskell
 description:    Apache Kafka bindings backed by the librdkafka C library.
                 .
@@ -49,7 +49,6 @@
     , bifunctors
     , bytestring
     , containers
-    , temporary
     , transformers
     , unix
   build-tools:
@@ -94,7 +93,6 @@
     , bytestring
     , containers
     , hw-kafka-client
-    , temporary
     , transformers
     , unix
   if !(flag(examples))
diff --git a/src/Kafka/Consumer.hs b/src/Kafka/Consumer.hs
--- a/src/Kafka/Consumer.hs
+++ b/src/Kafka/Consumer.hs
@@ -8,7 +8,7 @@
 , committed, position, seek
 , pollMessage, pollConsumerEvents
 , commitOffsetMessage, commitAllOffsets, commitPartitionsOffsets
-, storeOffsetMessage
+, storeOffsets, storeOffsetMessage
 , closeConsumer
 -- ReExport Types
 , KafkaConsumer
@@ -107,12 +107,20 @@
 
 -- | Stores message's offset locally for the message's partition.
 storeOffsetMessage :: MonadIO m
-                    => KafkaConsumer
-                    -> ConsumerRecord k v
-                    -> m (Maybe KafkaError)
+                   => KafkaConsumer
+                   -> ConsumerRecord k v
+                   -> m (Maybe KafkaError)
 storeOffsetMessage k m =
   liftIO $ toNativeTopicPartitionList [topicPartitionFromMessageForCommit m] >>= commitOffsetsStore k
 
+-- | Stores offsets locally
+storeOffsets :: MonadIO m
+             => KafkaConsumer
+             -> [TopicPartition]
+             -> m (Maybe KafkaError)
+storeOffsets k ps =
+  liftIO $ toNativeTopicPartitionList ps >>= commitOffsetsStore k
+
 -- | Commit offsets for all currently assigned partitions.
 commitAllOffsets :: MonadIO m
                  => OffsetCommit
@@ -171,12 +179,11 @@
   where
     seekAll = runExceptT $ do
       tr <- traverse (ExceptT . topicPair) tps
-      _  <- traverse (\(kt, p, o) -> ExceptT (rdSeek kt p o)) tr
+      void $ traverse (\(kt, p, o) -> ExceptT (rdSeek kt p o)) tr
       return ()
 
-    rdSeek kt (PartitionId p) o = do
-      res <- rdKafkaSeek kt (fromIntegral p) (offsetToInt64 o) timeout
-      return $ rdKafkaErrorToEither res
+    rdSeek kt (PartitionId p) o =
+      rdKafkaErrorToEither <$> rdKafkaSeek kt (fromIntegral p) (offsetToInt64 o) timeout
 
     topicPair tp = do
       let (TopicName tn) = tpTopicName tp
@@ -228,7 +235,7 @@
   CToken.cancel ct
   mbq <- readIORef qr
   void $ traverse rdKafkaQueueDestroy mbq
-  (kafkaErrorToMaybe . KafkaResponseError) <$> rdKafkaConsumerClose k
+  kafkaErrorToMaybe . KafkaResponseError <$> rdKafkaConsumerClose k
 
 -----------------------------------------------------------------------------
 newConsumerConf :: ConsumerProperties -> IO KafkaConf
@@ -256,11 +263,11 @@
 
 commitOffsets :: OffsetCommit -> KafkaConsumer -> RdKafkaTopicPartitionListTPtr -> IO (Maybe KafkaError)
 commitOffsets o (KafkaConsumer (Kafka k) _) pl =
-    (kafkaErrorToMaybe . KafkaResponseError) <$> rdKafkaCommit k pl (offsetCommitToBool o)
+    kafkaErrorToMaybe . KafkaResponseError <$> rdKafkaCommit k pl (offsetCommitToBool o)
 
 commitOffsetsStore :: KafkaConsumer -> RdKafkaTopicPartitionListTPtr -> IO (Maybe KafkaError)
 commitOffsetsStore (KafkaConsumer (Kafka k) _) pl =
-    (kafkaErrorToMaybe . KafkaResponseError) <$> rdKafkaOffsetsStore k pl
+    kafkaErrorToMaybe . KafkaResponseError <$> rdKafkaOffsetsStore k pl
 
 setConsumerLogLevel :: KafkaConsumer -> KafkaLogLevel -> IO ()
 setConsumerLogLevel (KafkaConsumer (Kafka k) _) level =
@@ -268,7 +275,7 @@
 
 redirectCallbacksPoll :: KafkaConsumer -> IO (Maybe KafkaError)
 redirectCallbacksPoll (KafkaConsumer (Kafka k) _) =
-  (kafkaErrorToMaybe . KafkaResponseError) <$> rdKafkaPollSetConsumer k
+  kafkaErrorToMaybe . KafkaResponseError <$> rdKafkaPollSetConsumer k
 
 runConsumerLoop :: KafkaConsumer -> CancellationToken -> Maybe Timeout -> IO ()
 runConsumerLoop k ct timeout =
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
@@ -9,6 +9,7 @@
 import qualified Data.List            as L
 import           Data.Map             (Map)
 import qualified Data.Map             as M
+import           Data.Semigroup       as Sem
 import           Kafka.Consumer.Types
 import           Kafka.Internal.Setup
 import           Kafka.Types
@@ -22,6 +23,11 @@
   , cpCallbacks :: [KafkaConf -> IO ()]
   }
 
+instance Sem.Semigroup ConsumerProperties where
+  (ConsumerProperties m1 ll1 cb1) <> (ConsumerProperties m2 ll2 cb2) =
+    ConsumerProperties (M.union m2 m1) (ll2 `mplus` ll1) (cb1 `mplus` cb2)
+  {-# INLINE (<>) #-}
+
 -- | /Right biased/ so we prefer newer properties over older ones.
 instance Monoid ConsumerProperties where
   mempty = ConsumerProperties
@@ -30,8 +36,7 @@
     , cpCallbacks      = []
     }
   {-# INLINE mempty #-}
-  mappend (ConsumerProperties m1 ll1 cb1) (ConsumerProperties m2 ll2 cb2) =
-    ConsumerProperties (M.union m2 m1) (ll2 `mplus` ll1) (cb1 `mplus` cb2)
+  mappend = (Sem.<>)
   {-# INLINE mappend #-}
 
 brokersList :: [BrokerAddress] -> ConsumerProperties
diff --git a/src/Kafka/Consumer/Convert.hs b/src/Kafka/Consumer/Convert.hs
--- a/src/Kafka/Consumer/Convert.hs
+++ b/src/Kafka/Consumer/Convert.hs
@@ -132,24 +132,6 @@
                 , crValue     = payload
                 }
 
-        readTopic msg = newForeignPtr_ (topic'RdKafkaMessageT msg) >>= rdKafkaTopicName
-        readPayload = readBS len'RdKafkaMessageT payload'RdKafkaMessageT
-        readKey = readBS keyLen'RdKafkaMessageT key'RdKafkaMessageT
-        readTimestamp msg =
-            alloca $ \p -> do
-                typeP <- newForeignPtr_ p
-                ts <- fromIntegral <$> rdKafkaMessageTimestamp msg typeP
-                tsType <- peek p
-                return $ case tsType of
-                    RdKafkaTimestampCreateTime    -> CreateTime (Millis ts)
-                    RdKafkaTimestampLogAppendTime -> LogAppendTime (Millis ts)
-                    RdKafkaTimestampNotAvailable  -> NoTimestamp
-
-        readBS flen fdata s = if fdata s == nullPtr
-                        then return Nothing
-                        else Just <$> word8PtrToBS (flen s) (fdata s)
-
-
 offsetCommitToBool :: OffsetCommit -> Bool
 offsetCommitToBool OffsetCommit      = False
 offsetCommitToBool OffsetCommitAsync = True
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
@@ -1,21 +1,27 @@
 module Kafka.Consumer.Subscription
 where
 
---
-import Data.Map (Map)
-import qualified Data.Map as M
-import qualified Data.List as L
-import Kafka.Types
-import Kafka.Consumer.Types
+import qualified Data.List            as L
+import           Data.Map             (Map)
+import qualified Data.Map             as M
+import           Data.Semigroup       as Sem
+import           Kafka.Consumer.Types
+import           Kafka.Types
 
 data Subscription = Subscription [TopicName] (Map String String)
 
-instance Monoid Subscription where
-  mempty = Subscription [] M.empty
-  mappend (Subscription ts1 m1) (Subscription ts2 m2) =
+instance Sem.Semigroup Subscription where
+  (Subscription ts1 m1) <> (Subscription ts2 m2) =
     let ts' = L.nub $ L.union ts1 ts2
         ps' = M.union m1 m2
      in Subscription ts' ps'
+  {-# INLINE (<>) #-}
+
+instance Monoid Subscription where
+  mempty = Subscription [] M.empty
+  {-# INLINE mempty #-}
+  mappend = (Sem.<>)
+  {-# INLINE mappend #-}
 
 topics :: [TopicName] -> Subscription
 topics ts = Subscription (L.nub ts) M.empty
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
@@ -26,7 +26,7 @@
 
 {#enum rd_kafka_type_t as ^ {underscoreToCase} deriving (Show, Eq) #}
 {#enum rd_kafka_conf_res_t as ^ {underscoreToCase} deriving (Show, Eq) #}
-{#enum rd_kafka_resp_err_t as ^ {underscoreToCase} deriving (Show, Eq) #}
+{#enum rd_kafka_resp_err_t as ^ {underscoreToCase} deriving (Show, Eq, Bounded) #}
 {#enum rd_kafka_timestamp_type_t as ^ {underscoreToCase} deriving (Show, Eq) #}
 
 type RdKafkaMsgFlag = Int
@@ -47,6 +47,9 @@
     {} -> `String' #}
 
 {#fun pure rd_kafka_err2str as ^
+    {enumToCInt `RdKafkaRespErrT'} -> `String' #}
+
+{#fun pure rd_kafka_err2name as ^
     {enumToCInt `RdKafkaRespErrT'} -> `String' #}
 
 {#fun pure rd_kafka_errno2err as ^
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
@@ -6,7 +6,9 @@
 import           Control.Monad                    (void, when)
 import qualified Data.ByteString                  as BS
 import qualified Data.ByteString.Internal         as BSI
+import           Foreign                          hiding (void)
 import           Foreign.C.Error
+import           Kafka.Consumer.Types
 import           Kafka.Internal.CancellationToken as CToken
 import           Kafka.Internal.RdKafka
 import           Kafka.Internal.Setup
@@ -70,3 +72,28 @@
 maybeToLeft :: Maybe a -> Either a ()
 maybeToLeft = maybe (Right ()) Left
 {-# INLINE maybeToLeft #-}
+
+readPayload :: RdKafkaMessageT -> IO (Maybe BS.ByteString)
+readPayload = readBS len'RdKafkaMessageT payload'RdKafkaMessageT
+
+readTopic :: RdKafkaMessageT -> IO String
+readTopic msg = newForeignPtr_ (topic'RdKafkaMessageT msg) >>= rdKafkaTopicName
+
+readKey :: RdKafkaMessageT -> IO (Maybe BSI.ByteString)
+readKey = readBS keyLen'RdKafkaMessageT key'RdKafkaMessageT
+
+readTimestamp :: RdKafkaMessageTPtr -> IO Timestamp
+readTimestamp msg =
+  alloca $ \p -> do
+    typeP <- newForeignPtr_ p
+    ts <- fromIntegral <$> rdKafkaMessageTimestamp msg typeP
+    tsType <- peek p
+    return $ case tsType of
+               RdKafkaTimestampCreateTime    -> CreateTime (Millis ts)
+               RdKafkaTimestampLogAppendTime -> LogAppendTime (Millis ts)
+               RdKafkaTimestampNotAvailable  -> NoTimestamp
+
+readBS :: (t -> Int) -> (t -> Ptr Word8) -> t -> IO (Maybe BS.ByteString)
+readBS flen fdata s = if fdata s == nullPtr
+                        then return Nothing
+                        else Just <$> word8PtrToBS (flen s) (fdata s)
diff --git a/src/Kafka/Producer/Callbacks.hs b/src/Kafka/Producer/Callbacks.hs
--- a/src/Kafka/Producer/Callbacks.hs
+++ b/src/Kafka/Producer/Callbacks.hs
@@ -1,28 +1,51 @@
 module Kafka.Producer.Callbacks
-( deliveryErrorsCallback
+( deliveryCallback
 , module X
 )
 where
 
-import Foreign
-import Foreign.C.Error
-import Kafka.Callbacks        as X
-import Kafka.Internal.RdKafka
-import Kafka.Internal.Setup
-import Kafka.Internal.Shared
-import Kafka.Types
+import           Foreign
+import           Foreign.C.Error
+import           Kafka.Callbacks        as X
+import           Kafka.Consumer.Types
+import           Kafka.Internal.RdKafka
+import           Kafka.Internal.Setup
+import           Kafka.Internal.Shared
+import           Kafka.Producer.Types
+import           Kafka.Types
 
--- | Sets the callback for delivery errors.
--- The callback is only called in case of errors.
-deliveryErrorsCallback :: (KafkaError -> IO ()) -> KafkaConf -> IO ()
-deliveryErrorsCallback callback kc = rdKafkaConfSetDrMsgCb (getRdKafkaConf kc) realCb
+-- | Sets the callback for delivery reports.
+deliveryCallback :: (DeliveryReport -> IO ()) -> KafkaConf -> IO ()
+deliveryCallback callback kc = rdKafkaConfSetDrMsgCb (getRdKafkaConf kc) realCb
   where
     realCb :: t -> Ptr RdKafkaMessageT -> IO ()
     realCb _ mptr =
       if mptr == nullPtr
-        then getErrno >>= (callback . kafkaRespErr)
+        then getErrno >>= (callback . NoMessageError . kafkaRespErr)
         else do
           s <- peek mptr
           if err'RdKafkaMessageT s /= RdKafkaRespErrNoError
-            then (callback . KafkaResponseError $ err'RdKafkaMessageT s)
-            else pure ()
+            then mkErrorReport s   >>= callback
+            else mkSuccessReport s >>= callback
+
+mkErrorReport :: RdKafkaMessageT -> IO DeliveryReport
+mkErrorReport msg = do
+  prodRec <- mkProdRec msg
+  pure $ DeliveryFailure prodRec (KafkaResponseError (err'RdKafkaMessageT msg))
+
+mkSuccessReport :: RdKafkaMessageT -> IO DeliveryReport
+mkSuccessReport msg = do
+  prodRec <- mkProdRec msg
+  pure $ DeliverySuccess prodRec (Offset $ offset'RdKafkaMessageT msg)
+
+mkProdRec :: RdKafkaMessageT -> IO ProducerRecord
+mkProdRec msg = do
+  topic     <- readTopic msg
+  key       <- readKey msg
+  payload   <- readPayload msg
+  pure ProducerRecord
+    { prTopic = TopicName topic
+    , prPartition = SpecifiedPartition (partition'RdKafkaMessageT msg)
+    , prKey = key
+    , prValue = payload
+    }
diff --git a/src/Kafka/Producer/ProducerProperties.hs b/src/Kafka/Producer/ProducerProperties.hs
--- a/src/Kafka/Producer/ProducerProperties.hs
+++ b/src/Kafka/Producer/ProducerProperties.hs
@@ -8,6 +8,7 @@
 import qualified Data.List                as L
 import           Data.Map                 (Map)
 import qualified Data.Map                 as M
+import           Data.Semigroup           as Sem
 import           Kafka.Internal.Setup
 import           Kafka.Producer.Callbacks
 import           Kafka.Types
@@ -20,6 +21,11 @@
   , ppCallbacks  :: [KafkaConf -> IO ()]
   }
 
+instance Sem.Semigroup ProducerProperties where
+  (ProducerProperties k1 t1 ll1 cb1) <> (ProducerProperties k2 t2 ll2 cb2) =
+    ProducerProperties (M.union k2 k1) (M.union t2 t1) (ll2 `mplus` ll1) (cb1 `mplus` cb2)
+  {-# INLINE (<>) #-}
+
 -- | /Right biased/ so we prefer newer properties over older ones.
 instance Monoid ProducerProperties where
   mempty = ProducerProperties
@@ -29,8 +35,7 @@
     , ppCallbacks      = []
     }
   {-# INLINE mempty #-}
-  mappend (ProducerProperties k1 t1 ll1 cb1) (ProducerProperties k2 t2 ll2 cb2) =
-    ProducerProperties (M.union k2 k1) (M.union t2 t1) (ll2 `mplus` ll1) (cb1 `mplus` cb2)
+  mappend = (Sem.<>)
   {-# INLINE mappend #-}
 
 brokersList :: [BrokerAddress] -> ProducerProperties
diff --git a/src/Kafka/Producer/Types.hs b/src/Kafka/Producer/Types.hs
--- a/src/Kafka/Producer/Types.hs
+++ b/src/Kafka/Producer/Types.hs
@@ -7,6 +7,7 @@
 import           Data.Typeable
 import           Kafka.Internal.Setup
 import           Kafka.Types
+import Kafka.Consumer.Types
 
 -- | Main pointer to Kafka object, which contains our brokers
 data KafkaProducer = KafkaProducer
@@ -39,3 +40,25 @@
     SpecifiedPartition {-# UNPACK #-} !Int  -- the partition number of the topic
   | UnassignedPartition
   deriving (Show, Eq, Ord, Typeable)
+
+data DeliveryReport
+  = DeliverySuccess ProducerRecord Offset
+  | DeliveryFailure ProducerRecord KafkaError
+  | NoMessageError KafkaError
+  deriving (Show, Eq)
+
+-- -- | Represents the report of a successfully delivered message.
+-- data ProducerSuccess = ProducerSuccess
+--   { psTopic     :: !TopicName    -- ^ Kafka topic this message was received from
+--   , psPartition :: !PartitionId  -- ^ Kafka partition this message was received from
+--   , psOffset    :: !Offset       -- ^ Offset within the 'crPartition' Kafka partition
+--   , psKey       :: !(Maybe BS.ByteString)
+--   , psValue     :: !(Maybe BS.ByteString)
+--   }
+--   deriving (Eq, Show, Read, Typeable)
+
+-- -- | Represents the failure to deliver a message.
+-- data ProducerError = ProducerError
+--   { peValue :: !(Maybe BS.ByteString)  -- ^ The message that failed.
+--   , peError :: KafkaError              -- ^ The reason for the failure.
+--   } deriving (Eq, Show)
diff --git a/src/Kafka/Types.hs b/src/Kafka/Types.hs
--- a/src/Kafka/Types.hs
+++ b/src/Kafka/Types.hs
@@ -68,7 +68,10 @@
   | KafkaBadConfiguration
     deriving (Eq, Show, Typeable)
 
-instance Exception KafkaError
+instance Exception KafkaError where
+  displayException (KafkaResponseError err) =
+    "[" ++ rdKafkaErr2name err ++ "] " ++ rdKafkaErr2str err
+  displayException err = show err
 
 data KafkaDebug =
     DebugGeneric