hw-kafka-client 4.0.0 → 4.0.1

raw patch · 18 files changed

+56/−47 lines, 18 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• README.md +8/−8
• example/ConsumerExample.hs +4/−4
• example/ProducerExample.hs +2/−2
• hw-kafka-client.cabal +1/−1
• src/Kafka/Consumer.hs +1/−1
• src/Kafka/Consumer/ConsumerProperties.hs +1/−1
• src/Kafka/Consumer/Types.hs +7/−3
• src/Kafka/Internal/RdKafka.chs +4/−4
• src/Kafka/Internal/Shared.hs +2/−2
• src/Kafka/Metadata.hs +3/−3
• src/Kafka/Producer.hs +1/−1
• src/Kafka/Producer/Convert.hs +2/−2
• src/Kafka/Producer/ProducerProperties.hs +1/−1
• src/Kafka/Types.hs +10/−5
• tests-it/Kafka/IntegrationSpec.hs +5/−5
• tests-it/Kafka/TestEnv.hs +2/−2
• tests/Kafka/Consumer/ConsumerRecordMapSpec.hs +1/−1
• tests/Kafka/Consumer/ConsumerRecordTraverseSpec.hs +1/−1

@@ -40,14 +40,14 @@  -- Global consumer properties consumerProps :: ConsumerProperties-consumerProps = brokersList [BrokerAddress "localhost:9092"]-             <> groupId (ConsumerGroupId "consumer_example_group")+consumerProps = brokersList ["localhost:9092"]+             <> groupId "consumer_example_group"              <> noAutoCommit              <> logLevel KafkaLogInfo  -- Subscription to topics consumerSub :: Subscription-consumerSub = topics [TopicName "kafka-client-example-topic"]+consumerSub = topics ["kafka-client-example-topic"]            <> offsetReset Earliest  -- Running an example@@ -58,7 +58,7 @@     where       mkConsumer = newConsumer consumerProps consumerSub       clConsumer (Left err) = return (Left err)-      clConsumer (Right kc) = (maybe (Right ()) Left) <$> closeConsumer kc+      clConsumer (Right kc) = maybe (Right ()) Left <$> closeConsumer kc       runHandler (Left err) = return (Left err)       runHandler (Right kc) = processMessages kc @@ -97,7 +97,7 @@  ```haskell producerProps :: ProducerProperties-producerProps = brokersList [BrokerAddress "localhost:9092"]+producerProps = brokersList ["localhost:9092"]              <> sendTimeout (Timeout 0)           -- for librdkafka "0" means "infinite" (see https://github.com/edenhill/librdkafka/issues/2015) ``` @@ -108,7 +108,7 @@  ```haskell producerProps :: ProducerProperties-producerProps = brokersList [BrokerAddress "localhost:9092"]+producerProps = brokersList ["localhost:9092"]              <> setCallback (deliveryCallback print) ``` @@ -126,12 +126,12 @@  -- Global producer properties producerProps :: ProducerProperties-producerProps = brokersList [BrokerAddress "localhost:9092"]+producerProps = brokersList ["localhost:9092"]              <> logLevel KafkaLogDebug  -- Topic to send messages to targetTopic :: TopicName-targetTopic = TopicName "kafka-client-example-topic"+targetTopic = "kafka-client-example-topic"  -- Run an example runProducerExample :: IO ()

@@ -11,8 +11,8 @@  -- Global consumer properties consumerProps :: ConsumerProperties-consumerProps = brokersList [BrokerAddress "localhost:9092"]-             <> groupId (ConsumerGroupId "consumer_example_group")+consumerProps = brokersList ["localhost:9092"]+             <> groupId "consumer_example_group"              <> noAutoCommit              <> setCallback (rebalanceCallback printingRebalanceCallback)              <> setCallback (offsetCommitCallback printingOffsetCallback)@@ -20,7 +20,7 @@  -- Subscription to topics consumerSub :: Subscription-consumerSub = topics [TopicName "kafka-client-example-topic"]+consumerSub = topics ["kafka-client-example-topic"]            <> offsetReset Earliest  -- Running an example@@ -32,7 +32,7 @@     where       mkConsumer = newConsumer consumerProps consumerSub       clConsumer (Left err) = return (Left err)-      clConsumer (Right kc) = (maybe (Right ()) Left) <$> closeConsumer kc+      clConsumer (Right kc) = maybe (Right ()) Left <$> closeConsumer kc       runHandler (Left err) = return (Left err)       runHandler (Right kc) = processMessages kc 

