hw-kafka-client 2.4.1 → 2.4.3
raw patch · 6 files changed
+179/−147 lines, 6 files
Files
- hw-kafka-client.cabal +135/−115
- src/Kafka/Consumer.hs +8/−8
- src/Kafka/Consumer/Callbacks.hs +13/−6
- src/Kafka/Consumer/Types.hs +2/−2
- src/Kafka/Types.hs +7/−9
- tests-it/Kafka/IntegrationSpec.hs +14/−7
hw-kafka-client.cabal view
@@ -1,128 +1,148 @@-name: hw-kafka-client-version: 2.4.1-homepage: https://github.com/haskell-works/hw-kafka-client-bug-reports: https://github.com/haskell-works/hw-kafka-client/issues-license: MIT-license-file: LICENSE-author: Alexey Raga <alexey.raga@gmail.com>-maintainer: Alexey Raga <alexey.raga@gmail.com>-category: Database-build-type: Simple-cabal-version: >=1.10-synopsis: Kafka bindings for Haskell-description: Apache Kafka bindings backed by the librdkafka C library.- .- Features include:- .- * Consumer groups: auto-rebalancing consumers- .- * Keyed and keyless messages producing/consuming- .- * Batch producing messages-+-- This file has been generated from package.yaml by hpack version 0.20.0.+--+-- see: https://github.com/sol/hpack+--+-- hash: 8d76dba4ea94855b431d568543c5cd5d2f86a3bab474cd1151a9f936673891fc -extra-source-files: README.md+name: hw-kafka-client+version: 2.4.3+synopsis: Kafka bindings for Haskell+description: Apache Kafka bindings backed by the librdkafka C library.+ .+ Features include:+ .+ * Consumer groups: auto-rebalancing consumers+ .+ * Keyed and keyless messages producing/consuming+ .+ * Batch producing messages+category: Database+homepage: https://github.com/haskell-works/hw-kafka-client+bug-reports: https://github.com/haskell-works/hw-kafka-client/issues+author: Alexey Raga <alexey.raga@gmail.com>+maintainer: Alexey Raga <alexey.raga@gmail.com>+license: MIT+license-file: LICENSE+build-type: Simple+cabal-version: >= 1.10 -flag examples- default: False- manual: True- description: Also compile examples+extra-source-files:+ README.md source-repository head- type: git+ type: git location: git://github.com/haskell-works/hw-kafka-client.git -executable kafka-client-example- main-is: Main.hs- other-modules: ConsumerExample- , ProducerExample- hs-source-dirs: example- default-language: Haskell2010- build-depends: base >=4.6 && < 5- , bifunctors- , bytestring- , containers- , temporary- , transformers- , unix- , hw-kafka-client- ghc-options: -threaded -rtsopts-- if flag(examples)- buildable: True- else- buildable: False+flag examples+ description: Also compile examples+ manual: True+ default: False library- Build-tools: c2hs- build-depends: base >=4.6 && < 5- , bifunctors- , bytestring- , containers- , temporary- , transformers- , unix- if impl(ghc < 8.0)- build-depends: semigroups-+ hs-source-dirs:+ src+ ghc-options: -Wall+ extra-libraries:+ rdkafka+ build-depends:+ base >=4.6 && <5+ , bifunctors+ , bytestring+ , containers+ , temporary+ , transformers+ , unix+ build-tools:+ c2hs+ if impl(ghc <8.0)+ build-depends:+ semigroups+ if os(osx)+ cpp-options: -D__attribute__(A)= -D_Nullable= -D_Nonnull= exposed-modules:- Kafka.Consumer- Kafka.Consumer.ConsumerProperties- Kafka.Consumer.Subscription- Kafka.Consumer.Types- Kafka.Dump- Kafka.Metadata- Kafka.Producer- Kafka.Producer.ProducerProperties- Kafka.Producer.Types- Kafka.Types+ Kafka.Consumer+ Kafka.Consumer.ConsumerProperties+ Kafka.Consumer.Subscription+ Kafka.Consumer.Types+ Kafka.Dump+ Kafka.Metadata+ Kafka.Producer+ Kafka.Producer.ProducerProperties+ Kafka.Producer.Types+ Kafka.Types other-modules:- Kafka.Callbacks- Kafka.Consumer.Callbacks- Kafka.Consumer.Convert- Kafka.Internal.CancellationToken- Kafka.Internal.RdKafka- Kafka.Internal.Setup- Kafka.Internal.Shared- Kafka.Producer.Convert- Kafka.Producer.Callbacks- hs-source-dirs: src- default-language: Haskell2010- ghc-options: -Wall- extra-libraries: rdkafka- if os(darwin)- cpp-options: -D__attribute__(A)= -D_Nullable= -D_Nonnull=+ Kafka.Callbacks+ Kafka.Consumer.Callbacks+ Kafka.Consumer.Convert+ Kafka.Internal.CancellationToken+ Kafka.Internal.RdKafka+ Kafka.Internal.Setup+ Kafka.Internal.Shared+ Kafka.Producer.Callbacks+ Kafka.Producer.Convert+ Paths_hw_kafka_client+ default-language: Haskell2010 -test-suite tests- type: exitcode-stdio-1.0- default-language: Haskell2010- hs-source-dirs: tests- other-modules: Kafka.Consumer.ConsumerRecordMapSpec- , Kafka.Consumer.ConsumerRecordTraverseSpec- main-is: Spec.hs- ghc-options: -Wall -threaded- build-depends: base >=4.6 && < 5- , bifunctors- , bytestring- , containers- , hw-kafka-client- , monad-loops- , hspec- , either+executable kafka-client-example+ main-is: Main.hs+ hs-source-dirs:+ example+ ghc-options: -threaded -rtsopts+ build-depends:+ base >=4.6 && <5+ , bifunctors+ , bytestring+ , containers+ , hw-kafka-client+ , temporary+ , transformers+ , unix+ if !(flag(examples))+ buildable: False+ other-modules:+ ConsumerExample+ ProducerExample+ Paths_hw_kafka_client+ default-language: Haskell2010 test-suite integration-tests- type: exitcode-stdio-1.0- default-language: Haskell2010- hs-source-dirs: tests-it- other-modules: Kafka.IntegrationSpec- , Kafka.TestEnv- main-is: Spec.hs- ghc-options: -Wall -threaded- build-depends: base >=4.6 && < 5- , bifunctors- , bytestring- , containers- , hw-kafka-client- , monad-loops- , hspec- , either+ type: exitcode-stdio-1.0+ main-is: Spec.hs+ hs-source-dirs:+ tests-it+ ghc-options: -Wall -threaded+ build-depends:+ base >=4.6 && <5+ , bifunctors+ , bytestring+ , containers+ , either+ , hspec+ , hw-kafka-client+ , monad-loops+ other-modules:+ Kafka.IntegrationSpec+ Kafka.TestEnv+ Paths_hw_kafka_client+ default-language: Haskell2010++test-suite tests+ type: exitcode-stdio-1.0+ main-is: Spec.hs+ hs-source-dirs:+ tests+ ghc-options: -Wall -threaded+ build-depends:+ base >=4.6 && <5+ , bifunctors+ , bytestring+ , containers+ , either+ , hspec+ , hw-kafka-client+ , monad-loops+ other-modules:+ Kafka.Consumer.ConsumerRecordMapSpec+ Kafka.Consumer.ConsumerRecordTraverseSpec+ Paths_hw_kafka_client+ default-language: Haskell2010
src/Kafka/Consumer.hs view
@@ -259,7 +259,7 @@ (kafkaErrorToMaybe . KafkaResponseError) <$> rdKafkaCommit k pl (offsetCommitToBool o) commitOffsetsStore :: KafkaConsumer -> RdKafkaTopicPartitionListTPtr -> IO (Maybe KafkaError)-commitOffsetsStore (KafkaConsumer (Kafka k) _) pl = do+commitOffsetsStore (KafkaConsumer (Kafka k) _) pl = (kafkaErrorToMaybe . KafkaResponseError) <$> rdKafkaOffsetsStore k pl setConsumerLogLevel :: KafkaConsumer -> KafkaLogLevel -> IO ()@@ -272,11 +272,11 @@ runConsumerLoop :: KafkaConsumer -> CancellationToken -> Maybe Timeout -> IO () runConsumerLoop k ct timeout =- when rtsSupportsBoundThreads $ void $ forkIO go- where- go = do- token <- CToken.status ct- case token of- Running -> pollConsumerEvents k timeout >> go- Cancelled -> return ()+ when rtsSupportsBoundThreads $ void $ forkIO go+ where+ go = do+ token <- CToken.status ct+ case token of+ Running -> pollConsumerEvents k timeout >> go+ Cancelled -> return ()
src/Kafka/Consumer/Callbacks.hs view
@@ -1,3 +1,4 @@+{-# LANGUAGE BangPatterns #-} module Kafka.Consumer.Callbacks ( rebalanceCallback , offsetCommitCallback@@ -17,6 +18,8 @@ import Kafka.Internal.Shared import Kafka.Types +import Control.Concurrent (threadDelay)+ -- | Sets a callback that is called when rebalance is needed. -- -- Callback implementations suppose to watch for 'KafkaResponseError' 'RdKafkaRespErrAssignPartitions' and@@ -71,13 +74,19 @@ setRebalanceCallback f k e pls = do ps <- fromNativeTopicPartitionList'' pls let assignment = (tpTopicName &&& tpPartition) <$> ps- case e of KafkaResponseError RdKafkaRespErrAssignPartitions -> do mbq <- getRdMsgQueue $ getKafkaConf k case mbq of Nothing -> pure ()- Just mq -> forM_ ps (\tp -> redirectPartitionQueue (getKafka k) (tpTopicName tp) (tpPartition tp) mq)+ Just mq -> do+ forM_ ps (\tp -> redirectPartitionQueue (getKafka k) (tpTopicName tp) (tpPartition tp) mq)+ -- sleep for 1 second.+ -- it looks like without it there is not enough time for redirect to happen+ -- or something similarly strange. I don't understand it.+ -- If you know WTH is going on PLEASE let me know because the current "fix" is ugly+ -- and is completely unreasonable :(+ threadDelay 1000000 f k (RebalanceBeforeAssign assignment) void $ assign' k pls -- pass as pointer to avoid possible serialisation issues f k (RebalanceAssign assignment)@@ -100,8 +109,6 @@ -- | Assigns specified partitions to a current consumer. -- Assigning an empty list means unassigning from all partitions that are currently assigned. assign' :: KafkaConsumer -> RdKafkaTopicPartitionListTPtr -> IO (Maybe KafkaError)-assign' (KafkaConsumer (Kafka k) _) pls = do- let er = KafkaResponseError <$> rdKafkaAssign k pls- m <- kafkaErrorToMaybe <$> er- return m+assign' (KafkaConsumer (Kafka k) _) pls =+ (kafkaErrorToMaybe . KafkaResponseError) <$> rdKafkaAssign k pls
src/Kafka/Consumer/Types.hs view
@@ -24,8 +24,8 @@ getKafkaConf = kcKafkaConf {-# INLINE getKafkaConf #-} -newtype ConsumerGroupId = ConsumerGroupId String deriving (Show, Eq)-newtype Offset = Offset Int64 deriving (Show, Eq, Read)+newtype ConsumerGroupId = ConsumerGroupId { unConsumerGroupId :: String} deriving (Show, Ord, Eq)+newtype Offset = Offset { unOffset :: Int64 } deriving (Show, Eq, Ord, Read) data OffsetReset = Earliest | Latest deriving (Show, Eq) -- | A set of events which happen during the rebalancing process
src/Kafka/Types.hs view
@@ -8,13 +8,11 @@ import Data.Typeable import Kafka.Internal.RdKafka -newtype BrokerId =- BrokerId Int- deriving (Show, Eq, Ord, Read)+newtype BrokerId = BrokerId { unBrokerId :: Int} deriving (Show, Eq, Ord, Read) -newtype PartitionId = PartitionId Int deriving (Show, Eq, Read, Ord, Enum)-newtype Millis = Millis Int64 deriving (Show, Read, Eq, Ord, Num)-newtype ClientId = ClientId String deriving (Show, Eq, Ord)+newtype PartitionId = PartitionId { unPartitionId :: Int} deriving (Show, Eq, Read, Ord, Enum)+newtype Millis = Millis { unMillis :: Int64 } deriving (Show, Read, Eq, Ord, Num)+newtype ClientId = ClientId { unClientId :: String} deriving (Show, Eq, Ord) -- | Topic name to be consumed --@@ -23,14 +21,14 @@ -- 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 String -- ^ a simple topic name or a regex if started with @^@+ TopicName { unTopicName :: String } -- ^ a simple topic name or a regex if started with @^@ deriving (Show, Eq, Ord, Read) -- | Kafka broker address string (e.g. @broker1:9092@)-newtype BrokerAddress = BrokerAddress String deriving (Show, Eq)+newtype BrokerAddress = BrokerAddress { unBrokerAddress :: String } deriving (Show, Eq) -- | Timeout in milliseconds-newtype Timeout = Timeout Int deriving (Show, Eq, Read)+newtype Timeout = Timeout { unTimeout :: Int } deriving (Show, Eq, Read) -- | Log levels for /librdkafka/. data KafkaLogLevel =
tests-it/Kafka/IntegrationSpec.hs view
@@ -4,7 +4,7 @@ module Kafka.IntegrationSpec where -import Control.Monad (forM_, forM)+import Control.Monad (forM, forM_) import Control.Monad.Loops import qualified Data.ByteString as BS import Data.Either@@ -197,14 +197,21 @@ ---------------------------------------------------------------------------------------------------------------- +data ReadState = Skip | Read+ receiveMessages :: KafkaConsumer -> IO (Either KafkaError [ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString)]) receiveMessages kafka =- (Right . rights) <$> www- where- www = whileJust maybeMsg return- isOK msg = if msg /= Left (KafkaResponseError RdKafkaRespErrPartitionEof) then Just msg else Nothing- maybeMsg = isOK <$> get- get = pollMessage kafka (Timeout 1000)+ (Right . rights) <$> allMessages+ where+ allMessages =+ unfoldrM (\s -> do+ msg <- pollMessage kafka (Timeout 1000)+ case (s, msg) of+ (Skip, Left _) -> pure $ Just (msg, Skip)+ (_, Right msg') -> pure $ Just (Right msg', Read)+ (Read, _) -> pure Nothing++ ) Skip testMessages :: TopicName -> [ProducerRecord] testMessages t =