freckle-app 1.13.0.1 → 1.14.0.0

raw patch · 4 files changed

+26/−44 lines, 4 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +5/−1
• freckle-app.cabal +1/−1
• library/Freckle/App/Kafka/Consumer.hs +19/−41
• package.yaml +1/−1

@@ -1,4 +1,8 @@-## [_Unreleased_](https://github.com/freckle/freckle-app/compare/v1.13.0.1...main)+## [_Unreleased_](https://github.com/freckle/freckle-app/compare/v1.14.0.0...main)++## [v1.14.0.0](https://github.com/freckle/freckle-app/compare/v1.13.0.1...v1.14.0.1)++- Reverted changes for explicit offset committing in `runConsumer` back to the auto-commit strategy [breaking]  ## [v1.13.0.1](https://github.com/freckle/freckle-app/compare/v1.13.0.0...v1.13.0.1) 

@@ -1,6 +1,6 @@ cabal-version:      1.18 name:               freckle-app-version:            1.13.0.1+version:            1.14.0.0 license:            MIT license-file:       LICENSE maintainer:         Freckle Education

@@ -21,10 +21,7 @@ import qualified Env import Freckle.App.Async import Freckle.App.Env-import Freckle.App.Exception-  ( AnnotatedException (..)-  , annotatedExceptionMessageFrom-  )+import Freckle.App.Exception (annotatedExceptionMessageFrom) import Freckle.App.Kafka.Producer (envKafkaBrokerAddresses) import Freckle.App.OpenTelemetry import Kafka.Consumer hiding@@ -53,6 +50,11 @@   -- ^ The offset reset parameter used when there is no initial offset in Kafka.   --   -- This is the `auto.offset.reset` Kafka consumer configuration property.+  , kafkaConsumerConfigAutoCommitInterval :: Millis+  -- ^ The interval that offsets are auto-committed to Kafka.+  --+  -- This sets the `auto.commit.interval.ms` and `enable.auto.commit` Kafka+  -- consumer configuration properties.   , kafkaConsumerConfigExtraSubscriptionProps :: Map Text Text   -- ^ Extra properties used to configure the Kafka consumer.   }@@ -92,6 +94,10 @@   consumerGroupId <- Env.var Env.nonempty "KAFKA_CONSUMER_GROUP_ID" mempty   kafkaTopic <- envKafkaTopic   kafkaOffsetReset <- envKafkaOffsetReset+  kafkaAutoOffsetInterval <-+    fromIntegral . timeoutMs <$$> Env.var timeout "KAFKA_AUTO_COMMIT_INTERVAL" $+      Env.def $+        TimeoutMilliseconds 5000   kafkaExtraProps <-     Env.var       (fmap Map.fromList . keyValues)@@ -103,6 +109,7 @@       consumerGroupId       kafkaTopic       kafkaOffsetReset+      kafkaAutoOffsetInterval       kafkaExtraProps  class HasKafkaConsumer env where@@ -112,8 +119,7 @@ consumerProps KafkaConsumerConfig {..} =   brokersList brokers     <> groupId kafkaConsumerConfigGroupId-    <> noAutoCommit -- we handle offsets-    <> noAutoOffsetStore+    <> autoCommit kafkaConsumerConfigAutoCommitInterval     <> logLevel KafkaLogInfo  where   brokers = NE.toList kafkaConsumerConfigBrokerAddresses@@ -168,26 +174,18 @@   -> (a -> m ())   -> m () runConsumer pollTimeout onMessage =-  withTraceIdContext $ immortalCreate onFinish $ forever $ do+  withTraceIdContext $ immortalCreateLogged $ forever $ do     consumer <- view kafkaConsumerL      flip catches handlers $ inSpan "kafka.consumer" consumerSpanArguments $ do       mRecord <- fromKafkaError =<< pollMessage consumer kTimeout -      for_ mRecord $ \r -> do-        for_ (crValue r) $ \bs -> do-          a <--            inSpan "kafka.consumer.message.decode" defaultSpanArguments $-              either (throwM . KafkaMessageDecodeError bs) pure $-                eitherDecodeStrict bs-          inSpan "kafka.consumer.message.handle" defaultSpanArguments $ onMessage a--        inSpan "kafka.consumer.message.commit" defaultSpanArguments $ do-          -- Store the offset of this record, then do a best-offort commit to-          -- the broker. If this fails and we crash or shutdown, the commit in-          -- the onFinish handler will pick it up.-          logExMay "Unable to store offset" $ storeOffsetMessage consumer r-          void $ commitAllOffsets OffsetCommitAsync consumer+      for_ (crValue =<< mRecord) $ \bs -> do+        a <-+          inSpan "kafka.consumer.message.decode" defaultSpanArguments $+            either (throwM . KafkaMessageDecodeError bs) pure $+              eitherDecodeStrict bs+        inSpan "kafka.consumer.message.handle" defaultSpanArguments $ onMessage a  where   kTimeout = Kafka.Timeout $ timeoutMs pollTimeout @@ -201,26 +199,6 @@           . annotatedExceptionMessageFrom @KafkaMessageDecodeError             (const "Could not decode message value")     ]--  onFinish finishResult = do-    consumer <- view kafkaConsumerL-    logExMay-      "Unable to commit offsets"-      (silenceNoOffsetError <$> commitAllOffsets OffsetCommit consumer)-    either (logEx "Unexpected finish") pure finishResult--  logExMay msg f = maybe (pure ()) (logEx msg) =<< f--  logEx msg =-    logErrorNS "kafka"-      . annotatedExceptionMessageFrom (const msg)-      . AnnotatedException []--  -- It should not be considered an error if we have no offsets to commit-  silenceNoOffsetError :: Maybe KafkaError -> Maybe KafkaError-  silenceNoOffsetError = \case-    Just (KafkaResponseError RdKafkaRespErrNoOffset) -> Nothing-    e -> e  fromKafkaError   :: (MonadIO m, MonadLogger m, HasCallStack)

@@ -1,5 +1,5 @@ name: freckle-app-version: 1.13.0.1+version: 1.14.0.0 maintainer: Freckle Education category: Utils github: freckle/freckle-app