@@ -16,14 +16,14 @@  -- Global producer properties producerProps :: ProducerProperties-producerProps = brokersList [BrokerAddress "localhost:9092"]+producerProps = brokersList ["localhost:9092"]              <> sendTimeout (Timeout 10000)              <> setCallback (deliveryCallback print)              <> logLevel KafkaLogDebug  -- Topic to send messages to targetTopic :: TopicName-targetTopic = TopicName "kafka-client-example-topic"+targetTopic = "kafka-client-example-topic"  mkMessage :: Maybe ByteString -> Maybe ByteString -> ProducerRecord mkMessage k v = ProducerRecord

@@ -1,7 +1,7 @@ cabal-version: 2.2  name:                   hw-kafka-client-version:                4.0.0+version:                4.0.1 synopsis:               Kafka bindings for Haskell description:            Apache Kafka bindings backed by the librdkafka C library.                         .

@@ -15,7 +15,7 @@ --  -- -- Global consumer properties -- consumerProps :: 'ConsumerProperties'--- consumerProps = 'brokersList' ['BrokerAddress' "localhost:9092"]+-- consumerProps = 'brokersList' ["localhost:9092"] --              <> 'groupId' ('ConsumerGroupId' "consumer_example_group") --              <> 'noAutoCommit' --              <> 'logLevel' 'KafkaLogInfo'

@@ -77,7 +77,7 @@ -- | Set the <https://kafka.apache.org/documentation/#bootstrap.servers list of brokers> to contact to connect to the Kafka cluster. brokersList :: [BrokerAddress] -> ConsumerProperties brokersList bs =-  let bs' = Text.intercalate "," ((\(BrokerAddress x) -> x) <$> bs)+  let bs' = Text.intercalate "," (unBrokerAddress <$> bs)    in extraProps $ M.fromList [("bootstrap.servers", bs')]  -- | Set the <https://kafka.apache.org/documentation/#auto.commit.interval.ms auto commit interval> and enables <https://kafka.apache.org/documentation/#enable.auto.commit auto commit>.

