hw-kafka-client 2.3.2 → 2.3.3
raw patch · 6 files changed
+84/−16 lines, 6 filesdep −regex-posix
Dependencies removed: regex-posix
Files
- hw-kafka-client.cabal +1/−3
- src/Kafka/Consumer.hs +13/−0
- src/Kafka/Consumer/ConsumerProperties.hs +5/−0
- src/Kafka/Internal/RdKafka.chs +4/−0
- tests-it/Kafka/IntegrationSpec.hs +53/−8
- tests-it/Kafka/TestEnv.hs +8/−5
hw-kafka-client.cabal view
@@ -1,5 +1,5 @@ name: hw-kafka-client-version: 2.3.2+version: 2.3.3 homepage: https://github.com/haskell-works/hw-kafka-client bug-reports: https://github.com/haskell-works/hw-kafka-client/issues license: MIT@@ -108,7 +108,6 @@ , hw-kafka-client , monad-loops , hspec- , regex-posix , either test-suite integration-tests@@ -126,5 +125,4 @@ , hw-kafka-client , monad-loops , hspec- , regex-posix , either
src/Kafka/Consumer.hs view
@@ -8,6 +8,7 @@ , committed, position, seek , pollMessage, pollConsumerEvents , commitOffsetMessage, commitAllOffsets, commitPartitionsOffsets+, storeOffsetMessage , closeConsumer -- ReExport Types , KafkaConsumer@@ -104,6 +105,14 @@ commitOffsetMessage o k m = liftIO $ toNativeTopicPartitionList [topicPartitionFromMessageForCommit m] >>= commitOffsets o k +-- | Stores message's offset locally for the message's partition.+storeOffsetMessage :: MonadIO m+ => KafkaConsumer+ -> ConsumerRecord k v+ -> m (Maybe KafkaError)+storeOffsetMessage k m =+ liftIO $ toNativeTopicPartitionList [topicPartitionFromMessageForCommit m] >>= commitOffsetsStore k+ -- | Commit offsets for all currently assigned partitions. commitAllOffsets :: MonadIO m => OffsetCommit@@ -248,6 +257,10 @@ commitOffsets :: OffsetCommit -> KafkaConsumer -> RdKafkaTopicPartitionListTPtr -> IO (Maybe KafkaError) commitOffsets o (KafkaConsumer (Kafka k) _) pl = (kafkaErrorToMaybe . KafkaResponseError) <$> rdKafkaCommit k pl (offsetCommitToBool o)++commitOffsetsStore :: KafkaConsumer -> RdKafkaTopicPartitionListTPtr -> IO (Maybe KafkaError)+commitOffsetsStore (KafkaConsumer (Kafka k) _) pl = do+ (kafkaErrorToMaybe . KafkaResponseError) <$> rdKafkaOffsetsStore k pl setConsumerLogLevel :: KafkaConsumer -> KafkaLogLevel -> IO () setConsumerLogLevel (KafkaConsumer (Kafka k) _) level =
src/Kafka/Consumer/ConsumerProperties.hs view
@@ -44,6 +44,11 @@ noAutoCommit = extraProps $ M.fromList [("enable.auto.commit", "false")] +-- | Disables auto offset store for the consumer+noAutoOffsetStore :: ConsumerProperties+noAutoOffsetStore =+ extraProps $ M.fromList [("enable.auto.offset.store", "false")]+ -- | Consumer group id groupId :: ConsumerGroupId -> ConsumerProperties groupId (ConsumerGroupId cid) =
src/Kafka/Internal/RdKafka.chs view
@@ -855,6 +855,10 @@ {`RdKafkaTopicTPtr', cIntConv `CInt32T', cIntConv `CInt64T'} -> `RdKafkaRespErrT' cIntToEnum #} +{#fun rd_kafka_offsets_store as rdKafkaOffsetsStore+ {`RdKafkaTPtr', `RdKafkaTopicPartitionListTPtr'}+ -> `RdKafkaRespErrT' cIntToEnum #}+ -- rd_kafka produce {#fun rd_kafka_produce as ^
tests-it/Kafka/IntegrationSpec.hs view
@@ -21,14 +21,14 @@ spec :: Spec spec = do- describe "Kafka.IntegrationSpec" $ do+ describe "Per-message commit" $ do specWithProducer "Run producer" $ do- it "sends messages to test topic" $ \prod -> do+ it "1. sends 2 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+ specWithConsumer "Consumer with per-message commit" consumerProps $ do+ it "2. should receive 2 messages" $ \k -> do res <- receiveMessages k length <$> res `shouldBe` Right 2 @@ -36,24 +36,69 @@ comRes `shouldBe` Right [Nothing, Nothing] specWithProducer "Run producer again" $ do- it "sends messages to test topic" $ \prod -> do+ it "3. sends 2 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+ specWithConsumer "Consumer after per-message commit" consumerProps $ do+ it "4. should receive 2 messages again" $ \k -> do res <- receiveMessages k comRes <- commitAllOffsets OffsetCommit k length <$> res `shouldBe` Right 2 comRes `shouldBe` Nothing + describe "Store offsets" $ do specWithProducer "Run producer" $ do+ it "1. sends 2 messages to test topic" $ \prod -> do+ res <- sendMessages (testMessages testTopic) prod+ res `shouldBe` Right ()++ specWithConsumer "Consumer with no auto store" consumerPropsNoStore $ do+ it "2. should receive 2 messages without storing" $ \k -> do+ res <- receiveMessages k+ length <$> res `shouldBe` Right 2++ comRes <- commitAllOffsets OffsetCommit k+ comRes `shouldBe` Just (KafkaResponseError RdKafkaRespErrNoOffset)++ specWithProducer "Run producer again" $ do+ it "3. sends 2 messages to test topic" $ \prod -> do+ res <- sendMessages (testMessages testTopic) prod+ res `shouldBe` Right ()++ specWithConsumer "Consumer after commit without store" consumerPropsNoStore $ do+ it "4. should receive 4 messages and store them" $ \k -> do+ res <- receiveMessages k+ storeRes <- forM res . mapM $ storeOffsetMessage k+ comRes <- commitAllOffsets OffsetCommit k++ length <$> storeRes `shouldBe` Right 4+ length <$> res `shouldBe` Right 4+ comRes `shouldBe` Nothing++ specWithProducer "Run producer again" $ do+ it "5. sends 2 messages to test topic" $ \prod -> do+ res <- sendMessages (testMessages testTopic) prod+ res `shouldBe` Right ()++ specWithConsumer "Consumer after commit with store" consumerPropsNoStore $ do+ it "6. should receive 2 messages" $ \k -> do+ res <- receiveMessages k+ storeRes <- forM res $ mapM (storeOffsetMessage k)+ comRes <- commitAllOffsets OffsetCommit k++ length <$> res `shouldBe` Right 2+ length <$> storeRes `shouldBe` Right 2+ comRes `shouldBe` Nothing++ 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 "Run consumer" $ do+ specWithConsumer "Run consumer" consumerProps $ do it "should get committed" $ \k -> do res <- committed k (Timeout 10000) [(testTopic, PartitionId 0)] res `shouldSatisfy` isRight
tests-it/Kafka/TestEnv.hs view
@@ -34,6 +34,9 @@ <> C.setCallback (errorCallback (\e r -> print $ show e <> ": " <> r)) <> noAutoCommit +consumerPropsNoStore :: BrokerAddress -> ConsumerProperties+consumerPropsNoStore broker = consumerProps broker <> noAutoOffsetStore+ producerProps :: BrokerAddress -> ProducerProperties producerProps broker = P.brokersList [broker] <> P.setCallback (logCallback (\l s1 s2 -> print $ show l <> ": " <> s1 <> ", " <> s2))@@ -48,13 +51,13 @@ (Right p) <- newProducer (producerProps brokerAddress) return p -mkConsumer :: IO KafkaConsumer-mkConsumer = do- (Right c) <- newConsumer (consumerProps brokerAddress) (testSubscription testTopic)+mkConsumerWith :: (BrokerAddress -> ConsumerProperties) -> IO KafkaConsumer+mkConsumerWith props = do+ (Right c) <- newConsumer (props brokerAddress) (testSubscription testTopic) return c -specWithConsumer :: String -> SpecWith KafkaConsumer -> Spec-specWithConsumer s f = beforeAll mkConsumer $ afterAll (void . closeConsumer) $ describe s f+specWithConsumer :: String -> (BrokerAddress -> ConsumerProperties) -> SpecWith KafkaConsumer -> Spec+specWithConsumer s p f = beforeAll (mkConsumerWith p) $ afterAll (void . closeConsumer) $ describe s f specWithProducer :: String -> SpecWith KafkaProducer -> Spec specWithProducer s f = beforeAll mkProducer $ afterAll (void . closeProducer) $ describe s f