diff --git a/hw-kafka-client.cabal b/hw-kafka-client.cabal
--- a/hw-kafka-client.cabal
+++ b/hw-kafka-client.cabal
@@ -1,7 +1,7 @@
 cabal-version: 2.2
 
 name:                   hw-kafka-client
-version:                4.0.2
+version:                4.0.3
 synopsis:               Kafka bindings for Haskell
 description:            Apache Kafka bindings backed by the librdkafka C library.
                         .
diff --git a/src/Kafka/Callbacks.hs b/src/Kafka/Callbacks.hs
--- a/src/Kafka/Callbacks.hs
+++ b/src/Kafka/Callbacks.hs
@@ -8,7 +8,7 @@
 
 import Data.ByteString (ByteString)
 import Kafka.Internal.RdKafka (rdKafkaConfSetErrorCb, rdKafkaConfSetLogCb, rdKafkaConfSetStatsCb)
-import Kafka.Internal.Setup (HasKafkaConf(..), getRdKafkaConf, Callback(..))
+import Kafka.Internal.Setup (getRdKafkaConf, Callback(..))
 import Kafka.Types (KafkaError(..), KafkaLogLevel(..))
 
 -- | Add a callback for errors.
diff --git a/src/Kafka/Consumer.hs b/src/Kafka/Consumer.hs
--- a/src/Kafka/Consumer.hs
+++ b/src/Kafka/Consumer.hs
@@ -69,7 +69,7 @@
 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              (forM_, void, when)
 import           Control.Monad.IO.Class     (MonadIO (liftIO))
 import           Control.Monad.Trans.Except (ExceptT (ExceptT), runExceptT)
 import           Data.Bifunctor             (bimap, first)
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
@@ -11,6 +11,7 @@
 import Data.Int (Int32, Int64)
 import Data.Word (Word8)
 import Foreign.Concurrent (newForeignPtr)
+import qualified Foreign.Concurrent as Concurrent
 import Foreign.Marshal.Alloc (alloca, allocaBytes)
 import Foreign.Marshal.Array (peekArray, allocaArray)
 import Foreign.Storable (Storable(..))
@@ -838,9 +839,6 @@
     {enumToCInt `RdKafkaTypeT', `RdKafkaConfTPtr', id `CCharBufPointer', cIntConv `CSize'}
     -> `RdKafkaTPtr' #}
 
-foreign import ccall safe "rdkafka.h &rd_kafka_destroy"
-    rdKafkaDestroy :: FunPtr (Ptr RdKafkaT -> IO ())
-
 newRdKafkaT :: RdKafkaTypeT -> RdKafkaConfTPtr -> IO (Either Text RdKafkaTPtr)
 newRdKafkaT kafkaType confPtr =
     allocaBytes nErrorBytes $ \charPtr -> do
@@ -849,10 +847,15 @@
         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
+                -- Issue #151
+                -- rd_kafka_destroy_flags may call back into Haskell if an
+                -- error or log callback is set, so we must use a concurrent
+                -- finalizer
+                Concurrent.addForeignPtrFinalizer ret $ do
+                  -- do not call 'rd_kafka_close_consumer' on destroying all Kafka.
+                  -- when needed, applications should do it explicitly.
+                  -- RD_KAFKA_DESTROY_F_NO_CONSUMER_CLOSE = 0x8
+                  {# call rd_kafka_destroy_flags #} realPtr 0x8
                 return $ Right ret
 
 {#fun rd_kafka_brokers_add as ^
diff --git a/src/Kafka/Producer.hs b/src/Kafka/Producer.hs
--- a/src/Kafka/Producer.hs
+++ b/src/Kafka/Producer.hs
@@ -85,7 +85,7 @@
 import           Kafka.Internal.Setup     (Kafka (..), KafkaConf (..), KafkaProps (..), TopicConf (..), TopicProps (..), kafkaConf, topicConf, Callback(..))
 import           Kafka.Internal.Shared    (pollEvents)
 import           Kafka.Producer.Convert   (copyMsgFlags, handleProduceErr', producePartitionCInt, producePartitionInt)
-import           Kafka.Producer.Types     (KafkaProducer (..), ImmediateError(..))
+import           Kafka.Producer.Types     (KafkaProducer (..))
 
 import Kafka.Producer.ProducerProperties as X
 import Kafka.Producer.Types              as X hiding (KafkaProducer)