hw-kafka-client 2.0.2 → 2.0.3

raw patch · 6 files changed

+39/−8 lines, 6 files

Files

• README.md +1/−1
• example/ConsumerExample.hs +1/−1
• hw-kafka-client.cabal +1/−1
• src/Kafka/Consumer.hs +21/−1
• src/Kafka/Consumer/ConsumerProperties.hs +4/−4
• tests-it/Kafka/IntegrationSpec.hs +11/−0

@@ -42,7 +42,7 @@ consumerProps = brokersList [BrokerAddress "localhost:9092"]              <> groupId (ConsumerGroupId "consumer_example_group")              <> noAutoCommit-             <> consumerLogLevel KafkaLogInfo+             <> logLevel KafkaLogInfo  -- Subscription to topics consumerSub :: Subscription

@@ -14,7 +14,7 @@              <> noAutoCommit              <> setCallback (rebalanceCallback printingRebalanceCallback)              <> setCallback (offsetCommitCallback printingOffsetCallback)-             <> consumerLogLevel KafkaLogInfo+             <> logLevel KafkaLogInfo  -- Subscription to topics consumerSub :: Subscription

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

@@ -4,7 +4,7 @@ , runConsumer , newConsumer , assign, assignment, subscription-, seek+, committed, position, seek , pollMessage , commitOffsetMessage, commitAllOffsets, commitPartitionsOffsets , closeConsumer@@ -164,6 +164,26 @@       let (TopicName tn) = tpTopicName tp       nt <- newRdKafkaTopicT k tn Nothing       return $ bimap KafkaError (,tpPartition tp, tpOffset tp) nt++-- | Retrieve committed offsets for topics+partitions.+committed :: MonadIO m => KafkaConsumer -> Timeout -> [(TopicName, PartitionId)] -> m (Either KafkaError [TopicPartition])+committed (KafkaConsumer (Kafka k) _) (Timeout timeout) tps = liftIO $ do+  ntps <- toNativeTopicPartitionList' tps+  res <- rdKafkaCommitted k ntps timeout+  case res of+    RdKafkaRespErrNoError -> Right <$> fromNativeTopicPartitionList'' ntps+    err                   -> return $ Left (KafkaResponseError err)++-- | Retrieve current positions (last consumed message offset+1) for the current running instance of the consumer.+-- If the current consumer hasn't received any messages for a given partition, 'PartitionOffsetInvalid' is returned.+position :: MonadIO m => KafkaConsumer -> [(TopicName, PartitionId)] -> m (Either KafkaError [TopicPartition])+position (KafkaConsumer (Kafka k) _) tps = liftIO $ do+  ntps <- toNativeTopicPartitionList' tps+  res <- rdKafkaPosition k ntps+  case res of+    RdKafkaRespErrNoError -> Right <$> fromNativeTopicPartitionList'' ntps+    err                   -> return $ Left (KafkaResponseError err)+  -- | Closes the consumer. closeConsumer :: MonadIO m => KafkaConsumer -> m (Maybe KafkaError)

@@ -25,11 +25,11 @@ instance Semigroup ConsumerProperties where   (<>) (ConsumerProperties m1 ll1 cb1) (ConsumerProperties m2 ll2 cb2) =     ConsumerProperties (M.union m2 m1) (ll2 `mplus` ll1) (cb2 <> cb1)-    + instance Monoid ConsumerProperties where   mempty = ConsumerProperties M.empty Nothing []   mappend = (<>)-  + brokersList :: [BrokerAddress] -> ConsumerProperties brokersList bs =   let bs' = L.intercalate "," ((\(BrokerAddress x) -> x) <$> bs)@@ -54,8 +54,8 @@  -- | Sets the logging level. -- Usually is used with 'debugOptions' to configure which logs are needed.-consumerLogLevel :: KafkaLogLevel -> ConsumerProperties-consumerLogLevel ll = ConsumerProperties M.empty (Just ll) []+logLevel :: KafkaLogLevel -> ConsumerProperties+logLevel ll = ConsumerProperties M.empty (Just ll) []  -- | Sets the compression codec for the consumer. compression :: KafkaCompressionCodec -> ConsumerProperties

@@ -28,6 +28,14 @@                 res `shouldBe` Right ()          specWithConsumer "Run consumer" $ do+            it "should get committed" $ \k -> do+                res <- committed k (Timeout 10000) [(testTopic, PartitionId 0)]+                res `shouldSatisfy` isRight++            it "should get position" $ \k -> do+                res <- position k [(testTopic, PartitionId 0)]+                res `shouldSatisfy` isRight+             it "should receive messages" $ \k -> do                 res <- receiveMessages k                 length <$> res `shouldBe` Right 2@@ -35,6 +43,9 @@                 let timestamps = crTimestamp <$> either (const []) id res                 forM_ timestamps $ \ts ->                     ts `shouldNotBe` NoTimestamp++                comRes <- commitAllOffsets OffsetCommit k+                comRes `shouldBe` Nothing              it "should get watermark offsets" $ \k -> do                 res <- sequence <$> watermarkOffsets k (Timeout 1000) testTopic