diff --git a/haskakafka.cabal b/haskakafka.cabal
--- a/haskakafka.cabal
+++ b/haskakafka.cabal
@@ -1,9 +1,8 @@
 name:                haskakafka
-version:             0.2.0.2
+version:             0.5.0
 synopsis:            Kafka bindings for Haskell
 description:         Use Apache Kafka in Haskell through the librdkafka
-                     C library. The library is preliminary but fully
-                     functional. 
+                     C library. 
 homepage:            http://github.com/cosbynator/haskakafka
 license:             MIT
 license-file:        LICENSE
@@ -21,14 +20,16 @@
 
 library
   Build-tools: c2hs
-  build-depends:       base >=4.7 && < 5
-                     , containers
-                     , unix
+  build-depends:       base >=4.6 && < 5
                      , bytestring
+                     , containers 
+                     , temporary
+                     , unix
   exposed-modules:
     Haskakafka
     Haskakafka.Internal
     Haskakafka.InternalEnum
+    Haskakafka.Example
   hs-source-dirs:      src
   default-language:    Haskell2010
   ghc-options: -Wall 
@@ -42,7 +43,10 @@
   Main-Is: TestMain.hs
   HS-Source-Dirs: tests
   ghc-options: -Wall -threaded
-  build-depends:  base >=4.7 && < 5
-                , hspec
+  build-depends:  base >=4.6 && < 5
                 , bytestring
+                , containers
                 , haskakafka
+                , hspec
+                , regex-posix
+                , either-unwrap
diff --git a/src/Haskakafka.hs b/src/Haskakafka.hs
--- a/src/Haskakafka.hs
+++ b/src/Haskakafka.hs
@@ -3,6 +3,7 @@
 module Haskakafka 
 (  Kafka
  , KafkaConf
+ , KafkaLogLevel(..)
  , KafkaMessage(..)
  , KafkaOffset(..)
  , KafkaMetadata(..)
@@ -15,9 +16,14 @@
  , KafkaTopicConf
  , KafkaType(..)
  , KafkaError (..)
+ , rdKafkaVersionStr
+ , supportedKafkaConfProperties
  , hPrintKafkaProperties
  , hPrintKafka
  , newKafkaConf
+ , setKafkaConfValue
+ , setKafkaTopicConfValue
+ , setKafkaLogLevel
  , dumpKafkaConf
  , newKafkaTopicConf
  , newKafka
@@ -26,13 +32,20 @@
  , addBrokers
  , startConsuming
  , consumeMessage
+ , consumeMessageBatch
+ , storeOffset
  , stopConsuming
  , produceMessage
+ , produceKeyedMessage
  , produceMessageBatch
  , pollEvents
  , drainOutQueue
  , getAllMetadata
  , getTopicMetadata
+ , withKafkaProducer
+ , withKafkaConsumer
+ , dumpConfFromKafkaTopic
+ , dumpConfFromKafka
  , module Haskakafka.InternalEnum
 ) where
 
@@ -47,15 +60,43 @@
 import Haskakafka.Internal
 import Haskakafka.InternalEnum
 import System.IO
+import System.IO.Temp (withSystemTempFile)
 import qualified Data.Map.Strict as Map
 import qualified Data.ByteString as BS
 import qualified Data.ByteString.Internal as BSI
 
+data KafkaLogLevel = 
+  KafkaLogEmerg | KafkaLogAlert | KafkaLogCrit | KafkaLogErr | KafkaLogWarning |
+  KafkaLogNotice | KafkaLogInfo | KafkaLogDebug
+
+instance Enum KafkaLogLevel where
+   toEnum 0 = KafkaLogEmerg
+   toEnum 1 = KafkaLogAlert
+   toEnum 2 = KafkaLogCrit
+   toEnum 3 = KafkaLogErr
+   toEnum 4 = KafkaLogWarning
+   toEnum 5 = KafkaLogNotice
+   toEnum 6 = KafkaLogInfo
+   toEnum 7 = KafkaLogDebug
+   toEnum _ = undefined
+
+   fromEnum KafkaLogEmerg = 0
+   fromEnum KafkaLogAlert = 1
+   fromEnum KafkaLogCrit = 2
+   fromEnum KafkaLogErr = 3
+   fromEnum KafkaLogWarning = 4
+   fromEnum KafkaLogNotice = 5
+   fromEnum KafkaLogInfo = 6
+   fromEnum KafkaLogDebug = 7
+
 data KafkaError = 
     KafkaError String
   | KafkaInvalidReturnValue
   | KafkaBadSpecification String
   | KafkaResponseError RdKafkaRespErrT
+  | KafkaInvalidConfigurationValue String
+  | KafkaUnknownConfigurationKey String
+  | KakfaBadConfiguration
     deriving (Eq, Show, Typeable)
 
 instance Exception KafkaError
@@ -86,20 +127,28 @@
                  | KafkaOffsetStored
                  | KafkaOffset Int64
 
+data KafkaConf = KafkaConf RdKafkaConfTPtr
+data KafkaTopicConf = KafkaTopicConf RdKafkaTopicConfTPtr
+
 data KafkaType = KafkaConsumer | KafkaProducer
-data Kafka = Kafka { kafkaPtr :: RdKafkaTPtr}
+data Kafka = Kafka { kafkaPtr :: RdKafkaTPtr, _kafkaConf :: KafkaConf}
 data KafkaTopic = KafkaTopic 
