diff --git a/hw-kafka-client.cabal b/hw-kafka-client.cabal --- a/hw-kafka-client.cabal +++ b/hw-kafka-client.cabal @@ -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 -maintainer: Alexey Raga -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 +maintainer: Alexey Raga +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 diff --git a/src/Kafka/Consumer.hs b/src/Kafka/Consumer.hs --- a/src/Kafka/Consumer.hs +++ b/src/Kafka/Consumer.hs @@ -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 () diff --git a/src/Kafka/Consumer/Callbacks.hs b/src/Kafka/Consumer/Callbacks.hs --- a/src/Kafka/Consumer/Callbacks.hs +++ b/src/Kafka/Consumer/Callbacks.hs @@ -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 diff --git a/src/Kafka/Consumer/Types.hs b/src/Kafka/Consumer/Types.hs --- a/src/Kafka/Consumer/Types.hs +++ b/src/Kafka/Consumer/Types.hs @@ -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 diff --git a/src/Kafka/Types.hs b/src/Kafka/Types.hs --- a/src/Kafka/Types.hs +++ b/src/Kafka/Types.hs @@ -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 = diff --git a/tests-it/Kafka/IntegrationSpec.hs b/tests-it/Kafka/IntegrationSpec.hs --- a/tests-it/Kafka/IntegrationSpec.hs +++ b/tests-it/Kafka/IntegrationSpec.hs @@ -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 =