nri-kafka 0.1.0.0 → 0.1.0.1
raw patch · 4 files changed
+28/−16 lines, 4 filesdep ~aesonPVP ok
version bump matches the API change (PVP)
Dependency ranges changed: aeson
API changes (from Hackage documentation)
Files
- CHANGELOG.md +4/−0
- nri-kafka.cabal +3/−3
- src/Kafka/Worker/Internal.hs +11/−8
- src/Kafka/Worker/Stopping.hs +10/−5
CHANGELOG.md view
@@ -1,3 +1,7 @@+# 0.1.0.1++- Support `aeson-2.0.x`.+ # 0.1.0.0 - First release, but we've battle-tested it against significant load for months now!
nri-kafka.cabal view
@@ -5,7 +5,7 @@ -- see: https://github.com/sol/hpack name: nri-kafka-version: 0.1.0.0+version: 0.1.0.1 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@@ -63,7 +63,7 @@ TypeOperators ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wpartial-fields -Wredundant-constraints -Wincomplete-uni-patterns -fno-warn-type-defaults -fplugin=NriPrelude.Plugin build-depends:- aeson >=1.4.6.0 && <1.6+ aeson >=1.4.6.0 && <2.1 , async >=2.2.2 && <2.3 , base >=4.12.0.0 && <4.16 , bytestring >=0.10.8.2 && <0.12@@ -122,7 +122,7 @@ TypeOperators ghc-options: -Wall -Wcompat -Widentities -Wincomplete-record-updates -Wpartial-fields -Wredundant-constraints -Wincomplete-uni-patterns -fno-warn-type-defaults -fplugin=NriPrelude.Plugin -threaded -rtsopts "-with-rtsopts=-N -T" -fno-warn-type-defaults build-depends:- aeson >=1.4.6.0 && <1.6+ aeson >=1.4.6.0 && <2.1 , async >=2.2.2 && <2.3 , base >=4.12.0.0 && <4.16 , bytestring >=0.10.8.2 && <0.12
src/Kafka/Worker/Internal.hs view
@@ -192,7 +192,7 @@ processWithoutShutdownEnsurance settings groupId topicSubscriptions = do let TopicSubscription {onMessage, topic, offsetSource} = topicSubscriptions state <- initState- onQuitSignal (Stopping.stopTakingRequests (stopping state))+ 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)@@ -408,8 +408,10 @@ cleanUp observabilityHandler rebalanceInfo stopping maybeException consumer = do Prelude.putStrLn "Cleaning up" _ <- Consumer.closeConsumer consumer- -- ensure workers shut down- Stopping.stopTakingRequests stopping+ -- In case we're already stopping, get the reason we're doing so.+ maybeStopReason <- Stopping.stopReason stopping+ -- Ensure we enter stopping mode if we weren't already.+ Stopping.stopTakingRequests stopping "Shutting down" requestId <- map Data.UUID.toText Data.UUID.V4.nextRandom -- at some point, k8s should report system crashes. In the mean time, we'll do it. Platform.rootTracingSpanIO@@ -428,9 +430,11 @@ Log.context "rebalance info" (Debug.toString rebalanceInfo') ] |> Task.perform log- writeCrashLogOnError maybeException- Prelude.putStrLn "Bye!"+ case (maybeException, maybeStopReason) of+ (Just exception, _) -> Prelude.putStrLn ("Shut down because of exception: " ++ Exception.displayException exception)+ (_, Just stopReason) -> Prelude.putStrLn ("Shut down because of: " ++ Text.toList stopReason)+ (Nothing, Nothing) -> Prelude.putStrLn "Shut down for an unknown reason." -- | Handle crash logging writeCrashLogOnError :: Maybe Exception.SomeException -> Prelude.IO ()@@ -514,8 +518,7 @@ ) data RuntimeExceptions- = ReceivedMsgNotInAssignedPartitions (Consumer.TopicName, Consumer.PartitionId)- | AskedToInitPartitionThatAlreadyExists (Consumer.TopicName, Consumer.PartitionId)+ = AskedToInitPartitionThatAlreadyExists (Consumer.TopicName, Consumer.PartitionId) deriving (Show) instance Exception.Exception RuntimeExceptions@@ -530,7 +533,7 @@ partitions' <- TVar.readTVar partitions let maybePartition = Dict.get key partitions' case maybePartition of- Nothing -> STM.throwSTM (ReceivedMsgNotInAssignedPartitions key)+ Nothing -> Prelude.pure Partition.NoSeek Just partition -> Partition.append record partition -- | Intermittently updates
src/Kafka/Worker/Stopping.hs view
@@ -1,5 +1,6 @@ module Kafka.Worker.Stopping ( init,+ stopReason, stopTakingRequests, runUnlessStopping, Stopping,@@ -10,21 +11,25 @@ import qualified Control.Concurrent.MVar as MVar import qualified Prelude -newtype Stopping = Stopping (MVar.MVar ())+newtype Stopping = Stopping (MVar.MVar Text) init :: Prelude.IO Stopping init = MVar.newEmptyMVar |> map Stopping -stopTakingRequests :: Stopping -> Prelude.IO ()-stopTakingRequests (Stopping stopping) = do+stopReason :: Stopping -> Prelude.IO (Maybe Text)+stopReason (Stopping stopping) =+ MVar.tryReadMVar stopping++stopTakingRequests :: Stopping -> Text -> Prelude.IO ()+stopTakingRequests (Stopping stopping) reason = do Prelude.putStrLn "Gracefully shutting down..."- MVar.tryPutMVar stopping ()+ MVar.tryPutMVar stopping reason |> map (\_ -> ()) runUnlessStopping :: Stopping -> a -> Prelude.IO a -> Prelude.IO a runUnlessStopping (Stopping stopping) stoppingVal action = Async.race- (MVar.readMVar stopping)+ (MVar.readMVar stopping |> map (\_ -> ())) action |> map ( \either ->