hw-kafka-client 2.3.1 → 2.3.2

raw patch · 4 files changed

+37/−3 lines, 4 files

Files

• hw-kafka-client.cabal +1/−1
• src/Kafka/Consumer.hs +1/−1
• src/Kafka/Consumer/Convert.hs +8/−0
• tests-it/Kafka/IntegrationSpec.hs +27/−1

@@ -1,5 +1,5 @@ name:                hw-kafka-client-version:             2.3.1+version:             2.3.2 homepage:            https://github.com/haskell-works/hw-kafka-client bug-reports:         https://github.com/haskell-works/hw-kafka-client/issues license:             MIT

@@ -102,7 +102,7 @@                     -> ConsumerRecord k v                     -> m (Maybe KafkaError) commitOffsetMessage o k m =-  liftIO $ toNativeTopicPartitionList [topicPartitionFromMessage m] >>= commitOffsets o k+  liftIO $ toNativeTopicPartitionList [topicPartitionFromMessageForCommit m] >>= commitOffsets o k  -- | Commit offsets for all currently assigned partitions. commitAllOffsets :: MonadIO m

@@ -95,6 +95,14 @@   let (Offset moff) = crOffset m    in TopicPartition (crTopic m) (crPartition m) (PartitionOffset moff) +-- | Creates a topic partition message for use with the offset commit message.+-- We increment the offset by 1 here because when we commit, the offset is the position+-- the consumer reads from to process the next message.+topicPartitionFromMessageForCommit :: ConsumerRecord k v -> TopicPartition+topicPartitionFromMessageForCommit m =+  let (TopicPartition t p (PartitionOffset moff)) = topicPartitionFromMessage m+   in TopicPartition t p (PartitionOffset $ moff + 1)+ toMap :: Ord k => [(k, v)] -> Map k [v] toMap kvs = fromListWith (++) [(k, [v]) | (k, v) <- kvs] 

@@ -4,7 +4,7 @@ module Kafka.IntegrationSpec where -import           Control.Monad       (forM_)+import           Control.Monad       (forM_, forM) import           Control.Monad.Loops import qualified Data.ByteString     as BS import           Data.Either@@ -22,6 +22,32 @@ spec :: Spec spec = do     describe "Kafka.IntegrationSpec" $ do+        specWithProducer "Run producer" $ do+            it "sends messages to test topic" $ \prod -> do+                res    <- sendMessages (testMessages testTopic) prod+                res `shouldBe` Right ()++        specWithConsumer "Consumer with per-message commit" $ do+            it "should receive 2 messages" $ \k -> do+                res <- receiveMessages k+                length <$> res `shouldBe` Right 2++                comRes <- forM res . mapM $ commitOffsetMessage OffsetCommit k+                comRes `shouldBe` Right [Nothing, Nothing]++        specWithProducer "Run producer again" $ do+            it "sends messages to test topic" $ \prod -> do+                res    <- sendMessages (testMessages testTopic) prod+                res `shouldBe` Right ()++        specWithConsumer "Consumer after per-message commit" $ do+            it "should receive 2 messages again" $ \k -> do+                res <- receiveMessages k+                comRes <- commitAllOffsets OffsetCommit k++                length <$> res `shouldBe` Right 2+                comRes `shouldBe` Nothing+         specWithProducer "Run producer" $ do             it "sends messages to test topic" $ \prod -> do                 res    <- sendMessages (testMessages testTopic) prod