-  { kafkaTopicPtr :: RdKafkaTopicTPtr 
-  , owningKafka :: Kafka -- prevents garbage collection 
-  } 
+    RdKafkaTopicTPtr  
+    Kafka -- Kept around to prevent garbage collection 
+    KafkaTopicConf
 
-data KafkaConf = KafkaConf {kafkaConfPtr :: RdKafkaConfTPtr}
-data KafkaTopicConf = KafkaTopicConf {kafkaTopicConfPtr :: RdKafkaTopicConfTPtr}
 
 kafkaTypeToRdKafkaType :: KafkaType -> RdKafkaTypeT
 kafkaTypeToRdKafkaType KafkaConsumer = RdKafkaConsumer
 kafkaTypeToRdKafkaType KafkaProducer = RdKafkaProducer
 
+-- Because of the rdKafka API, we have to create a temp file to get properties into a string
+supportedKafkaConfProperties :: IO (String)
+supportedKafkaConfProperties = do 
+  withSystemTempFile "haskakafka.rdkafka_properties" $ \tP tH -> do
+    hPrintKafkaProperties tH
+    readFile tP
+
 hPrintKafkaProperties :: Handle -> IO ()
 hPrintKafkaProperties h = handleToCFile h "w" >>= rdKafkaConfPropertiesShow
 
@@ -112,32 +161,58 @@
 newKafkaConf :: IO KafkaConf
 newKafkaConf = newRdKafkaConfT >>= return . KafkaConf
 
+checkConfSetValue :: RdKafkaConfResT -> CCharBufPointer -> IO ()
+checkConfSetValue err charPtr = case err of 
+    RdKafkaConfOk -> return ()
+    RdKafkaConfInvalid -> do 
+      str <- peekCString charPtr
+      throw $ KafkaInvalidConfigurationValue str
+    RdKafkaConfUnknown -> do
+      str <- peekCString charPtr
+      throw $ KafkaUnknownConfigurationKey str
+
+setKafkaConfValue :: KafkaConf -> String -> String -> IO ()
+setKafkaConfValue (KafkaConf confPtr) key value = do
+  allocaBytes nErrorBytes $ \charPtr -> do
+    err <- rdKafkaConfSet confPtr key value charPtr (fromIntegral nErrorBytes)
+    checkConfSetValue err charPtr
+
+setKafkaTopicConfValue :: KafkaTopicConf -> String -> String -> IO ()
+setKafkaTopicConfValue (KafkaTopicConf confPtr) key value = do
+  allocaBytes nErrorBytes $ \charPtr -> do
+    err <- rdKafkaTopicConfSet confPtr key value charPtr (fromIntegral nErrorBytes)
+    checkConfSetValue err charPtr
+
+setKafkaLogLevel :: Kafka -> KafkaLogLevel -> IO ()
+setKafkaLogLevel (Kafka kptr _) level = 
+  rdKafkaSetLogLevel kptr (fromEnum level)
+
 newKafka :: KafkaType -> KafkaConf -> IO Kafka
-newKafka kafkaType (KafkaConf confPtr) = do
+newKafka kafkaType c@(KafkaConf confPtr) = do
     et <- newRdKafkaT (kafkaTypeToRdKafkaType kafkaType) confPtr 
     case et of 
         Left e -> error e
-        Right x -> return $ Kafka x
+        Right x -> return $ Kafka x c
 
 newKafkaTopic :: Kafka -> String -> KafkaTopicConf -> IO (KafkaTopic)
-newKafkaTopic k@(Kafka kafkaPtr) topicName (KafkaTopicConf confPtr) = do
-    et <- newRdKafkaTopicT kafkaPtr topicName confPtr
+newKafkaTopic k@(Kafka kPtr _) tName conf@(KafkaTopicConf confPtr) = do
+    et <- newRdKafkaTopicT kPtr tName confPtr
     case et of 
         Left e -> throw $ KafkaError e
-        Right x -> return $ KafkaTopic x k
+        Right x -> return $ KafkaTopic x k conf
 
 addBrokers :: Kafka -> String -> IO ()
-addBrokers (Kafka kptr) brokerStr = do
+addBrokers (Kafka kptr _) brokerStr = do
     numBrokers <- rdKafkaBrokersAdd kptr brokerStr
     when (numBrokers == 0) 
         (throw $ KafkaBadSpecification "No valid brokers specified")
 
 startConsuming :: KafkaTopic -> Int -> KafkaOffset -> IO ()
-startConsuming (KafkaTopic topicPtr _) partition offset = 
+startConsuming (KafkaTopic topicPtr _ _) partition offset = 
     let trueOffset = case offset of
-                        KafkaOffsetBeginning -> -2
-                        KafkaOffsetEnd -> -1
-                        KafkaOffsetStored -> -1000
+                        KafkaOffsetBeginning -> (- 2)
+                        KafkaOffsetEnd -> (- 1)
+                        KafkaOffsetStored -> (- 1000)
                         KafkaOffset i -> i
     in throwOnError $ rdKafkaConsumeStart topicPtr partition trueOffset
 
@@ -147,29 +222,64 @@
     BSI.memcpy bsptr ptr len
     
 