@@ -1,5 +1,6 @@-{-# LANGUAGE DeriveDataTypeable #-}-{-# LANGUAGE DeriveGeneric      #-}+{-# LANGUAGE DeriveDataTypeable         #-}+{-# LANGUAGE DeriveGeneric              #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}  ----------------------------------------------------------------------------- -- |@@ -37,6 +38,7 @@ import Data.Bifunctor       (Bifunctor (..)) import Data.Bitraversable   (Bitraversable (..), bimapM, bisequence) import Data.Int             (Int64)+import Data.String          (IsString) import Data.Text            (Text) import Data.Typeable        (Typeable) import GHC.Generics         (Generic)@@ -62,7 +64,9 @@ -- | Consumer group ID. Different consumers with the same consumer group ID will get assigned different partitions of each subscribed topic.  --  -- See <https://kafka.apache.org/documentation/#group.id Kafka documentation on consumer group>-newtype ConsumerGroupId = ConsumerGroupId { unConsumerGroupId :: Text } deriving (Show, Ord, Eq, Generic)+newtype ConsumerGroupId = ConsumerGroupId+  { unConsumerGroupId :: Text+  } deriving (Show, Ord, Eq, IsString, Generic)  -- | A message offset in a partition newtype Offset          = Offset { unOffset :: Int64 } deriving (Show, Eq, Ord, Read, Generic)

@@ -415,7 +415,7 @@     withForeignPtr conf $ \c -> rdKafkaConfSetLogCb' c cb'  ---- Stats Callback-type StatsCallback' = Ptr RdKafkaT -> CString -> CSize -> Word8Ptr -> IO ()+type StatsCallback' = Ptr RdKafkaT -> CString -> CSize -> Word8Ptr -> IO CInt type StatsCallback = Ptr RdKafkaT -> ByteString -> IO ()  foreign import ccall safe "wrapper"@@ -426,12 +426,12 @@  rdKafkaConfSetStatsCb :: RdKafkaConfTPtr -> StatsCallback -> IO () rdKafkaConfSetStatsCb conf cb = do-    cb' <- mkStatsCallback $ \k j jl _ -> BS.packCStringLen (j, cIntConv jl) >>= cb k+    cb' <- mkStatsCallback $ \k j jl _ -> BS.packCStringLen (j, cIntConv jl) >>= cb k >> pure 0     withForeignPtr conf $ \c -> rdKafkaConfSetStatsCb' c cb'     return ()  ---- Socket Callback-type SocketCallback = Int -> Int -> Int -> Word8Ptr -> IO ()+type SocketCallback = Int -> Int -> Int -> Word8Ptr -> IO CInt  foreign import ccall safe "wrapper"     mkSocketCallback :: SocketCallback -> IO (FunPtr SocketCallback)@@ -442,7 +442,7 @@ rdKafkaConfSetSocketCb :: RdKafkaConfTPtr -> SocketCallback -> IO () rdKafkaConfSetSocketCb conf cb = do     cb' <- mkSocketCallback cb-    withForeignPtr conf $ \c -> rdKafkaConfSetSocketCb' c cb'+    withForeignPtr conf $ \c -> rdKafkaConfSetSocketCb' c cb' >> pure 0     return ()  {#fun rd_kafka_conf_set_opaque as ^

@@ -35,8 +35,8 @@  pollEvents :: HasKafka a => a -> Maybe Timeout -> IO () pollEvents a tm =-  let timeout = maybe 0 (\(Timeout ms) -> ms) tm-      (Kafka k) = getKafka a+  let timeout = maybe 0 unTimeout tm+      Kafka k = getKafka a   in void (rdKafkaPoll k timeout)  word8PtrToBS :: Int -> Word8Ptr -> IO BS.ByteString

@@ -260,8 +260,8 @@     { pmPartitionId     = PartitionId (id'RdKafkaMetadataPartitionT pm)     , pmError           = kafkaErrorToMaybe $ KafkaResponseError (err'RdKafkaMetadataPartitionT pm)     , pmLeader          = BrokerId (leader'RdKafkaMetadataPartitionT pm)-    , pmReplicas        = (BrokerId . fromIntegral) <$> reps-    , pmInSyncReplicas  = (BrokerId . fromIntegral) <$> isrs+    , pmReplicas        = BrokerId . fromIntegral <$> reps+    , pmInSyncReplicas  = BrokerId . fromIntegral <$> isrs     }  @@ -296,4 +296,4 @@   "Stable"             -> GroupStable   "Dead"               -> GroupDead   "Empty"              -> GroupEmpty-  _                    -> error $ "Unknown group state: " <> (Text.unpack s)+  _                    -> error $ "Unknown group state: " <> Text.unpack s

@@ -15,7 +15,7 @@ --  -- -- Global producer properties -- producerProps :: 'ProducerProperties'--- producerProps = 'brokersList' ['BrokerAddress' "localhost:9092"]+-- producerProps = 'brokersList' ["localhost:9092"] --              <> 'logLevel' 'KafkaLogDebug' --  -- -- Topic to send messages to

@@ -28,13 +28,13 @@ {-# INLINE producePartitionCInt #-}  handleProduceErr :: Int -> IO (Maybe KafkaError)-handleProduceErr (- 1) = (Just . kafkaRespErr) <$> getErrno+handleProduceErr (- 1) = Just . kafkaRespErr <$> getErrno handleProduceErr 0 = return Nothing handleProduceErr _ = return $ Just KafkaInvalidReturnValue {-# INLINE handleProduceErr #-}  handleProduceErr' :: Int -> IO (Either KafkaError ())-handleProduceErr' (- 1) = (Left . kafkaRespErr) <$> getErrno+handleProduceErr' (- 1) = Left . kafkaRespErr <$> getErrno handleProduceErr' 0 = return (Right ()) handleProduceErr' _ = return $ Left KafkaInvalidReturnValue {-# INLINE handleProduceErr' #-}

@@ -60,7 +60,7 @@ -- | Set the <https://kafka.apache.org/documentation/#bootstrap.servers list of brokers> to contact to connect to the Kafka cluster. brokersList :: [BrokerAddress] -> ProducerProperties brokersList bs =-  let bs' = Text.intercalate "," ((\(BrokerAddress x) -> x) <$> bs)+  let bs' = Text.intercalate "," (unBrokerAddress <$> bs)    in extraProps $ M.fromList [("bootstrap.servers", bs')]  -- | Set the producer callback.

@@ -29,6 +29,7 @@  import Control.Exception      (Exception (..)) import Data.Int               (Int64)+import Data.String            (IsString) import Data.Text              (Text, isPrefixOf) import Data.Typeable          (Typeable) import GHC.Generics           (Generic)@@ -46,7 +47,9 @@ -- | Client ID used by Kafka to better track requests --  -- See <https://kafka.apache.org/documentation/#client.id Kafka documentation on client ID>-newtype ClientId    = ClientId { unClientId :: Text } deriving (Show, Eq, Ord, Generic)+newtype ClientId = ClientId+  { unClientId :: Text+  } deriving (Show, Eq, IsString, Ord, Generic)  -- | Batch size used for polling newtype BatchSize   = BatchSize { unBatchSize :: Int } deriving (Show, Read, Eq, Ord, Num, Generic)@@ -63,9 +66,9 @@ -- any topic name in the topics list that is prefixed with @^@ will -- be regex-matched to the full list of topics in the cluster and matching -- topics will be added to the subscription list.-newtype TopicName =-    TopicName { unTopicName :: Text } -- ^ a simple topic name or a regex if started with @^@-    deriving (Show, Eq, Ord, Read, Generic)+newtype TopicName = TopicName+  { unTopicName :: Text -- ^ a simple topic name or a regex if started with @^@+  } deriving (Show, Eq, Ord, IsString, Read, Generic)  -- | Deduce the type of a topic from its name, by checking if it starts with a double underscore "\__" topicType :: TopicName -> TopicType@@ -74,7 +77,9 @@ {-# INLINE topicType #-}  -- | Kafka broker address string (e.g. @broker1:9092@)-newtype BrokerAddress = BrokerAddress { unBrokerAddress :: Text } deriving (Show, Eq, Generic)+newtype BrokerAddress = BrokerAddress+  { unBrokerAddress :: Text+  } deriving (Show, Eq, IsString, Generic)  -- | Timeout in milliseconds newtype Timeout = Timeout { unTimeout :: Int } deriving (Show, Eq, Read, Generic)

@@ -118,7 +118,7 @@                 var <- newEmptyMVar                 let                   msg = ProducerRecord-                    { prTopic = TopicName "callback-topic"+                    { prTopic = "callback-topic"                     , prPartition = UnassignedPartition                     , prKey = Nothing                     , prValue = Just "test from producer"@@ -136,7 +136,7 @@         specWithConsumer "Run consumer with sync polling" (consumerProps <> groupId (makeGroupId "sync") <> callbackPollMode CallbackPollModeSync) runConsumerSpec      describe "Kafka.Consumer.BatchSpec" $ do-        specWithConsumer "Batch consumer" (consumerProps <> groupId (ConsumerGroupId "batch-consumer")) $ do+        specWithConsumer "Batch consumer" (consumerProps <> groupId "batch-consumer") $ do             it "should consume first batch" $ \k -> do                 res <- pollMessageBatch k (Timeout 1000) (BatchSize 5)                 length res `shouldBe` 5@@ -232,8 +232,8 @@   it "should return topic metadata" $ \k -> do     res <- topicMetadata k (Timeout 1000) testTopic     res `shouldSatisfy` isRight-    (length . kmBrokers) <$> res `shouldBe` Right 1-    (length . kmTopics) <$> res `shouldBe` Right 1+    length . kmBrokers <$> res `shouldBe` Right 1+    length . kmTopics <$> res `shouldBe` Right 1    it "should describe all consumer groups" $ \k -> do     res <- allConsumerGroupsInfo k (Timeout 1000)@@ -248,7 +248,7 @@     fmap giGroup <$> res `shouldBe` Right [testGroupId]    it "should describe non-existent consumer group" $ \k -> do-    res <- consumerGroupInfo k (Timeout 1000) (ConsumerGroupId "does-not-exist")+    res <- consumerGroupInfo k (Timeout 1000) "does-not-exist"     res `shouldBe` Right []    it "should read topic offsets for time" $ \k -> do

@@ -24,12 +24,12 @@  brokerAddress :: BrokerAddress brokerAddress = unsafePerformIO $-  (BrokerAddress . Text.pack) <$> getEnv "KAFKA_TEST_BROKER" `catch` \(_ :: SomeException) -> return "localhost:9092"+  BrokerAddress . Text.pack <$> getEnv "KAFKA_TEST_BROKER" `catch` \(_ :: SomeException) -> return "localhost:9092" {-# NOINLINE brokerAddress #-}  testTopic :: TopicName testTopic = unsafePerformIO $-  (TopicName . Text.pack) <$> getEnv "KAFKA_TEST_TOPIC" `catch` \(_ :: SomeException) -> return $ testPrefix <> "-topic"+  TopicName . Text.pack <$> getEnv "KAFKA_TEST_TOPIC" `catch` \(_ :: SomeException) -> return $ testPrefix <> "-topic" {-# NOINLINE testTopic #-}  testGroupId :: ConsumerGroupId

@@ -15,7 +15,7 @@  testRecord :: ConsumerRecord (Maybe Text) (Maybe Text) testRecord = ConsumerRecord-  { crTopic     = TopicName "some-topic"+  { crTopic     = "some-topic"   , crPartition = PartitionId 0   , crOffset    = Offset 5   , crTimestamp = NoTimestamp

@@ -17,7 +17,7 @@  testRecord :: ConsumerRecord Text Text testRecord = ConsumerRecord-  { crTopic     = TopicName "some-topic"+  { crTopic     = "some-topic"   , crPartition = PartitionId 0   , crOffset    = Offset 5   , crTimestamp = NoTimestamp