diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,11 @@ +# 0.1.0.4 + +- Added new `ElsewhereButToKafkaAsWell` mode to `CommitOffsets`, which commits offsets to Kafka once the external Offset storage has been updated. Kafka commits are performed only to keep Kafka informed about consumer lag. + +# 0.1.0.3 + +- Relax version bounds to encompass `text-2.0.x`, `base-4.16.x` and `template-haskell-2.18.x` + # 0.1.0.2 - Support `time-1.x`. diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -1,6 +1,6 @@ BSD 3-Clause License -Copyright (c) 2021, NoRedInk +Copyright (c) 2022, NoRedInk All rights reserved. Redistribution and use in source and binary forms, with or without diff --git a/nri-kafka.cabal b/nri-kafka.cabal --- a/nri-kafka.cabal +++ b/nri-kafka.cabal @@ -1,11 +1,11 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.34.4. +-- This file has been generated from package.yaml by hpack version 0.34.5. -- -- see: https://github.com/sol/hpack name: nri-kafka -version: 0.1.0.2 +version: 0.1.0.4 synopsis: Functions for working with Kafka description: Please see the README at . category: Web @@ -13,7 +13,7 @@ bug-reports: https://github.com/NoRedInk/haskell-libraries/issues author: NoRedInk maintainer: haskell-open-source@noredink.com -copyright: 2021 NoRedInk Corp. +copyright: 2022 NoRedInk Corp. license: BSD3 license-file: LICENSE build-type: Simple @@ -65,7 +65,7 @@ build-depends: aeson >=1.4.6.0 && <2.1 , async >=2.2.2 && <2.3 - , base >=4.12.0.0 && <4.16 + , base >=4.12.0.0 && <4.17 , bytestring >=0.10.8.2 && <0.12 , conduit >=1.3.0 && <1.4 , containers >=0.6.0.1 && <0.7 @@ -75,7 +75,7 @@ , nri-prelude >=0.1.0.0 && <0.7 , safe-exceptions >=0.1.7.0 && <1.3 , stm >=2.4 && <2.6 - , text >=1.2.3.1 && <1.3 + , text >=1.2.3.1 && <2.1 , time >=1.8.0.2 && <2 , unix >=2.7.2.2 && <2.8.0.0 , uuid >=1.3.0 && <1.4 @@ -124,7 +124,7 @@ build-depends: aeson >=1.4.6.0 && <2.1 , async >=2.2.2 && <2.3 - , base >=4.12.0.0 && <4.16 + , base >=4.12.0.0 && <4.17 , bytestring >=0.10.8.2 && <0.12 , conduit >=1.3.0 && <1.4 , containers >=0.6.0.1 && <0.7 @@ -134,7 +134,7 @@ , nri-prelude >=0.1.0.0 && <0.7 , safe-exceptions >=0.1.7.0 && <1.3 , stm >=2.4 && <2.6 - , text >=1.2.3.1 && <1.3 + , text >=1.2.3.1 && <2.1 , time >=1.8.0.2 && <2 , unix >=2.7.2.2 && <2.8.0.0 , uuid >=1.3.0 && <1.4 diff --git a/src/Kafka/Worker.hs b/src/Kafka/Worker.hs --- a/src/Kafka/Worker.hs +++ b/src/Kafka/Worker.hs @@ -26,6 +26,7 @@ Internal.subscriptionManageOwnOffsets, Internal.PartitionOffset (..), Partition.SeekCmd (..), + Internal.CommitToKafkaAsWell (..), ) where diff --git a/src/Kafka/Worker/Internal.hs b/src/Kafka/Worker/Internal.hs --- a/src/Kafka/Worker/Internal.hs +++ b/src/Kafka/Worker/Internal.hs @@ -52,9 +52,15 @@ data TopicSubscription = TopicSubscription { topic :: Kafka.Topic, onMessage :: Partition.MessageCallback, - offsetSource :: OffsetSource + offsetSource :: OffsetSource, + commitToKafkaAsWell :: CommitToKafkaAsWell } +-- | Commit the offset to Kafka in addition to an externally managed storage. +data CommitToKafkaAsWell + = CommitToKafkaAsWell + | DoNotCommitToKafkaAsWell + -- | Params needed to write / read offsets to another data store data PartitionOffset = PartitionOffset { -- | The partition of a topic. @@ -87,7 +93,8 @@ callback msg Task.succeed Partition.NoSeek ), - offsetSource = InKafka + offsetSource = InKafka, + commitToKafkaAsWell = CommitToKafkaAsWell } -- | Create a subscription for a topic and manage offsets for that topic @@ -102,6 +109,7 @@ -- > let subscription = -- > subscriptionManageOwnOffsets -- > "the-topic" +-- > CommitToKafkaAsWell -- > (\partitions -> -- > sql -- > "SELECT partition, offset FROM offsets WHERE partition = %" @@ -111,12 +119,14 @@ subscriptionManageOwnOffsets :: (Aeson.FromJSON msg, Aeson.ToJSON msg) => Text -> + CommitToKafkaAsWell -> ([Int] -> Task Text (List PartitionOffset)) -> (PartitionOffset -> msg -> Task Text Partition.SeekCmd) -> TopicSubscription -subscriptionManageOwnOffsets topic fetchOffsets callback = +subscriptionManageOwnOffsets topic commitToKafkaAsWell fetchOffsets callback = TopicSubscription { topic = Kafka.Topic topic, + commitToKafkaAsWell, onMessage = Partition.MessageCallback ( \record msg -> do @@ -190,12 +200,12 @@ -- the worker shares the OS process with other test code and the test runner. processWithoutShutdownEnsurance :: Settings.Settings -> Consumer.ConsumerGroupId -> TopicSubscription -> Prelude.IO () processWithoutShutdownEnsurance settings groupId topicSubscriptions = do - let TopicSubscription {onMessage, topic, offsetSource} = topicSubscriptions + let TopicSubscription {onMessage, topic, offsetSource, commitToKafkaAsWell} = topicSubscriptions state <- initState onQuitSignal (Stopping.stopTakingRequests (stopping state) "Received stop signal") Conduit.withAcquire (Observability.handler (Settings.observability settings)) <| \observabilityHandler -> do Exception.bracketWithError - (createConsumer settings groupId observabilityHandler offsetSource onMessage topic state) + (createConsumer settings groupId observabilityHandler offsetSource commitToKafkaAsWell onMessage topic state) (cleanUp observabilityHandler (rebalanceInfo state) (stopping state)) (runThreads settings state) @@ -220,6 +230,7 @@ Consumer.ConsumerGroupId -> Observability.Handler -> OffsetSource -> + CommitToKafkaAsWell -> Partition.MessageCallback -> Kafka.Topic -> State -> @@ -234,6 +245,7 @@ groupId observability offsetSource + commitToKafkaAsWell callback topic state = do @@ -243,6 +255,7 @@ observability callback offsetSource + commitToKafkaAsWell state let properties = Consumer.brokersList @@ -283,11 +296,12 @@ Observability.Handler -> Partition.MessageCallback -> OffsetSource -> + CommitToKafkaAsWell -> State -> Consumer.KafkaConsumer -> Consumer.RebalanceEvent -> Prelude.IO () -rebalanceCallback skipOrNot observability callback offsetSource state consumer rebalanceEvent = do +rebalanceCallback skipOrNot observability callback offsetSource commitToKafkaAsWell state consumer rebalanceEvent = do now <- GHC.Clock.getMonotonicTime Analytics.updateTimeOfLastRebalance now (analytics state) case rebalanceEvent of @@ -302,8 +316,13 @@ case fetchResult of Err err -> Exception.throwString (Text.toList err) Ok fetched -> do + let elsewhere = case commitToKafkaAsWell of + CommitToKafkaAsWell -> + Partition.ElsewhereButToKafkaAsWell + DoNotCommitToKafkaAsWell -> + Partition.Elsewhere let storedOffsets = - List.map (Tuple.mapSecond Partition.Elsewhere) fetched + List.map (Tuple.mapSecond elsewhere) fetched |> Dict.fromList let storedKeys = fetched diff --git a/src/Kafka/Worker/Partition.hs b/src/Kafka/Worker/Partition.hs --- a/src/Kafka/Worker/Partition.hs +++ b/src/Kafka/Worker/Partition.hs @@ -102,6 +102,10 @@ | -- | Commit offsets elsewhere. Takes the offset of the last committed -- message so we can skip old messages. Elsewhere Int + | -- | Same as `Elsewhere` but commits offsets to Kafka. This is an experiment + -- to check if brokers stay healthier when their offsets aren't completely + -- out of date. + ElsewhereButToKafkaAsWell Int -- | A thread that processes messages for a particular partition. Cleans itself -- up if it ever runs out. @@ -154,6 +158,7 @@ <| case commitOffsets of ToKafka -> Assigned Seq.empty Elsewhere offset -> AwaitingSeekTo offset + ElsewhereButToKafkaAsWell offset -> AwaitingSeekTo offset onStartup partition Exception.finally (processMsgLoop skipOrNot commitOffsets observabilityHandler State {analytics, stopping, partition} consumer callback) @@ -223,6 +228,16 @@ -- of a larger database transaction as part of an -- exactly-once-delivery scheme. Prelude.pure () + ElsewhereButToKafkaAsWell _ -> + -- The user of the Kafka module is responsible for + -- comitting the offsets. To help them do so we pass + -- them the record in the callback function. + -- + -- It's important the module user can determine when to + -- commit, for example to allow them to commit as part + -- of a larger database transaction as part of an + -- exactly-once-delivery scheme. + commitRecord doAnything consumer record -- finally, let's remove it from the queue dequeueRecord (partition state) record