freckle-app 1.9.2.1 → 1.9.3.0
raw patch · 4 files changed
+20/−11 lines, 4 files
Files
- CHANGELOG.md +4/−0
- freckle-app.cabal +1/−1
- library/Freckle/App/Kafka/Consumer.hs +14/−9
- package.yaml +1/−1
CHANGELOG.md view
@@ -1,5 +1,9 @@ ## [_Unreleased_](https://github.com/freckle/freckle-app/compare/v1.9.1.1...main) +## [v1.9.3.0](https://github.com/freckle/freckle-app/compare/v1.9.2.1...v1.9.3.0)++- Auto-commit offset when consuming Kafka events.+ ## [v1.9.2.1](https://github.com/freckle/freckle-app/compare/v1.9.2.0...v1.9.2.1) - Fixed segfault bug in `Freckle.App.Kafka.Consumer`.
freckle-app.cabal view
@@ -1,6 +1,6 @@ cabal-version: 1.18 name: freckle-app-version: 1.9.2.1+version: 1.9.3.0 license: MIT license-file: LICENSE maintainer: Freckle Education
library/Freckle/App/Kafka/Consumer.hs view
@@ -48,6 +48,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. }@@ -87,6 +92,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)@@ -98,6 +107,7 @@ consumerGroupId kafkaTopic kafkaOffsetReset+ kafkaAutoOffsetInterval kafkaExtraProps class HasKafkaConsumer env where@@ -107,7 +117,7 @@ consumerProps KafkaConsumerConfig {..} = brokersList brokers <> groupId kafkaConsumerConfigGroupId- <> noAutoCommit+ <> autoCommit kafkaConsumerConfigAutoCommitInterval <> logLevel KafkaLogInfo where brokers = NE.toList kafkaConsumerConfigBrokerAddresses@@ -150,17 +160,12 @@ eMessage <- pollMessage consumer $ Kafka.Timeout $ timeoutMs pollTimeout case eMessage of- Left (KafkaResponseError RdKafkaRespErrTimedOut) -> logDebug $ "Polling timeout"+ Left (KafkaResponseError RdKafkaRespErrTimedOut) -> logDebug "Polling timeout" Left err -> logError $ "Error polling for message from Kafka" :# ["error" .= show err]- Right m@(ConsumerRecord {..}) -> for_ crValue $ \bs -> do+ Right ConsumerRecord {..} -> for_ crValue $ \bs -> case eitherDecodeStrict bs of Left err -> logError $ "Could not decode message value" :# ["error" .= err]- Right a -> do- onMessage a- maybe- (logInfo "Committed offsets")- (\err -> logError $ "Error committing offsets" :# ["error" .= show err])- =<< commitOffsetMessage OffsetCommit consumer m+ Right a -> onMessage a forever $ liftIO $ threadDelay maxBound where handleException = \case
package.yaml view
@@ -1,5 +1,5 @@ name: freckle-app-version: 1.9.2.1+version: 1.9.3.0 maintainer: Freckle Education category: Utils github: freckle/freckle-app