nri-kafka 0.1.0.2 → 0.1.0.4

raw patch · 6 files changed

+58/−15 lines, 6 filesdep ~basedep ~textnew-uploaderPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: base, text

API changes (from Hackage documentation)

Files

• CHANGELOG.md +8/−0
• LICENSE +1/−1
• nri-kafka.cabal +7/−7
• src/Kafka/Worker.hs +1/−0
• src/Kafka/Worker/Internal.hs +26/−7
• src/Kafka/Worker/Partition.hs +15/−0

@@ -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`.

@@ -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

@@ -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 <https://github.com/NoRedInk/haskell-libraries/tree/trunk/nri-kafka#readme>. 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

@@ -26,6 +26,7 @@     Internal.subscriptionManageOwnOffsets,     Internal.PartitionOffset (..),     Partition.SeekCmd (..),+    Internal.CommitToKafkaAsWell (..),   ) where 

@@ -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

@@ -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