+fromMessageStorable :: RdKafkaMessageT -> IO (KafkaMessage)
+fromMessageStorable s = do
+    payload <- word8PtrToBS (len'RdKafkaMessageT s) (payload'RdKafkaMessageT s) 
+
+    key <- if (key'RdKafkaMessageT s) == nullPtr then return Nothing 
+           else word8PtrToBS (keyLen'RdKafkaMessageT s) (key'RdKafkaMessageT s) >>= return . Just
+
+    return $ KafkaMessage
+             (partition'RdKafkaMessageT s) 
+             (offset'RdKafkaMessageT s)
+             payload
+             key 
+
 consumeMessage :: KafkaTopic -> Int -> Int -> IO (Either KafkaError KafkaMessage)
-consumeMessage (KafkaTopic topicPtr _) partition timeout = do
-    ptr <- rdKafkaConsume topicPtr (fromIntegral partition) (fromIntegral timeout)
-    withForeignPtr ptr $ \realPtr ->
-        if realPtr == nullPtr then getErrno >>= return . Left . kafkaRespErr 
+consumeMessage (KafkaTopic topicPtr _ _) partition timeout = do
+  ptr <- rdKafkaConsume topicPtr (fromIntegral partition) (fromIntegral timeout)
+  withForeignPtr ptr $ \realPtr ->
+    if realPtr == nullPtr then getErrno >>= return . Left . kafkaRespErr 
+    else do
+        addForeignPtrFinalizer rdKafkaMessageDestroy ptr
+        s <- peek realPtr
+        if (err'RdKafkaMessageT s) /= RdKafkaRespErrNoError then return $ Left $ KafkaResponseError $ err'RdKafkaMessageT s
         else do
-            addForeignPtrFinalizer rdKafkaMessageDestroy ptr
-            s <- peek realPtr
-            if (err'RdKafkaMessageT s) /= RdKafkaRespErrNoError then return $ Left $ KafkaResponseError $ err'RdKafkaMessageT s
-            else do
-                payload <- word8PtrToBS (len'RdKafkaMessageT s) (payload'RdKafkaMessageT s) 
+            payload <- word8PtrToBS (len'RdKafkaMessageT s) (payload'RdKafkaMessageT s) 
 
-                key <- if (key'RdKafkaMessageT s) == nullPtr then return Nothing 
-                       else word8PtrToBS (keyLen'RdKafkaMessageT s) (key'RdKafkaMessageT s) >>= return . Just
+            key <- if (key'RdKafkaMessageT s) == nullPtr then return Nothing 
+                   else word8PtrToBS (keyLen'RdKafkaMessageT s) (key'RdKafkaMessageT s) >>= return . Just
 
-                return $ Right $ KafkaMessage
-                    (partition'RdKafkaMessageT s) 
-                    (offset'RdKafkaMessageT s)
-                    payload
-                    key 
+            return $ Right $ KafkaMessage
+                (partition'RdKafkaMessageT s) 
+                (offset'RdKafkaMessageT s)
+                payload
+                key 
 
+consumeMessageBatch :: KafkaTopic -> Int -> Int -> Int -> IO (Either KafkaError [KafkaMessage])
+consumeMessageBatch (KafkaTopic topicPtr _ _) partition timeout maxMessages = 
+  allocaArray maxMessages $ \outputPtr -> do
+    numMessages <- rdKafkaConsumeBatch topicPtr (fromIntegral partition) timeout outputPtr (fromIntegral maxMessages)
+    if numMessages < 0 then getErrno >>= return . Left . kafkaRespErr
+    else do
+      ms <- forM [0..(numMessages - 1)] $ \mnum -> do 
+              storablePtr <- peekElemOff outputPtr (fromIntegral mnum)
+              storable <- peek storablePtr
+              ret <- fromMessageStorable storable
+              fptr <- newForeignPtr_ storablePtr
+              addForeignPtrFinalizer rdKafkaMessageDestroy fptr
+              return ret
+      return $ Right ms 
+
+storeOffset :: KafkaTopic -> Int -> Int -> IO (Maybe KafkaError)
+storeOffset (KafkaTopic topicPtr _ _) partition offset = do
+  err <- rdKafkaOffsetStore topicPtr (fromIntegral partition) (fromIntegral offset)
+  case err of 
+    RdKafkaRespErrNoError -> return Nothing
+    e -> return $ Just $ KafkaResponseError e
+
 produceMessage :: KafkaTopic -> KafkaProducePartition -> KafkaProduceMessage -> IO (Maybe KafkaError)
-produceMessage (KafkaTopic topicPtr _) partition (KafkaProduceMessage payload) = do
+produceMessage (KafkaTopic topicPtr _ _) partition (KafkaProduceMessage payload) = do
     let (payloadFPtr, payloadOffset, payloadLength) = BSI.toForeignPtr payload
 
     withForeignPtr payloadFPtr $ \payloadPtr -> do
@@ -180,7 +290,11 @@
             copyMsgFlags passedPayload (fromIntegral payloadLength)
             nullPtr (CSize 0) nullPtr
 
-produceMessage (KafkaTopic topicPtr _) partition (KafkaProduceKeyedMessage key payload) = do
+produceMessage _ _ (KafkaProduceKeyedMessage _ _) = undefined
+
+produceKeyedMessage :: KafkaTopic -> KafkaProduceMessage -> IO (Maybe KafkaError)
+produceKeyedMessage _ (KafkaProduceMessage _) = undefined
+produceKeyedMessage (KafkaTopic topicPtr _ _) (KafkaProduceKeyedMessage key payload) = do
     let (payloadFPtr, payloadOffset, payloadLength) = BSI.toForeignPtr payload
         (keyFPtr, keyOffset, keyLength) = BSI.toForeignPtr key
 
@@ -190,12 +304,12 @@
               passedKey = keyPtr `plusPtr` keyOffset
 
           handleProduceErr =<< 
-            rdKafkaProduce topicPtr (producePartitionInteger partition)
+            rdKafkaProduce topicPtr (producePartitionInteger KafkaUnassignedPartition)
               copyMsgFlags passedPayload (fromIntegral payloadLength)
               passedKey (fromIntegral keyLength) nullPtr
 
 produceMessageBatch :: KafkaTopic -> KafkaProducePartition -> [KafkaProduceMessage] -> IO ([(KafkaProduceMessage, KafkaError)])
-produceMessageBatch (KafkaTopic topicPtr _) partition pms = do
+produceMessageBatch (KafkaTopic topicPtr _ _) partition pms = do
   storables <- forM pms produceMessageToMessage
   withArray storables $ \batchPtr -> do
     batchPtrF <- newForeignPtr_ batchPtr
@@ -220,7 +334,7 @@
               , keyLen'RdKafkaMessageT = 0
               , key'RdKafkaMessageT = nullPtr
               }
-    produceMessageToMessage (KafkaProduceKeyedMessage bs kbs) =  do
+    produceMessageToMessage (KafkaProduceKeyedMessage kbs bs) =  do
         let (payloadFPtr, payloadOffset, payloadLength) = BSI.toForeignPtr bs
             (keyFPtr, keyOffset, keyLength) = BSI.toForeignPtr kbs
              
@@ -236,11 +350,49 @@
                 , payload'RdKafkaMessageT = passedPayload
                 , offset'RdKafkaMessageT = 0
                 , keyLen'RdKafkaMessageT = keyLength
-                , key'RdKafkaMessageT = keyPtr
+                , key'RdKafkaMessageT = passedKey
                 }
-      
-      
 
+type ConfigOverrides = [(String, String)]
+withKafkaProducer :: ConfigOverrides -> ConfigOverrides 
+                     -> String -> String 
+                     -> (Kafka -> KafkaTopic -> IO a) 
+                     -> IO a
+withKafkaProducer configOverrides topicConfigOverrides brokerString tName cb =
+  bracket 
+    (do
+      conf <- newKafkaConf
+      mapM_ (\(k, v) -> setKafkaConfValue conf k v) configOverrides
+      kafka <- newKafka KafkaProducer conf
+      addBrokers kafka brokerString
+      topicConf <- newKafkaTopicConf
+      mapM_ (\(k, v) -> setKafkaTopicConfValue topicConf k v) topicConfigOverrides 
+      topic <- newKafkaTopic kafka tName topicConf
+      return (kafka, topic)
+    )
+    (\(kafka, _) -> drainOutQueue kafka)
+    (\(k, t) -> cb k t)
+
+withKafkaConsumer :: ConfigOverrides -> ConfigOverrides 
+                     -> String -> String -> Int -> KafkaOffset 
+                     -> (Kafka -> KafkaTopic -> IO a) 
+                     -> IO a
+withKafkaConsumer configOverrides topicConfigOverrides brokerString tName partition offset cb =
+  bracket
+    (do
+      conf <- newKafkaConf
+      mapM_ (\(k, v) -> setKafkaConfValue conf k v) configOverrides
+      kafka <- newKafka KafkaConsumer conf
+      addBrokers kafka brokerString
+      topicConf <- newKafkaTopicConf
+      mapM_ (\(k, v) -> setKafkaTopicConfValue topicConf k v) topicConfigOverrides 
+      topic <- newKafkaTopic kafka tName topicConf
+      startConsuming topic partition offset
+      return (kafka, topic)
+    )
+    (\(_, topic) -> stopConsuming topic partition)
+    (\(k, t) -> cb k t)
+
 {-# INLINE copyMsgFlags  #-}
 copyMsgFlags :: Int
 copyMsgFlags = rdKafkaMsgFlagCopy
@@ -280,6 +432,7 @@
     , partitionIsrs :: [Int]
     } deriving (Eq, Show, Typeable)
 
+
 getAllMetadata :: Kafka -> Int -> IO (Either KafkaError KafkaMetadata)
 getAllMetadata k timeout = getMetadata k Nothing timeout
 
@@ -294,9 +447,9 @@
       _ -> return $ Left $ KafkaError "Incorrect number of topics returned"
 
 getMetadata :: Kafka -> Maybe KafkaTopic -> Int -> IO (Either KafkaError KafkaMetadata)
-getMetadata (Kafka kPtr) mTopic timeout = alloca $ \mdDblPtr -> do
+getMetadata (Kafka kPtr _) mTopic timeout = alloca $ \mdDblPtr -> do
     err <- case mTopic of  
-      Just (KafkaTopic kTopicPtr _) -> 
+      Just (KafkaTopic kTopicPtr _ _) -> 
         rdKafkaMetadata kPtr False kTopicPtr mdDblPtr timeout
       Nothing -> do
         nullTopic <- newForeignPtr_ nullPtr
@@ -357,10 +510,10 @@
           e -> return $ Left $ KafkaResponseError e
 
 pollEvents :: Kafka -> Int -> IO ()
-pollEvents (Kafka kPtr) timeout = rdKafkaPoll kPtr timeout >> return ()
+pollEvents (Kafka kPtr _) timeout = rdKafkaPoll kPtr timeout >> return ()
 
 outboundQueueLength :: Kafka -> IO (Int)
-outboundQueueLength (Kafka kPtr) = rdKafkaOutqLen kPtr
+outboundQueueLength (Kafka kPtr _) = rdKafkaOutqLen kPtr
 
 drainOutQueue :: Kafka -> IO ()
 drainOutQueue k = do
@@ -373,7 +526,7 @@
 kafkaRespErr (Errno num) = KafkaResponseError $ rdKafkaErrno2err (fromIntegral num)
 
 stopConsuming :: KafkaTopic -> Int -> IO ()
-stopConsuming (KafkaTopic topicPtr _) partition = 
+stopConsuming (KafkaTopic topicPtr _ _) partition = 
     throwOnError $ rdKafkaConsumeStop topicPtr partition
 
 throwOnError :: IO (Maybe String) -> IO ()
@@ -391,6 +544,13 @@
 dumpKafkaConf :: KafkaConf -> IO (Map String String)
 dumpKafkaConf (KafkaConf kptr) = do
     parseDump (\sizeptr -> rdKafkaConfDump kptr sizeptr)
+
+dumpConfFromKafka :: Kafka -> IO (Map String String) 
+dumpConfFromKafka (Kafka _ cfg) = dumpKafkaConf cfg
+
+dumpConfFromKafkaTopic :: KafkaTopic -> IO (Map String String)
+dumpConfFromKafkaTopic (KafkaTopic _ _ conf) = dumpKafkaTopicConf conf
+
 
 parseDump :: (CSizePtr -> IO (Ptr CString)) -> IO (Map String String)
 parseDump cstr = alloca $ \sizeptr -> do
diff --git a/src/Haskakafka/Example.hs b/src/Haskakafka/Example.hs
new file mode 100644
--- /dev/null
+++ b/src/Haskakafka/Example.hs
@@ -0,0 +1,64 @@
+module Haskakafka.Example where
+
+import Haskakafka
+
+import qualified Data.ByteString.Char8 as C8
+
+example :: IO ()
+example = do
+  let 
+      -- Optionally, we can configure certain parameters for Kafka
+      kafkaConfig = [("socket.timeout.ms", "50000")]
+      topicConfig = [("request.timeout.ms", "50000")]
+
+      -- Payloads are just ByteStrings
+      samplePayload = C8.pack "Hello world"
+
+
+  -- withKafkaProducer opens a producer connection and gives us 
+  -- two objects for subsequent use.
+  withKafkaProducer kafkaConfig topicConfig 
+                    "localhost:9092" "test_topic" 
+                    $ \kafka topic -> do
+
+    -- Produce a single unkeyed message to partition 0
+    let message = KafkaProduceMessage samplePayload
+    _ <- produceMessage topic (KafkaSpecifiedPartition 0) message
+
+    -- Produce a single keyed message
+    let keyMessage = KafkaProduceKeyedMessage (C8.pack "Key") samplePayload
+    _ <- produceKeyedMessage topic message
+
+    -- We can also use the batch API for better performance
+    _ <- produceMessageBatch topic KafkaUnassignedPartition [message, keyMessage]
+
+    putStrLn "Done producing messages, here was our config: "
+    dumpConfFromKafka kafka >>= \d -> putStrLn $ "Kafka config: " ++ (show d)
+    dumpConfFromKafkaTopic topic >>= \d -> putStrLn $ "Topic config: " ++ (show d)
+
+
+  -- withKafkaConsumer opens a consumer connection and starts consuming
+  let partition = 0
+  withKafkaConsumer kafkaConfig topicConfig 
+                    "localhost:9092" "test_topic"
+                    partition -- locked to a specific partition for each consumer
+                    KafkaOffsetBeginning -- start reading from beginning (alternatively, use
+                                         -- KafkaOffsetEnd, KafkaOffset or KafkaOffsetStored)
+                    $ \kafka topic -> do
+    -- Consume a single message at a time
+    let timeoutMs = 1000
+    me <- consumeMessage topic partition timeoutMs
+    case me of 
+      (Left err) -> putStrLn $ "Uh oh, an error! " ++ (show err)
+      (Right m) -> putStrLn $ "Woo, payload was " ++ (C8.unpack $ messagePayload m)
+
+    -- For better performance, consume in batches
+    let maxMessages = 10
+    mes <- consumeMessageBatch topic partition timeoutMs maxMessages
+    case mes of 
+      (Left err) -> putStrLn $ "Something went wrong in batch consume! " ++ (show err)
+      (Right ms) -> putStrLn $ "Woohoo, we got " ++ (show $ length ms) ++ " messages"
+
+
+    -- Be a little less noisy
+    setKafkaLogLevel kafka KafkaLogCrit
diff --git a/src/Haskakafka/Internal.chs b/src/Haskakafka/Internal.chs
--- a/src/Haskakafka/Internal.chs
+++ b/src/Haskakafka/Internal.chs
@@ -6,7 +6,7 @@
 import Control.Applicative
 import Control.Monad
 import Data.Word
-import Foreign
+import Foreign hiding (unsafePerformIO)
 import Foreign.C.Error
 import Foreign.C.String
 import Foreign.C.Types
@@ -16,8 +16,6 @@
 import System.Posix.IO
 import System.Posix.Types
 
-import qualified Data.Map.Strict as Map
-
 #include "rdkafka.h"
 
 type CInt64T = {#type int64_t #}
@@ -35,6 +33,10 @@
 rdKafkaMsgFlagCopy :: RdKafkaMsgFlag
 rdKafkaMsgFlagCopy = 0x2
 
+-- Number of bytes allocated for an error buffer
+nErrorBytes ::  Int
+nErrorBytes = 1024 * 8
+
 -- Helper functions
 {#fun pure unsafe rd_kafka_version as ^
     {} -> `Int' #}
@@ -184,6 +186,10 @@
 
 {#pointer *rd_kafka_metadata_t as RdKafkaMetadataTPtr foreign -> RdKafkaMetadataT #}
 
+-- rd_kafka_message
+foreign import ccall unsafe "rdkafka.h &rd_kafka_message_destroy"
+    rdKafkaMessageDestroy :: FunPtr (Ptr RdKafkaMessageT -> IO ())
+
 -- rd_kafka_conf
 {#fun unsafe rd_kafka_conf_new as ^
     {} -> `RdKafkaConfTPtr' #}
@@ -194,6 +200,10 @@
 {#fun unsafe rd_kafka_conf_dup as ^
     {`RdKafkaConfTPtr'} -> `RdKafkaConfTPtr' #}
 
+{#fun unsafe rd_kafka_conf_set as ^
+  {`RdKafkaConfTPtr', `String', `String', id `CCharBufPointer', cIntConv `CSize'} 
+  -> `RdKafkaConfResT' cIntToEnum #}
+
 newRdKafkaConfT :: IO RdKafkaConfTPtr
 newRdKafkaConfT = do
     ret <- rdKafkaConfNew
@@ -218,6 +228,10 @@
 foreign import ccall unsafe "rdkafka.h &rd_kafka_topic_conf_destroy"
     rdKafkaTopicConfDestroy :: FunPtr (Ptr RdKafkaTopicConfT -> IO ())
 
+{#fun unsafe rd_kafka_topic_conf_set as ^
+  {`RdKafkaTopicConfTPtr', `String', `String', id `CCharBufPointer', cIntConv `CSize'} 
+  -> `RdKafkaConfResT' cIntToEnum #}
+
 newRdKafkaTopicConfT :: IO RdKafkaTopicConfTPtr
 newRdKafkaTopicConfT = do
     ret <- rdKafkaTopicConfNew
@@ -235,9 +249,6 @@
 foreign import ccall unsafe "rdkafka.h &rd_kafka_destroy"
     rdKafkaDestroy :: FunPtr (Ptr RdKafkaT -> IO ())
 
-nErrorBytes ::  Int
-nErrorBytes = 1024 * 8
-
 newRdKafkaT :: RdKafkaTypeT -> RdKafkaConfTPtr -> IO (Either String RdKafkaTPtr)
 newRdKafkaT kafkaType confPtr = 
     allocaBytes nErrorBytes $ \charPtr -> do
@@ -252,6 +263,11 @@
 {#fun unsafe rd_kafka_brokers_add as ^
     {`RdKafkaTPtr', `String'} -> `Int' #}
 
+{#fun unsafe rd_kafka_set_log_level as ^
+  {`RdKafkaTPtr', `Int'} -> `()' #}
+
+-- rd_kafka consume
+
 {#fun unsafe rd_kafka_consume_start as rdKafkaConsumeStartInternal
     {`RdKafkaTopicTPtr', cIntConv `CInt32T', cIntConv `CInt64T'} -> `Int' #}
 
@@ -265,10 +281,11 @@
     {`RdKafkaTopicTPtr', cIntConv `CInt32T'} -> `Int' #}
 
 {#fun unsafe rd_kafka_consume as ^
-    {`RdKafkaTopicTPtr', cIntConv `CInt32T', `Int'} -> `RdKafkaMessageTPtr' #}
+  {`RdKafkaTopicTPtr', cIntConv `CInt32T', `Int'} -> `RdKafkaMessageTPtr' #}
 
-foreign import ccall unsafe "rdkafka.h &rd_kafka_message_destroy"
-    rdKafkaMessageDestroy :: FunPtr (Ptr RdKafkaMessageT -> IO ())
+{#fun unsafe rd_kafka_consume_batch as ^
+  {`RdKafkaTopicTPtr', cIntConv `CInt32T', `Int', castPtr `Ptr (Ptr RdKafkaMessageT)', cIntConv `CSize'}
+  -> `CSize' cIntConv #}
 
 rdKafkaConsumeStop :: RdKafkaTopicTPtr -> Int -> IO (Maybe String)
 rdKafkaConsumeStop topicPtr partition = do
@@ -277,7 +294,12 @@
         -1 -> kafkaErrnoString >>= return . Just
         _ -> return Nothing
 
+{#fun unsafe rd_kafka_offset_store as rdKafkaOffsetStore
+  {`RdKafkaTopicTPtr', cIntConv `CInt32T', cIntConv `CInt64T'} 
+  -> `RdKafkaRespErrT' cIntToEnum #}
 
+-- rd_kafka produce
+
 {#fun unsafe rd_kafka_produce as ^
     {`RdKafkaTopicTPtr', cIntConv `CInt32T', `Int', castPtr `Word8Ptr', 
      cIntConv `CSize', castPtr `Word8Ptr', cIntConv `CSize', castPtr `Word8Ptr'}
@@ -289,6 +311,8 @@
 castMetadata :: Ptr (Ptr RdKafkaMetadataT) -> Ptr (Ptr ())
 castMetadata ptr = castPtr ptr
 
+-- rd_kafka_metadata
+
 {#fun unsafe rd_kafka_metadata as ^
    {`RdKafkaTPtr', boolToCInt `Bool', `RdKafkaTopicTPtr', 
     castMetadata `Ptr (Ptr RdKafkaMetadataT)', `Int'}
@@ -323,7 +347,6 @@
         else do
             addForeignPtrFinalizer rdKafkaTopicDestroy ret
             return $ Right ret
-
 
 -- Marshall / Unmarshall
 enumToCInt :: Enum a => a -> CInt
diff --git a/tests/TestMain.hs b/tests/TestMain.hs
--- a/tests/TestMain.hs
+++ b/tests/TestMain.hs
@@ -1,64 +1,185 @@
-module Main (testmain, main) where
-
+{-# LANGUAGE ScopedTypeVariables #-} 
+module Main (main) where
 import Haskakafka
 
+import Control.Exception
 import Control.Monad
+import System.Environment
 import Test.Hspec
-import Control.Exception (evaluate)
+import Text.Regex.Posix
 
-import qualified Data.ByteString.Char8 as BS
+import qualified Data.Map as Map
+import qualified Data.ByteString.Char8 as C8
 
+brokerAddress :: IO String
+brokerAddress = (getEnv "HASKAKAFKA_TEST_BROKER") `catch` \(_ :: SomeException) -> (return "localhost:9092")
+brokerTopic :: IO String
+brokerTopic = (getEnv "HASKAKAFKA_TEST_TOPIC") `catch` \(_ :: SomeException) -> (return "haskakafka_tests")
+kafkaDelay :: Int -- Little delay for operation
+kafkaDelay = 5 *  1000
+
+getAddressTopic :: (String -> String -> IO ()) -> IO ()
+getAddressTopic cb = do
+  b <- brokerAddress
+  t <- brokerTopic
+  cb b t
+
+sampleProduceMessages :: [KafkaProduceMessage]
+sampleProduceMessages = 
+  [ (KafkaProduceMessage $ C8.pack "hello")
+  , (KafkaProduceKeyedMessage (C8.pack "key") (C8.pack "value"))
+  , (KafkaProduceMessage $ C8.pack "goodbye")
+  ]
+
+shouldBeProduceConsume :: KafkaProduceMessage -> KafkaMessage -> IO ()
+shouldBeProduceConsume (KafkaProduceMessage ppayload) m = do
+  (messagePayload m) `shouldBe` ppayload
+  (messageKey m) `shouldBe` Nothing
+
+shouldBeProduceConsume (KafkaProduceKeyedMessage pkey ppayload) m = do
+  ppayload `shouldBe` (messagePayload m)
+  (Just pkey) `shouldBe` (messageKey m)
+
+shouldBeEmptyTopic :: KafkaTopic  -> IO ()
+shouldBeEmptyTopic kt = do
+  eof <- consumeMessage kt 0 kafkaDelay
+  eof `shouldBe` (Left $ KafkaResponseError $ RdKafkaRespErrPartitionEof)
+
 testmain :: IO ()
 testmain = hspec $ do
-  describe "Prelude.head" $ do
-    it "returns the first element of a list" $ do
-      head [23 ..] `shouldBe` (23 :: Int)
+  describe "RdKafka versioning" $ do
+    it "should be a valid version number" $ do
+      rdKafkaVersionStr `shouldSatisfy` (=~"[0-9]+(.[0-9]+)+")
 
-    it "throws an exception if used with an empty list" $ do
-      evaluate (head []) `shouldThrow` anyException
+  describe "Supported properties" $ do
+    it "should list supported properties" $ do
+      props <- supportedKafkaConfProperties
+      props `shouldSatisfy` (\x -> (length x) > 0)
 
-doConsume :: IO ()
-doConsume = do
-    kConf <- newKafkaConf
-    kafka <- newKafka KafkaConsumer kConf
-    addBrokers kafka "localhost:9092"
-    kTopicConf <- newKafkaTopicConf
-    topic <- newKafkaTopic kafka "test" kTopicConf
+  describe "Kafka Configuration" $ do
+    it "should allow dumping" $ do
+      kConf <- newKafkaConf 
+      kvs <- dumpKafkaConf kConf
+      (Map.size kvs) `shouldSatisfy` (>0)
 
-    startConsuming topic 0 (KafkaOffsetBeginning)
-    _ <- forever $ do
-          m <- consumeMessage topic 0 (1000 * 1000)
-          print m
-    stopConsuming topic 0
+    it "should change when set is called" $ do
+      kConf <- newKafkaConf 
+      setKafkaConfValue kConf "socket.timeout.ms" "50000"
+      kvs <- dumpKafkaConf kConf
+      (kvs Map.! "socket.timeout.ms") `shouldBe` "50000"
 
-doProduce :: IO ()
-doProduce = do
-    kConf <- newKafkaConf
-    kafka <- newKafka KafkaProducer kConf
-    addBrokers kafka "localhost:9092"
-    kTopicConf <- newKafkaTopicConf
-    topic <- newKafkaTopic kafka "test" kTopicConf
-    let me = KafkaMessage 0 0 (BS.pack "hi") Nothing
-    err <- produceMessage topic me
+    it "should throw an exception on unknown property" $ do
+      kConf <- newKafkaConf
+      (setKafkaConfValue kConf "blippity.blop.cosby" "120") `shouldThrow`
+        (\(KafkaUnknownConfigurationKey str) -> (length str) > 0)
 
-    drainOutQueue kafka
-            
-    print err
+    it "should throw an exception on an invalid value" $ do
+      kConf <- newKafkaConf
+      (setKafkaConfValue kConf "socket.timeout.ms" "monorail") `shouldThrow`
+        (\(KafkaInvalidConfigurationValue str) -> (length str) > 0)
+  
+  describe "Kafka topic configuration" $ do
+    it "should allow dumping" $ do
+      kConf <- newKafkaTopicConf
+      kvs <- dumpKafkaTopicConf kConf
+      (Map.size kvs) `shouldSatisfy` (>0)
 
+    it "should change when set is called" $ do
+      kConf <- newKafkaTopicConf
+      setKafkaTopicConfValue kConf "request.timeout.ms" "20000"
+      kvs <- dumpKafkaTopicConf kConf
+      (kvs Map.! "request.timeout.ms") `shouldBe` "20000"
 
-main :: IO ()
-main = do
-    doProduce
-    doConsume
+    it "should throw an exception on unknown property" $ do
+      kConf <- newKafkaTopicConf
+      (setKafkaTopicConfValue kConf "blippity.blop.cosby" "120") `shouldThrow`
+        (\(KafkaUnknownConfigurationKey str) -> (length str) > 0)
 
+    it "should throw an exception on an invalid value" $ do
+      kConf <- newKafkaTopicConf
+      (setKafkaTopicConfValue kConf "request.timeout.ms" "mono...doh!") `shouldThrow`
+        (\(KafkaInvalidConfigurationValue str) -> (length str) > 0)
 
-    -- hPrintKafkaProperties stdout
-    --o <- c_stdout
-    --rdKafkaConfPropertiesShow o
-    --putStrLn $ "Kafka version: " ++ rdKafkaVersionStr
-    --t <- c_rd_kafka_topic_conf_new
-    --topics <- dumpTopics t
-    --print topics
-    --c_rd_kafka_topic_conf_destroy t
-    --putStrLn $ "Cleaned up"
-  
+  describe "Logging" $ do
+    it "should allow setting of log level" $ getAddressTopic $ \a t -> do
+      withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \kafka _ -> do
+        setKafkaLogLevel kafka KafkaLogDebug 
+
+  describe "Consume and produce cycle" $ do
+    it "should be able to produce and consume a unkeyed message off of the broker" $ getAddressTopic $ \a t -> do
+      let message = KafkaProduceMessage (C8.pack "hey hey we're the monkeys")
+      withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \_ topic -> do
+        eof <- consumeMessage topic 0 kafkaDelay
+        eof `shouldBe` (Left $ KafkaResponseError $ RdKafkaRespErrPartitionEof)
+        perr <- withKafkaProducer [] [] a t $ \_ producerTopic -> do 
+                produceMessage producerTopic (KafkaSpecifiedPartition 0) message
+        perr `shouldBe` Nothing
+        
+        et <- consumeMessage topic 0 kafkaDelay
+        case et of 
+          Left err -> error $ show err
+          Right m -> message `shouldBeProduceConsume` m
+
+    it "should be able to produce and consume a keyed message" $ getAddressTopic $ \a t -> do
+      let message = KafkaProduceKeyedMessage (C8.pack "key") (C8.pack "monkey around")
+
+      withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \_ topic -> do
+        shouldBeEmptyTopic topic
+        perr <- withKafkaProducer [] [] a t $ \_ producerTopic -> do
+                  produceKeyedMessage producerTopic message
+        perr `shouldBe` Nothing
+
+        et <- consumeMessage topic 0 kafkaDelay
+        case et of
+          Left err -> error $ show err
+          Right m -> message `shouldBeProduceConsume` m
+
+    it "should be able to batch produce messages" $ getAddressTopic $ \a t -> do
+      withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \_ topic -> do
+        shouldBeEmptyTopic topic
+        errs <- withKafkaProducer [] [] a t $ \_ producerTopic -> do
+                  produceMessageBatch producerTopic (KafkaSpecifiedPartition 0 ) sampleProduceMessages
+        errs `shouldBe` []
+
+        ets <- mapM (\_ -> consumeMessage topic 0 kafkaDelay) ([1..3] :: [Integer])
+
+        forM_ (zip sampleProduceMessages ets) $ \(pm, et) -> 
+          case (pm, et) of
+            (_, Left err) -> error $ show err
+            (pmessage, Right cm) -> pmessage `shouldBeProduceConsume` cm
+
+    it "should be able to batch consume messages" $ getAddressTopic $ \a t -> do
+      withKafkaConsumer [] [] a t 0 KafkaOffsetEnd $ \_ topic -> do
+        shouldBeEmptyTopic topic
+        errs <- withKafkaProducer [] [] a t $ \_ producerTopic -> do
+                  produceMessageBatch producerTopic (KafkaSpecifiedPartition 0 ) sampleProduceMessages
+        errs `shouldBe` []
+
+        et <- consumeMessageBatch topic 0 (1000) 3
+        case et of 
+          (Left err) -> error $ show err
+          (Right oms) -> do
+            (length oms) `shouldBe` 3
+            forM_ (zip sampleProduceMessages oms) $ \(pm, om) -> pm `shouldBeProduceConsume` om
+
+-- Test setup (error on no Kafka)
+checkForKafka :: IO (Bool)
+checkForKafka = do
+  kConf <- newKafkaConf 
+  kafka <- newKafka KafkaConsumer kConf
+  a <- brokerAddress
+  addBrokers kafka a
+  me <- getAllMetadata kafka 1000
+  return $ case me of 
+    (Left _) -> False
+    (Right _) -> True
+
+main :: IO () 
+main = do 
+  hasKafka <- checkForKafka 
+  if hasKafka then testmain
+  else error "\n\n\
+    \*******************************************************************************\n\
+    \*Haskakafka's tests require an operable Kafka broker running on localhost:9092*\n\
+    \*please follow the guide in Readme.md to set this up                          *\n\
+    \*******************************************************************************\n"