From 0d7a5680703677e8173e66c5bf2935d99fbcc739 Mon Sep 17 00:00:00 2001 From: Vladislav Filatov Date: Wed, 10 Dec 2025 15:48:31 +0100 Subject: [PATCH 1/2] Fine-grained logging for PartitionFlow owned classes. --- .../kafka/flow/KeyContext.scala | 15 +- .../kafka/flow/KeyFlowOf.scala | 21 +- .../kafka/flow/KeyStateOf.scala | 11 +- .../kafka/flow/PartitionFlow.scala | 7 +- .../kafka/flow/timer/TimerFlowOf.scala | 276 ++++++++++-------- .../kafka/flow/FoldToStateSpec.scala | 1 + .../kafka/flow/KeyFlowSpec.scala | 8 +- .../kafka/flow/PartitionFlowSpec.scala | 2 +- .../kafka/flow/timer/TimerFlowOfSpec.scala | 7 +- .../kafka/flow/KeyStateMetrics.scala | 4 +- 10 files changed, 202 insertions(+), 150 deletions(-) diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyContext.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyContext.scala index 379ddb19d..14d0b1d10 100644 --- a/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyContext.scala +++ b/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyContext.scala @@ -17,6 +17,7 @@ trait KeyContext[F[_]] { def hold(offset: Offset): F[Unit] def remove: F[Unit] def log: Log[F] + def key: String } object KeyContext { @@ -27,29 +28,33 @@ object KeyContext { def holding = none[Offset].pure[F] def hold(offset: Offset) = ().pure[F] def remove = ().pure[F] + val key = "" } - def of[F[_]: Ref.Make: Monad: Log](removeFromCache: F[Unit]): F[KeyContext[F]] = + def of[F[_]: Ref.Make: Monad: Log](removeFromCache: F[Unit], key: String): F[KeyContext[F]] = Ref.of[F, Option[Offset]](None) map { storage => - KeyContext(storage.stateInstance, removeFromCache) + KeyContext(storage.stateInstance, removeFromCache, key) } def apply[F[_]: Monad: Log]( storage: Stateful[F, Option[Offset]], - removeFromCache: F[Unit] + removeFromCache: F[Unit], + _key: String ): KeyContext[F] = new KeyContext[F] { def holding = storage.get def hold(offset: Offset) = storage.set(Some(offset)) def remove = storage.set(None) *> removeFromCache def log = Log[F] + val key = _key } def resource[F[_]: Ref.Make: Monad]( removeFromCache: F[Unit], - log: Log[F] + log: Log[F], + key: String ): Resource[F, KeyContext[F]] = { implicit val _log = log - Resource.eval(of(removeFromCache)) + Resource.eval(of(removeFromCache, key)) } } diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyFlowOf.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyFlowOf.scala index ddee01535..74b6c6a0a 100644 --- a/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyFlowOf.scala +++ b/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyFlowOf.scala @@ -2,6 +2,7 @@ package com.evolutiongaming.kafka.flow import cats.Monad import cats.effect.{Ref, Resource} +import com.evolutiongaming.catshelper.LogOf import com.evolutiongaming.kafka.flow.persistence.Persistence import com.evolutiongaming.kafka.flow.registry.EntityRegistry import com.evolutiongaming.kafka.flow.timer.{TimerContext, TimerFlowOf} @@ -15,7 +16,7 @@ trait KeyFlowOf[F[_], S, A] { timers: TimerContext[F], additionalPersist: AdditionalStatePersist[F, S, A], registry: EntityRegistry[F, KafkaKey, S], - ): Resource[F, KeyFlow[F, A]] + )(implicit logOf: LogOf[F]): Resource[F, KeyFlow[F, A]] } object KeyFlowOf { @@ -50,11 +51,19 @@ object KeyFlowOf { timerFlowOf: TimerFlowOf[F], fold: EnhancedFold[F, S, A], tick: TickOption[F, S], - ): KeyFlowOf[F, S, A] = { (key, context, persistence, timers, additionalPersist, registry) => - implicit val _context = context - timerFlowOf(context, persistence, timers) flatMap { timerFlow => - KeyFlow.of(key, fold, tick, persistence, additionalPersist, timerFlow, registry) + ): KeyFlowOf[F, S, A] = new KeyFlowOf[F, S, A] { + override def apply( + key: KafkaKey, + context: KeyContext[F], + persistence: Persistence[F, S, A], + timers: TimerContext[F], + additionalPersist: AdditionalStatePersist[F, S, A], + registry: EntityRegistry[F, KafkaKey, S] + )(implicit logOf: LogOf[F]): Resource[F, KeyFlow[F, A]] = { + implicit val _context = context + timerFlowOf(context, persistence, timers) flatMap { timerFlow => + KeyFlow.of(key, fold, tick, persistence, additionalPersist, timerFlow, registry) + } } } - } diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyStateOf.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyStateOf.scala index 512be80c7..240a39191 100644 --- a/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyStateOf.scala +++ b/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyStateOf.scala @@ -3,6 +3,7 @@ package com.evolutiongaming.kafka.flow import cats.Applicative import cats.effect.{Resource, Sync} import cats.syntax.all.* +import com.evolutiongaming.catshelper.LogOf import com.evolutiongaming.kafka.flow.key.KeysOf import com.evolutiongaming.kafka.flow.persistence.{PersistenceOf, SnapshotPersistenceOf} import com.evolutiongaming.kafka.flow.registry.EntityRegistry @@ -20,7 +21,7 @@ trait KeyStateOf[F[_]] { self => key: String, createdAt: Timestamp, context: KeyContext[F] - ): Resource[F, KeyState[F, ConsumerRecord[String, ByteVector]]] + )(implicit logOf: LogOf[F]): Resource[F, KeyState[F, ConsumerRecord[String, ByteVector]]] /** Restores a state for all keys present in persistence. * @@ -71,7 +72,9 @@ object KeyStateOf { registry: EntityRegistry[F, KafkaKey, S], ): KeyStateOf[F] = new KeyStateOf[F] { - def apply(topicPartition: TopicPartition, key: String, createdAt: Timestamp, context: KeyContext[F]) = { + def apply(topicPartition: TopicPartition, key: String, createdAt: Timestamp, context: KeyContext[F])( + implicit logOf: LogOf[F] + ) = { implicit val _context = context val kafkaKey = KafkaKey( applicationId = applicationId, @@ -217,7 +220,9 @@ object KeyStateOf { registry: EntityRegistry[F, KafkaKey, S], ): KeyStateOf[F] = new KeyStateOf[F] { - def apply(topicPartition: TopicPartition, key: String, createdAt: Timestamp, context: KeyContext[F]) = { + def apply(topicPartition: TopicPartition, key: String, createdAt: Timestamp, context: KeyContext[F])( + implicit logOf: LogOf[F] + ) = { val kafkaKey = KafkaKey( applicationId = applicationId, groupId = groupId, diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/PartitionFlow.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/PartitionFlow.scala index 3ecf0ddc4..2660e0382 100644 --- a/core/src/main/scala/com/evolutiongaming/kafka/flow/PartitionFlow.scala +++ b/core/src/main/scala/com/evolutiongaming/kafka/flow/PartitionFlow.scala @@ -66,7 +66,7 @@ object PartitionFlow { } } - def of[F[_]: Async]( + def of[F[_]: Async: LogOf]( topicPartition: TopicPartition, assignedAt: Offset, keyStateOf: KeyStateOf[F], @@ -97,7 +97,7 @@ object PartitionFlow { } yield flow // TODO: put most `Ref` variables into one state class? - def of[F[_]: Async]( + def of[F[_]: Async: LogOf]( topicPartition: TopicPartition, keyStateOf: KeyStateOf[F], committedOffset: Ref[F, Offset], @@ -116,7 +116,8 @@ object PartitionFlow { for { context <- KeyContext.resource[F]( removeFromCache = cache.remove(key).flatten.void, - log = log.prefixed(key) + log = log.prefixed(key), + key = key ) keyState <- keyStateOf(topicPartition, key, createdAt, context) } yield PartitionKey(keyState, context) diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOf.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOf.scala index 51c08125c..d63ecfabf 100644 --- a/core/src/main/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOf.scala +++ b/core/src/main/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOf.scala @@ -2,8 +2,11 @@ package com.evolutiongaming.kafka.flow.timer import cats.{Applicative, Monad, MonadThrow} import cats.effect.Resource +import cats.effect.syntax.all.* import cats.effect.kernel.Resource.ExitCase import cats.syntax.all.* +import com.evolutiongaming.catshelper.Log.Mdc +import com.evolutiongaming.catshelper.{Log, LogOf} import com.evolutiongaming.kafka.flow.KeyContext import com.evolutiongaming.kafka.flow.persistence.FlushBuffers import com.evolutiongaming.skafka.Offset @@ -16,8 +19,7 @@ trait TimerFlowOf[F[_]] { context: KeyContext[F], persistence: FlushBuffers[F], timers: TimerContext[F] - ): Resource[F, TimerFlow[F]] - + )(implicit logOf: LogOf[F]): Resource[F, TimerFlow[F]] } object TimerFlowOf { @@ -38,43 +40,48 @@ object TimerFlowOf { maxOffsetDifference: Int = 100000, maxIdle: FiniteDuration = 10.minutes, flushOnRevoke: Boolean = false, - ): TimerFlowOf[F] = { (context, persistence, timers) => - def register(touchedAt: Timestamp) = - timers.registerProcessing(touchedAt.clock plusMillis fireEvery.toMillis) - - val acquire = Resource.eval { - for { - current <- timers.current - persistedAt <- timers.persistedAt - committedAt = persistedAt getOrElse current - _ <- context.hold(committedAt.offset) - _ <- register(committedAt) - } yield new TimerFlow[F] { - def onTimer = for { - current <- timers.current - processedAt <- timers.processedAt - touchedAt = processedAt getOrElse committedAt - expiredAt = touchedAt.clock plusMillis maxIdle.toMillis - expired = current.clock isAfter expiredAt - offsetDifference = current.offset.value - touchedAt.offset.value - canUnload = expired || offsetDifference > maxOffsetDifference - _ <- - if (canUnload) { - context.log.info(s"flush, offset difference: $offsetDifference") *> - persistence.flush *> - context.remove - } else { - register(touchedAt) - } - } yield () - } - } + ): TimerFlowOf[F] = + new TimerFlowOf[F] { + override def apply(context: KeyContext[F], persistence: FlushBuffers[F], timers: TimerContext[F])( + implicit logOf: LogOf[F] + ): Resource[F, TimerFlow[F]] = { + def register(touchedAt: Timestamp): F[Unit] = + timers.registerProcessing(touchedAt.clock plusMillis fireEvery.toMillis) - val cancel = flushOnCancel.apply(context, persistence, timers) + val acquire = Resource.eval { + for { + log <- logOf(classOf[TimerFlowOf[F]]).map(_.withMdc(Mdc.Eager("key" -> context.key))) + current <- timers.current + persistedAt <- timers.persistedAt + committedAt = persistedAt getOrElse current + _ <- context.hold(committedAt.offset) + _ <- register(committedAt) + } yield new TimerFlow[F] { + def onTimer: F[Unit] = for { + current <- timers.current + processedAt <- timers.processedAt + touchedAt = processedAt getOrElse committedAt + expiredAt = touchedAt.clock plusMillis maxIdle.toMillis + expired = current.clock isAfter expiredAt + offsetDifference = current.offset.value - touchedAt.offset.value + canUnload = expired || offsetDifference > maxOffsetDifference + _ <- + if (canUnload) { + log.info(s"flush, offset difference: $offsetDifference") *> + persistence.flush *> + context.remove + } else { + register(touchedAt) + } + } yield () + } + } - if (flushOnRevoke) acquire <* cancel else acquire + val cancel = flushOnCancel.apply(context, persistence, timers) - } + if (flushOnRevoke) acquire <* cancel else acquire + } + } /** Performs flush periodically. * @@ -102,40 +109,45 @@ object TimerFlowOf { persistEvery: FiniteDuration = 1.minute, flushOnRevoke: Boolean = false, ignorePersistErrors: Boolean = false, - ): TimerFlowOf[F] = { (context, persistence, timers) => - def register(current: Timestamp): F[Unit] = - timers.registerProcessing(current.clock plusMillis fireEvery.toMillis) - - val acquire = Resource.eval { - for { - current <- timers.current - persistedAt <- timers.persistedAt - committedAt = persistedAt getOrElse current - _ <- context.hold(committedAt.offset) - _ <- register(current) - } yield new TimerFlow[F] { - def onTimer: F[Unit] = for { - current <- timers.current - persistedAt <- timers.persistedAt - flushedAt = persistedAt getOrElse committedAt - triggerFlushAt = flushedAt.clock plusMillis persistEvery.toMillis - canPersist = (current.clock compareTo triggerFlushAt) >= 0 - _ <- MonadThrow[F].whenA(canPersist)( - persistence.attemptToPersist( - ignorePersistErrors = ignorePersistErrors, - context = context, - currentOffset = current.offset - ) - ) - _ <- register(current) - } yield () - } - } + ): TimerFlowOf[F] = new TimerFlowOf[F] { + override def apply(context: KeyContext[F], persistence: FlushBuffers[F], timers: TimerContext[F])( + implicit logOf: LogOf[F] + ): Resource[F, TimerFlow[F]] = { + def register(current: Timestamp): F[Unit] = + timers.registerProcessing(current.clock plusMillis fireEvery.toMillis) - val cancel = flushOnCancel.apply(context, persistence, timers) + val acquire = Resource.eval { + for { + log <- logOf(classOf[TimerFlowOf[F]]).map(_.withMdc(Mdc.Eager("key" -> context.key))) + current <- timers.current + persistedAt <- timers.persistedAt + committedAt = persistedAt getOrElse current + _ <- context.hold(committedAt.offset) + _ <- register(current) + } yield new TimerFlow[F] { + def onTimer: F[Unit] = for { + current <- timers.current + persistedAt <- timers.persistedAt + flushedAt = persistedAt getOrElse committedAt + triggerFlushAt = flushedAt.clock plusMillis persistEvery.toMillis + canPersist = (current.clock compareTo triggerFlushAt) >= 0 + _ <- MonadThrow[F] + .whenA(canPersist)( + persistence.attemptToPersist( + ignorePersistErrors = ignorePersistErrors, + context = context, + currentOffset = current.offset + )(log) + ) + _ <- register(current) + } yield () + } + } - if (flushOnRevoke) acquire <* cancel else acquire + val cancel = flushOnCancel.apply(context, persistence, timers) + if (flushOnRevoke) acquire <* cancel else acquire + } } /** Combines [[unloadOrphaned]] with [[persistPeriodically]] in a single TimerFlow @@ -162,83 +174,93 @@ object TimerFlowOf { maxIdle: FiniteDuration = 10.minutes, flushOnRevoke: Boolean = false, ignorePersistErrors: Boolean = false, - ): TimerFlowOf[F] = (context, persistence, timers) => { - def register(touchedAt: Timestamp): F[Unit] = - timers.registerProcessing(touchedAt.clock plusMillis fireEvery.toMillis) - - val acquire: Resource[F, TimerFlow[F]] = Resource.eval { - for { - current <- timers.current - persistedAt <- timers.persistedAt - committedAt = persistedAt getOrElse current - _ <- context.hold(committedAt.offset) - _ <- register(committedAt) - } yield new TimerFlow[F] { - def onTimer: F[Unit] = for { - current <- timers.current - processedAt <- timers.processedAt - touchedAt = processedAt getOrElse committedAt - expiredAt = touchedAt.clock plusMillis maxIdle.toMillis - offsetDifference = current.offset.value - touchedAt.offset.value - flushedAt = persistedAt getOrElse committedAt - triggerFlushAt = flushedAt.clock plusMillis persistEvery.toMillis - expired = current.clock isAfter expiredAt - canUnload = expired || offsetDifference > maxOffsetDifference - canPersist = (current.clock compareTo triggerFlushAt) >= 0 - _ <- Applicative[F].whenA(canPersist || canUnload)( - persistence.attemptToPersist( - ignorePersistErrors = ignorePersistErrors, - context = context, - currentOffset = current.offset + ): TimerFlowOf[F] = new TimerFlowOf[F] { + override def apply(context: KeyContext[F], persistence: FlushBuffers[F], timers: TimerContext[F])( + implicit logOf: LogOf[F] + ): Resource[F, TimerFlow[F]] = { + def register(touchedAt: Timestamp): F[Unit] = + timers.registerProcessing(touchedAt.clock plusMillis fireEvery.toMillis) + + val acquire: Resource[F, TimerFlow[F]] = Resource.eval { + for { + log <- logOf(classOf[TimerFlowOf[F]]).map(_.withMdc(Mdc.Eager("key" -> context.key))) + current <- timers.current + persistedAt <- timers.persistedAt + committedAt = persistedAt getOrElse current + _ <- context.hold(committedAt.offset) + _ <- register(committedAt) + } yield new TimerFlow[F] { + def onTimer: F[Unit] = for { + current <- timers.current + processedAt <- timers.processedAt + touchedAt = processedAt getOrElse committedAt + expiredAt = touchedAt.clock plusMillis maxIdle.toMillis + offsetDifference = current.offset.value - touchedAt.offset.value + flushedAt = persistedAt getOrElse committedAt + triggerFlushAt = flushedAt.clock plusMillis persistEvery.toMillis + expired = current.clock isAfter expiredAt + canUnload = expired || offsetDifference > maxOffsetDifference + canPersist = (current.clock compareTo triggerFlushAt) >= 0 + _ <- Applicative[F].whenA(canPersist || canUnload)( + persistence.attemptToPersist( + ignorePersistErrors = ignorePersistErrors, + context = context, + currentOffset = current.offset + )(log) + ) + _ <- Applicative[F].whenA(canUnload)( + log.info(s"flush, offset difference: $offsetDifference") *> context.remove ) - ) - _ <- Applicative[F].whenA(canUnload)( - context.log.info(s"flush, offset difference: $offsetDifference") *> context.remove - ) - _ <- register(current) - } yield () + _ <- register(current) + } yield () + } } - } - val cancel = flushOnCancel.apply(context, persistence, timers) + val cancel = flushOnCancel.apply(context, persistence, timers) - if (flushOnRevoke) acquire <* cancel else acquire + if (flushOnRevoke) acquire <* cancel else acquire + } } /** Performs flush when `Resource` is cancelled only */ - def flushOnCancel[F[_]: Monad]: TimerFlowOf[F] = { (context, persistence, _) => - val cancel = context.holding flatMap { holding => - Applicative[F].whenA(holding.isDefined) { - context.log.info(s"flush on revoke, holding offset: $holding") *> - persistence.flush *> - context.remove - } - } + def flushOnCancel[F[_]: Monad]: TimerFlowOf[F] = new TimerFlowOf[F] { + override def apply(context: KeyContext[F], persistence: FlushBuffers[F], timers: TimerContext[F])( + implicit logOf: LogOf[F] + ): Resource[F, TimerFlow[F]] = + logOf(classOf[TimerFlowOf[F]]).toResource.map(_.withMdc(Mdc.Eager("key" -> context.key))).flatMap { log => + val cancel = context.holding flatMap { holding => + Applicative[F].whenA(holding.isDefined) { + log.info(s"flush on revoke, holding offset: $holding") *> + persistence.flush *> + context.remove + } + } - Resource.makeCase(TimerFlow.empty.pure) { - case (_, ExitCase.Succeeded) => - cancel - case (_, ExitCase.Canceled) => - cancel - // there is no point to try flushing if it failed with an error - // the state might not be consistend and storage not accessible - // plus this is a concurrent operation, and we do not want anything - // to happen concurrently for a specific key - case (_, _) => ().pure[F] - } + Resource.makeCase(TimerFlow.empty.pure) { + case (_, ExitCase.Succeeded) => + cancel + case (_, ExitCase.Canceled) => + cancel + // there is no point to try flushing if it failed with an error + // the state might not be consistend and storage not accessible + // plus this is a concurrent operation, and we do not want anything + // to happen concurrently for a specific key + case (_, _) => ().pure[F] + } + } } private implicit class AttemptToPersist[F[_]: MonadThrow](persistence: FlushBuffers[F]) { - def attemptToPersist(ignorePersistErrors: Boolean, context: KeyContext[F], currentOffset: Offset): F[Unit] = + def attemptToPersist(ignorePersistErrors: Boolean, context: KeyContext[F], currentOffset: Offset)( + log: Log[F] + ): F[Unit] = persistence.flush.attempt.flatMap { case Left(err) if ignorePersistErrors => // 'context' will continue holding the previous offset from the last time the state was persisted // and offsets committed (or just the last committed offset if no state has ever been persisted before). // Thus, when calculating the next offset to commit in `PartitionFlow#offsetToCommit` it will take // the minimal one (previous) and won't commit any offsets - context - .log - .info(s"Failed to persist state, the error is ignored and offsets won't be committed, error: $err") + log.info(s"Failed to persist state, the error is ignored and offsets won't be committed, error: $err") case Left(err) => err.raiseError[F, Unit] case Right(_) => context.hold(currentOffset) } diff --git a/core/src/test/scala/com/evolutiongaming/kafka/flow/FoldToStateSpec.scala b/core/src/test/scala/com/evolutiongaming/kafka/flow/FoldToStateSpec.scala index 1d57d49e7..1ee2b12be 100644 --- a/core/src/test/scala/com/evolutiongaming/kafka/flow/FoldToStateSpec.scala +++ b/core/src/test/scala/com/evolutiongaming/kafka/flow/FoldToStateSpec.scala @@ -135,6 +135,7 @@ object FoldToStateSpec { context.copy(removeCalled = context.removeCalled + 1) } def log = Log.empty + def key = "" } } diff --git a/core/src/test/scala/com/evolutiongaming/kafka/flow/KeyFlowSpec.scala b/core/src/test/scala/com/evolutiongaming/kafka/flow/KeyFlowSpec.scala index 2f207f537..32df37f79 100644 --- a/core/src/test/scala/com/evolutiongaming/kafka/flow/KeyFlowSpec.scala +++ b/core/src/test/scala/com/evolutiongaming/kafka/flow/KeyFlowSpec.scala @@ -4,7 +4,7 @@ import cats.data.NonEmptyList import cats.effect.syntax.resource.* import cats.effect.{Ref, SyncIO} import cats.syntax.all.* -import com.evolutiongaming.catshelper.Log +import com.evolutiongaming.catshelper.{Log, LogOf} import com.evolutiongaming.kafka.flow.KeyFlowSpec.* import com.evolutiongaming.kafka.flow.kafka.ToOffset import com.evolutiongaming.kafka.flow.persistence.Persistence @@ -19,6 +19,8 @@ import java.time.Instant class KeyFlowSpec extends FunSuite { + private implicit val log: LogOf[SyncIO] = LogOf.empty + test("KeyFlow processes messages correctly") { val f = new ConstFixture @@ -89,6 +91,7 @@ class KeyFlowSpec extends FunSuite { def hold(offset: Offset) = SyncIO.unit def remove = removeCalled.set(true) def log = Log.empty + def key = "" } val key = KafkaKey(applicationId = "test", groupId = "test", topicPartition = TopicPartition.empty, key = "key") val keyFlow = timerFlowOf(context, persistence, timers).flatMap(tf => @@ -139,6 +142,7 @@ class KeyFlowSpec extends FunSuite { def hold(offset: Offset) = SyncIO.unit def remove = removeCalled.set(true) def log = Log.empty + def key = "" } val key = KafkaKey(applicationId = "test", groupId = "test", topicPartition = TopicPartition.empty, key = "key") @@ -294,7 +298,7 @@ object KeyFlowSpec { implicit val log: Log[SyncIO] = Log.empty implicit val context: KeyContext[SyncIO] = - KeyContext.of(().pure[SyncIO]).unsafeRunSync() + KeyContext.of(().pure[SyncIO], "").unsafeRunSync() implicit val stateToOffset: ToOffset[State] = { case (offset, _) => diff --git a/core/src/test/scala/com/evolutiongaming/kafka/flow/PartitionFlowSpec.scala b/core/src/test/scala/com/evolutiongaming/kafka/flow/PartitionFlowSpec.scala index fa19a1ce8..3e0bb9c3c 100644 --- a/core/src/test/scala/com/evolutiongaming/kafka/flow/PartitionFlowSpec.scala +++ b/core/src/test/scala/com/evolutiongaming/kafka/flow/PartitionFlowSpec.scala @@ -508,7 +508,7 @@ object PartitionFlowSpec { key: String, createdAt: Timestamp, context: KeyContext[IO] - ): Resource[IO, KeyState[IO, ConsumerRecord[String, ByteVector]]] = { + )(implicit logOf: LogOf[IO]): Resource[IO, KeyState[IO, ConsumerRecord[String, ByteVector]]] = { implicit val _context = context val fold0 = fold val kafkaKey = KafkaKey("test", "test", topicPartition, key) diff --git a/core/src/test/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOfSpec.scala b/core/src/test/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOfSpec.scala index 338c8348a..21a2573b5 100644 --- a/core/src/test/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOfSpec.scala +++ b/core/src/test/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOfSpec.scala @@ -4,7 +4,7 @@ import cats.effect.{IO, Resource} import cats.effect.kernel.Ref import cats.effect.unsafe.implicits.global import cats.syntax.all.* -import com.evolutiongaming.catshelper.Log +import com.evolutiongaming.catshelper.{Log, LogOf} import com.evolutiongaming.kafka.flow.KeyContext import com.evolutiongaming.kafka.flow.MonadStateHelper.* import com.evolutiongaming.kafka.flow.persistence.FlushBuffers @@ -20,6 +20,8 @@ import scala.concurrent.duration.* class TimerFlowOfSpec extends FunSuite { + private implicit val log: LogOf[IO] = LogOf.empty + test("unloadOrphaned holds commits when started") { val f = new ConstFixture @@ -600,7 +602,8 @@ object TimerFlowSpec { implicit val keyContext: KeyContext[IO] = KeyContext( storage = contextRef.stateInstance.focus(Context.lens(_.holding)), - removeFromCache = contextRef.update(ctx => ctx.copy(removed = ctx.removed + 1)) + removeFromCache = contextRef.update(ctx => ctx.copy(removed = ctx.removed + 1)), + _key = "test-key" ) implicit val timerContext: TimerContext[IO] = { diff --git a/metrics/src/main/scala/com/evolutiongaming/kafka/flow/KeyStateMetrics.scala b/metrics/src/main/scala/com/evolutiongaming/kafka/flow/KeyStateMetrics.scala index 11f666139..7509084b8 100644 --- a/metrics/src/main/scala/com/evolutiongaming/kafka/flow/KeyStateMetrics.scala +++ b/metrics/src/main/scala/com/evolutiongaming/kafka/flow/KeyStateMetrics.scala @@ -3,6 +3,7 @@ package com.evolutiongaming.kafka.flow import cats.Monad import cats.effect.Resource import cats.syntax.all.* +import com.evolutiongaming.catshelper.LogOf import com.evolutiongaming.kafka.flow.metrics.MetricsOf import com.evolutiongaming.kafka.flow.timer.Timestamp import com.evolutiongaming.skafka.TopicPartition @@ -28,7 +29,8 @@ object KeyStateMetrics { key: String, createdAt: Timestamp, context: KeyContext[F] - ) = count(topicPartition.topic) *> keyStateOf(topicPartition, key, createdAt, context) + )(implicit logOf: LogOf[F]) = + count(topicPartition.topic) *> keyStateOf(topicPartition, key, createdAt, context) def all(topicPartition: TopicPartition) = keyStateOf.all(topicPartition) From 1ec3a400072b5b8dd072e635fb7b9f99448bd102 Mon Sep 17 00:00:00 2001 From: Vladislav Filatov Date: Tue, 2 Jun 2026 14:26:54 +0200 Subject: [PATCH 2/2] Fine-grained logging for PartitionFlow owned classes. Keep log factory in KeyContext, add topic-partition to MDC --- .../kafka/flow/AdditionalStatePersist.scala | 15 +- .../kafka/flow/KeyContext.scala | 42 ++- .../evolutiongaming/kafka/flow/KeyFlow.scala | 3 +- .../kafka/flow/KeyFlowOf.scala | 5 +- .../kafka/flow/KeyStateOf.scala | 11 +- .../kafka/flow/PartitionFlow.scala | 3 +- .../kafka/flow/timer/TimerFlowOf.scala | 239 ++++++++---------- .../kafka/flow/FoldToStateSpec.scala | 4 +- .../kafka/flow/KeyFlowSpec.scala | 25 +- .../kafka/flow/PartitionFlowSpec.scala | 2 +- .../kafka/flow/timer/TimerFlowOfSpec.scala | 7 +- .../kafka/flow/KeyStateMetrics.scala | 3 +- 12 files changed, 163 insertions(+), 196 deletions(-) diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/AdditionalStatePersist.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/AdditionalStatePersist.scala index a8318a5dc..b2a8b1bf3 100644 --- a/core/src/main/scala/com/evolutiongaming/kafka/flow/AdditionalStatePersist.scala +++ b/core/src/main/scala/com/evolutiongaming/kafka/flow/AdditionalStatePersist.scala @@ -4,6 +4,7 @@ import cats.Applicative import cats.effect.syntax.all.* import cats.effect.{Clock, MonadCancel, MonadCancelThrow, Ref} import cats.syntax.all.* +import com.evolutiongaming.catshelper.Log import com.evolutiongaming.kafka.flow.kafka.OffsetToCommit import com.evolutiongaming.kafka.flow.persistence.Persistence import com.evolutiongaming.skafka.consumer.ConsumerRecord @@ -56,9 +57,10 @@ object AdditionalStatePersist { ignorePersistErrors: Boolean = false, ): F[AdditionalStatePersist[F, S, ConsumerRecord[String, ByteVector]]] = { for { + log <- keyContext.log(classOf[AdditionalStatePersist[F, S, ConsumerRecord[String, ByteVector]]]) requestedRef <- Ref.of(false) lastPersistedRef <- Ref.of(none[Instant]) - } yield of(persistence, keyContext, cooldown, requestedRef, lastPersistedRef, ignorePersistErrors) + } yield of(persistence, keyContext, cooldown, requestedRef, lastPersistedRef, ignorePersistErrors, log) } private[flow] def of[F[_]: MonadCancelThrow: Clock, S]( @@ -68,6 +70,7 @@ object AdditionalStatePersist { requestedRef: Ref[F, Boolean], lastPersistedRef: Ref[F, Option[Instant]], ignorePersistErrors: Boolean, + log: Log[F], ): AdditionalStatePersist[F, S, ConsumerRecord[String, ByteVector]] = new AdditionalStatePersist[F, S, ConsumerRecord[String, ByteVector]] { private val F = MonadCancel[F, Throwable] @@ -76,7 +79,7 @@ object AdditionalStatePersist { private val charsToPrint = 1024 override def request: F[Unit] = - requestedRef.set(true) >> keyContext.log.info("Additional persisting requested") + requestedRef.set(true) >> log.info("Additional persisting requested") override def persistIfNeeded(record: ConsumerRecord[String, ByteVector], state: S): F[Unit] = { for { @@ -90,16 +93,14 @@ object AdditionalStatePersist { _ <- persistence.flush.attempt.flatMap { case Left(e) if ignorePersistErrors => val trimmedState = state.toString.take(charsToPrint) - keyContext - .log + log .warn( s"Additional persisting failed, error ignored, error: $e, first $charsToPrint chars of state: $trimmedState", e ) case Left(e) => val trimmedState = state.toString.take(charsToPrint) - keyContext - .log + log .error( s"Additional persisting failed, error: $e, first $charsToPrint chars of state: $trimmedState", e @@ -108,7 +109,7 @@ object AdditionalStatePersist { for { _ <- OffsetToCommit[F](record.offset).flatMap(keyContext.hold) _ <- lastPersistedRef.set(Instant.ofEpochMilli(now).some) - _ <- keyContext.log.info("Additional persisting success") + _ <- log.info("Additional persisting success") } yield () } } yield () diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyContext.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyContext.scala index 14d0b1d10..e394e3785 100644 --- a/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyContext.scala +++ b/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyContext.scala @@ -4,7 +4,7 @@ import cats.effect.{Ref, Resource} import cats.mtl.Stateful import cats.syntax.all.* import cats.{Applicative, Monad} -import com.evolutiongaming.catshelper.Log +import com.evolutiongaming.catshelper.{Log, LogOf} import com.evolutiongaming.kafka.flow.effect.CatsEffectMtlInstances.* import com.evolutiongaming.skafka.Offset @@ -16,45 +16,39 @@ trait KeyContext[F[_]] { def holding: F[Option[Offset]] def hold(offset: Offset): F[Unit] def remove: F[Unit] - def log: Log[F] - def key: String + def log(source: Class[_]): F[Log[F]] } object KeyContext { def apply[F[_]](implicit F: KeyContext[F]): KeyContext[F] = F def empty[F[_]: Applicative]: KeyContext[F] = new KeyContext[F] { - def log = Log.empty - def holding = none[Offset].pure[F] - def hold(offset: Offset) = ().pure[F] - def remove = ().pure[F] - val key = "" + def holding = none[Offset].pure[F] + def hold(offset: Offset) = ().pure[F] + def remove = ().pure[F] + def log(source: Class[_]) = Log.empty[F].pure[F] } - def of[F[_]: Ref.Make: Monad: Log](removeFromCache: F[Unit], key: String): F[KeyContext[F]] = + def of[F[_]: Ref.Make: Monad: LogOf](removeFromCache: F[Unit], mdc: Log.Mdc): F[KeyContext[F]] = Ref.of[F, Option[Offset]](None) map { storage => - KeyContext(storage.stateInstance, removeFromCache, key) + KeyContext(storage.stateInstance, removeFromCache, mdc) } - def apply[F[_]: Monad: Log]( + def apply[F[_]: Monad: LogOf]( storage: Stateful[F, Option[Offset]], removeFromCache: F[Unit], - _key: String + mdc: Log.Mdc ): KeyContext[F] = new KeyContext[F] { - def holding = storage.get - def hold(offset: Offset) = storage.set(Some(offset)) - def remove = storage.set(None) *> removeFromCache - def log = Log[F] - val key = _key + def holding = storage.get + def hold(offset: Offset) = storage.set(Some(offset)) + def remove = storage.set(None) *> removeFromCache + def log(source: Class[_]) = LogOf[F].apply(source).map(_.withMdc(mdc)) } - def resource[F[_]: Ref.Make: Monad]( + def resource[F[_]: Ref.Make: Monad: LogOf]( removeFromCache: F[Unit], - log: Log[F], - key: String - ): Resource[F, KeyContext[F]] = { - implicit val _log = log - Resource.eval(of(removeFromCache, key)) - } + mdc: Log.Mdc + ): Resource[F, KeyContext[F]] = + Resource.eval(of(removeFromCache, mdc)) } diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyFlow.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyFlow.scala index ab6d0e7e2..82ae6b36d 100644 --- a/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyFlow.scala +++ b/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyFlow.scala @@ -74,7 +74,8 @@ object KeyFlow { registry: EntityRegistry[F, KafkaKey, S], ): Resource[F, KeyFlow[F, A]] = for { - state <- persistence.read(KeyContext[F].log).toResource + log <- KeyContext[F].log(classOf[KeyFlow[F, A]]).toResource + state <- persistence.read(log).toResource _ <- storage.set(state).toResource // we should not run any timers if there was decision // by fold or tick to run the state, because in this diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyFlowOf.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyFlowOf.scala index 74b6c6a0a..b3b98901b 100644 --- a/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyFlowOf.scala +++ b/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyFlowOf.scala @@ -2,7 +2,6 @@ package com.evolutiongaming.kafka.flow import cats.Monad import cats.effect.{Ref, Resource} -import com.evolutiongaming.catshelper.LogOf import com.evolutiongaming.kafka.flow.persistence.Persistence import com.evolutiongaming.kafka.flow.registry.EntityRegistry import com.evolutiongaming.kafka.flow.timer.{TimerContext, TimerFlowOf} @@ -16,7 +15,7 @@ trait KeyFlowOf[F[_], S, A] { timers: TimerContext[F], additionalPersist: AdditionalStatePersist[F, S, A], registry: EntityRegistry[F, KafkaKey, S], - )(implicit logOf: LogOf[F]): Resource[F, KeyFlow[F, A]] + ): Resource[F, KeyFlow[F, A]] } object KeyFlowOf { @@ -59,7 +58,7 @@ object KeyFlowOf { timers: TimerContext[F], additionalPersist: AdditionalStatePersist[F, S, A], registry: EntityRegistry[F, KafkaKey, S] - )(implicit logOf: LogOf[F]): Resource[F, KeyFlow[F, A]] = { + ): Resource[F, KeyFlow[F, A]] = { implicit val _context = context timerFlowOf(context, persistence, timers) flatMap { timerFlow => KeyFlow.of(key, fold, tick, persistence, additionalPersist, timerFlow, registry) diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyStateOf.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyStateOf.scala index 240a39191..512be80c7 100644 --- a/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyStateOf.scala +++ b/core/src/main/scala/com/evolutiongaming/kafka/flow/KeyStateOf.scala @@ -3,7 +3,6 @@ package com.evolutiongaming.kafka.flow import cats.Applicative import cats.effect.{Resource, Sync} import cats.syntax.all.* -import com.evolutiongaming.catshelper.LogOf import com.evolutiongaming.kafka.flow.key.KeysOf import com.evolutiongaming.kafka.flow.persistence.{PersistenceOf, SnapshotPersistenceOf} import com.evolutiongaming.kafka.flow.registry.EntityRegistry @@ -21,7 +20,7 @@ trait KeyStateOf[F[_]] { self => key: String, createdAt: Timestamp, context: KeyContext[F] - )(implicit logOf: LogOf[F]): Resource[F, KeyState[F, ConsumerRecord[String, ByteVector]]] + ): Resource[F, KeyState[F, ConsumerRecord[String, ByteVector]]] /** Restores a state for all keys present in persistence. * @@ -72,9 +71,7 @@ object KeyStateOf { registry: EntityRegistry[F, KafkaKey, S], ): KeyStateOf[F] = new KeyStateOf[F] { - def apply(topicPartition: TopicPartition, key: String, createdAt: Timestamp, context: KeyContext[F])( - implicit logOf: LogOf[F] - ) = { + def apply(topicPartition: TopicPartition, key: String, createdAt: Timestamp, context: KeyContext[F]) = { implicit val _context = context val kafkaKey = KafkaKey( applicationId = applicationId, @@ -220,9 +217,7 @@ object KeyStateOf { registry: EntityRegistry[F, KafkaKey, S], ): KeyStateOf[F] = new KeyStateOf[F] { - def apply(topicPartition: TopicPartition, key: String, createdAt: Timestamp, context: KeyContext[F])( - implicit logOf: LogOf[F] - ) = { + def apply(topicPartition: TopicPartition, key: String, createdAt: Timestamp, context: KeyContext[F]) = { val kafkaKey = KafkaKey( applicationId = applicationId, groupId = groupId, diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/PartitionFlow.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/PartitionFlow.scala index 2660e0382..b73d1e1cf 100644 --- a/core/src/main/scala/com/evolutiongaming/kafka/flow/PartitionFlow.scala +++ b/core/src/main/scala/com/evolutiongaming/kafka/flow/PartitionFlow.scala @@ -116,8 +116,7 @@ object PartitionFlow { for { context <- KeyContext.resource[F]( removeFromCache = cache.remove(key).flatten.void, - log = log.prefixed(key), - key = key + mdc = Log.Mdc.Eager("key" -> key, "topicPartition" -> topicPartition.toString) ) keyState <- keyStateOf(topicPartition, key, createdAt, context) } yield PartitionKey(keyState, context) diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOf.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOf.scala index d63ecfabf..61d96dc9c 100644 --- a/core/src/main/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOf.scala +++ b/core/src/main/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOf.scala @@ -5,8 +5,7 @@ import cats.effect.Resource import cats.effect.syntax.all.* import cats.effect.kernel.Resource.ExitCase import cats.syntax.all.* -import com.evolutiongaming.catshelper.Log.Mdc -import com.evolutiongaming.catshelper.{Log, LogOf} +import com.evolutiongaming.catshelper.Log import com.evolutiongaming.kafka.flow.KeyContext import com.evolutiongaming.kafka.flow.persistence.FlushBuffers import com.evolutiongaming.skafka.Offset @@ -19,7 +18,7 @@ trait TimerFlowOf[F[_]] { context: KeyContext[F], persistence: FlushBuffers[F], timers: TimerContext[F] - )(implicit logOf: LogOf[F]): Resource[F, TimerFlow[F]] + ): Resource[F, TimerFlow[F]] } object TimerFlowOf { @@ -40,49 +39,44 @@ object TimerFlowOf { maxOffsetDifference: Int = 100000, maxIdle: FiniteDuration = 10.minutes, flushOnRevoke: Boolean = false, - ): TimerFlowOf[F] = - new TimerFlowOf[F] { - override def apply(context: KeyContext[F], persistence: FlushBuffers[F], timers: TimerContext[F])( - implicit logOf: LogOf[F] - ): Resource[F, TimerFlow[F]] = { - def register(touchedAt: Timestamp): F[Unit] = - timers.registerProcessing(touchedAt.clock plusMillis fireEvery.toMillis) - - val acquire = Resource.eval { - for { - log <- logOf(classOf[TimerFlowOf[F]]).map(_.withMdc(Mdc.Eager("key" -> context.key))) - current <- timers.current - persistedAt <- timers.persistedAt - committedAt = persistedAt getOrElse current - _ <- context.hold(committedAt.offset) - _ <- register(committedAt) - } yield new TimerFlow[F] { - def onTimer: F[Unit] = for { - current <- timers.current - processedAt <- timers.processedAt - touchedAt = processedAt getOrElse committedAt - expiredAt = touchedAt.clock plusMillis maxIdle.toMillis - expired = current.clock isAfter expiredAt - offsetDifference = current.offset.value - touchedAt.offset.value - canUnload = expired || offsetDifference > maxOffsetDifference - _ <- - if (canUnload) { - log.info(s"flush, offset difference: $offsetDifference") *> - persistence.flush *> - context.remove - } else { - register(touchedAt) - } - } yield () - } - } - - val cancel = flushOnCancel.apply(context, persistence, timers) - - if (flushOnRevoke) acquire <* cancel else acquire + ): TimerFlowOf[F] = { (context, persistence, timers) => + def register(touchedAt: Timestamp): F[Unit] = + timers.registerProcessing(touchedAt.clock plusMillis fireEvery.toMillis) + + val acquire = Resource.eval { + for { + log <- context.log(classOf[TimerFlowOf[F]]) + current <- timers.current + persistedAt <- timers.persistedAt + committedAt = persistedAt getOrElse current + _ <- context.hold(committedAt.offset) + _ <- register(committedAt) + } yield new TimerFlow[F] { + def onTimer: F[Unit] = for { + current <- timers.current + processedAt <- timers.processedAt + touchedAt = processedAt getOrElse committedAt + expiredAt = touchedAt.clock plusMillis maxIdle.toMillis + expired = current.clock isAfter expiredAt + offsetDifference = current.offset.value - touchedAt.offset.value + canUnload = expired || offsetDifference > maxOffsetDifference + _ <- + if (canUnload) { + log.info(s"flush, offset difference: $offsetDifference") *> + persistence.flush *> + context.remove + } else { + register(touchedAt) + } + } yield () } } + val cancel = flushOnCancel.apply(context, persistence, timers) + + if (flushOnRevoke) acquire <* cancel else acquire + } + /** Performs flush periodically. * * The flush will be called every `persistEvery` FiniteDuration. @@ -109,45 +103,41 @@ object TimerFlowOf { persistEvery: FiniteDuration = 1.minute, flushOnRevoke: Boolean = false, ignorePersistErrors: Boolean = false, - ): TimerFlowOf[F] = new TimerFlowOf[F] { - override def apply(context: KeyContext[F], persistence: FlushBuffers[F], timers: TimerContext[F])( - implicit logOf: LogOf[F] - ): Resource[F, TimerFlow[F]] = { - def register(current: Timestamp): F[Unit] = - timers.registerProcessing(current.clock plusMillis fireEvery.toMillis) - - val acquire = Resource.eval { - for { - log <- logOf(classOf[TimerFlowOf[F]]).map(_.withMdc(Mdc.Eager("key" -> context.key))) - current <- timers.current - persistedAt <- timers.persistedAt - committedAt = persistedAt getOrElse current - _ <- context.hold(committedAt.offset) - _ <- register(current) - } yield new TimerFlow[F] { - def onTimer: F[Unit] = for { - current <- timers.current - persistedAt <- timers.persistedAt - flushedAt = persistedAt getOrElse committedAt - triggerFlushAt = flushedAt.clock plusMillis persistEvery.toMillis - canPersist = (current.clock compareTo triggerFlushAt) >= 0 - _ <- MonadThrow[F] - .whenA(canPersist)( - persistence.attemptToPersist( - ignorePersistErrors = ignorePersistErrors, - context = context, - currentOffset = current.offset - )(log) - ) - _ <- register(current) - } yield () - } + ): TimerFlowOf[F] = { (context, persistence, timers) => + def register(current: Timestamp): F[Unit] = + timers.registerProcessing(current.clock plusMillis fireEvery.toMillis) + + val acquire = Resource.eval { + for { + log <- context.log(classOf[TimerFlowOf[F]]) + current <- timers.current + persistedAt <- timers.persistedAt + committedAt = persistedAt getOrElse current + _ <- context.hold(committedAt.offset) + _ <- register(current) + } yield new TimerFlow[F] { + def onTimer: F[Unit] = for { + current <- timers.current + persistedAt <- timers.persistedAt + flushedAt = persistedAt getOrElse committedAt + triggerFlushAt = flushedAt.clock plusMillis persistEvery.toMillis + canPersist = (current.clock compareTo triggerFlushAt) >= 0 + _ <- MonadThrow[F] + .whenA(canPersist)( + persistence.attemptToPersist( + ignorePersistErrors = ignorePersistErrors, + context = context, + currentOffset = current.offset + )(log) + ) + _ <- register(current) + } yield () } + } - val cancel = flushOnCancel.apply(context, persistence, timers) + val cancel = flushOnCancel.apply(context, persistence, timers) - if (flushOnRevoke) acquire <* cancel else acquire - } + if (flushOnRevoke) acquire <* cancel else acquire } /** Combines [[unloadOrphaned]] with [[persistPeriodically]] in a single TimerFlow @@ -174,60 +164,54 @@ object TimerFlowOf { maxIdle: FiniteDuration = 10.minutes, flushOnRevoke: Boolean = false, ignorePersistErrors: Boolean = false, - ): TimerFlowOf[F] = new TimerFlowOf[F] { - override def apply(context: KeyContext[F], persistence: FlushBuffers[F], timers: TimerContext[F])( - implicit logOf: LogOf[F] - ): Resource[F, TimerFlow[F]] = { - def register(touchedAt: Timestamp): F[Unit] = - timers.registerProcessing(touchedAt.clock plusMillis fireEvery.toMillis) - - val acquire: Resource[F, TimerFlow[F]] = Resource.eval { - for { - log <- logOf(classOf[TimerFlowOf[F]]).map(_.withMdc(Mdc.Eager("key" -> context.key))) - current <- timers.current - persistedAt <- timers.persistedAt - committedAt = persistedAt getOrElse current - _ <- context.hold(committedAt.offset) - _ <- register(committedAt) - } yield new TimerFlow[F] { - def onTimer: F[Unit] = for { - current <- timers.current - processedAt <- timers.processedAt - touchedAt = processedAt getOrElse committedAt - expiredAt = touchedAt.clock plusMillis maxIdle.toMillis - offsetDifference = current.offset.value - touchedAt.offset.value - flushedAt = persistedAt getOrElse committedAt - triggerFlushAt = flushedAt.clock plusMillis persistEvery.toMillis - expired = current.clock isAfter expiredAt - canUnload = expired || offsetDifference > maxOffsetDifference - canPersist = (current.clock compareTo triggerFlushAt) >= 0 - _ <- Applicative[F].whenA(canPersist || canUnload)( - persistence.attemptToPersist( - ignorePersistErrors = ignorePersistErrors, - context = context, - currentOffset = current.offset - )(log) - ) - _ <- Applicative[F].whenA(canUnload)( - log.info(s"flush, offset difference: $offsetDifference") *> context.remove - ) - _ <- register(current) - } yield () - } + ): TimerFlowOf[F] = { (context, persistence, timers) => + def register(touchedAt: Timestamp): F[Unit] = + timers.registerProcessing(touchedAt.clock plusMillis fireEvery.toMillis) + + val acquire: Resource[F, TimerFlow[F]] = Resource.eval { + for { + log <- context.log(classOf[TimerFlowOf[F]]) + current <- timers.current + persistedAt <- timers.persistedAt + committedAt = persistedAt getOrElse current + _ <- context.hold(committedAt.offset) + _ <- register(committedAt) + } yield new TimerFlow[F] { + def onTimer: F[Unit] = for { + current <- timers.current + processedAt <- timers.processedAt + touchedAt = processedAt getOrElse committedAt + expiredAt = touchedAt.clock plusMillis maxIdle.toMillis + offsetDifference = current.offset.value - touchedAt.offset.value + flushedAt = persistedAt getOrElse committedAt + triggerFlushAt = flushedAt.clock plusMillis persistEvery.toMillis + expired = current.clock isAfter expiredAt + canUnload = expired || offsetDifference > maxOffsetDifference + canPersist = (current.clock compareTo triggerFlushAt) >= 0 + _ <- Applicative[F].whenA(canPersist || canUnload)( + persistence.attemptToPersist( + ignorePersistErrors = ignorePersistErrors, + context = context, + currentOffset = current.offset + )(log) + ) + _ <- Applicative[F].whenA(canUnload)( + log.info(s"flush, offset difference: $offsetDifference") *> context.remove + ) + _ <- register(current) + } yield () } + } - val cancel = flushOnCancel.apply(context, persistence, timers) + val cancel = flushOnCancel.apply(context, persistence, timers) - if (flushOnRevoke) acquire <* cancel else acquire - } + if (flushOnRevoke) acquire <* cancel else acquire } /** Performs flush when `Resource` is cancelled only */ - def flushOnCancel[F[_]: Monad]: TimerFlowOf[F] = new TimerFlowOf[F] { - override def apply(context: KeyContext[F], persistence: FlushBuffers[F], timers: TimerContext[F])( - implicit logOf: LogOf[F] - ): Resource[F, TimerFlow[F]] = - logOf(classOf[TimerFlowOf[F]]).toResource.map(_.withMdc(Mdc.Eager("key" -> context.key))).flatMap { log => + def flushOnCancel[F[_]: Monad]: TimerFlowOf[F] = + (context: KeyContext[F], persistence: FlushBuffers[F], _: TimerContext[F]) => + context.log(classOf[TimerFlowOf[F]]).toResource.flatMap { log => val cancel = context.holding flatMap { holding => Applicative[F].whenA(holding.isDefined) { log.info(s"flush on revoke, holding offset: $holding") *> @@ -248,7 +232,6 @@ object TimerFlowOf { case (_, _) => ().pure[F] } } - } private implicit class AttemptToPersist[F[_]: MonadThrow](persistence: FlushBuffers[F]) { def attemptToPersist(ignorePersistErrors: Boolean, context: KeyContext[F], currentOffset: Offset)( diff --git a/core/src/test/scala/com/evolutiongaming/kafka/flow/FoldToStateSpec.scala b/core/src/test/scala/com/evolutiongaming/kafka/flow/FoldToStateSpec.scala index 1ee2b12be..c66270c70 100644 --- a/core/src/test/scala/com/evolutiongaming/kafka/flow/FoldToStateSpec.scala +++ b/core/src/test/scala/com/evolutiongaming/kafka/flow/FoldToStateSpec.scala @@ -2,6 +2,7 @@ package com.evolutiongaming.kafka.flow import cats.data.{NonEmptyList, State} import cats.mtl.Stateful +import cats.syntax.all.* import com.evolutiongaming.catshelper.Log import com.evolutiongaming.kafka.flow.FoldToStateSpec.* import com.evolutiongaming.kafka.flow.MonadStateHelper.* @@ -134,8 +135,7 @@ object FoldToStateSpec { def remove: F[Unit] = State.modify { context => context.copy(removeCalled = context.removeCalled + 1) } - def log = Log.empty - def key = "" + def log(source: Class[_]) = Log.empty[F].pure[F] } } diff --git a/core/src/test/scala/com/evolutiongaming/kafka/flow/KeyFlowSpec.scala b/core/src/test/scala/com/evolutiongaming/kafka/flow/KeyFlowSpec.scala index 32df37f79..be9e23773 100644 --- a/core/src/test/scala/com/evolutiongaming/kafka/flow/KeyFlowSpec.scala +++ b/core/src/test/scala/com/evolutiongaming/kafka/flow/KeyFlowSpec.scala @@ -19,8 +19,6 @@ import java.time.Instant class KeyFlowSpec extends FunSuite { - private implicit val log: LogOf[SyncIO] = LogOf.empty - test("KeyFlow processes messages correctly") { val f = new ConstFixture @@ -87,11 +85,10 @@ class KeyFlowSpec extends FunSuite { } val timerFlowOf = TimerFlowOf.unloadOrphaned[SyncIO]() implicit val context: KeyContext[SyncIO] = new KeyContext[SyncIO] { - def holding = none[Offset].pure[SyncIO] - def hold(offset: Offset) = SyncIO.unit - def remove = removeCalled.set(true) - def log = Log.empty - def key = "" + def holding = none[Offset].pure[SyncIO] + def hold(offset: Offset) = SyncIO.unit + def remove = removeCalled.set(true) + def log(source: Class[_]) = Log.empty[SyncIO].pure[SyncIO] } val key = KafkaKey(applicationId = "test", groupId = "test", topicPartition = TopicPartition.empty, key = "key") val keyFlow = timerFlowOf(context, persistence, timers).flatMap(tf => @@ -138,11 +135,10 @@ class KeyFlowSpec extends FunSuite { val timerFlowOf = TimerFlowOf.unloadOrphaned[SyncIO]() implicit val context: KeyContext[SyncIO] = new KeyContext[SyncIO] { - def holding = none[Offset].pure[SyncIO] - def hold(offset: Offset) = SyncIO.unit - def remove = removeCalled.set(true) - def log = Log.empty - def key = "" + def holding = none[Offset].pure[SyncIO] + def hold(offset: Offset) = SyncIO.unit + def remove = removeCalled.set(true) + def log(source: Class[_]) = Log.empty[SyncIO].pure[SyncIO] } val key = KafkaKey(applicationId = "test", groupId = "test", topicPartition = TopicPartition.empty, key = "key") @@ -295,10 +291,11 @@ object KeyFlowSpec { val registry: EntityRegistry[SyncIO, KafkaKey, State] = EntityRegistry.empty } - implicit val log: Log[SyncIO] = Log.empty + implicit val log: Log[SyncIO] = Log.empty + implicit val logOf: LogOf[SyncIO] = LogOf.empty implicit val context: KeyContext[SyncIO] = - KeyContext.of(().pure[SyncIO], "").unsafeRunSync() + KeyContext.of(().pure[SyncIO], Log.Mdc.empty).unsafeRunSync() implicit val stateToOffset: ToOffset[State] = { case (offset, _) => diff --git a/core/src/test/scala/com/evolutiongaming/kafka/flow/PartitionFlowSpec.scala b/core/src/test/scala/com/evolutiongaming/kafka/flow/PartitionFlowSpec.scala index 3e0bb9c3c..fa19a1ce8 100644 --- a/core/src/test/scala/com/evolutiongaming/kafka/flow/PartitionFlowSpec.scala +++ b/core/src/test/scala/com/evolutiongaming/kafka/flow/PartitionFlowSpec.scala @@ -508,7 +508,7 @@ object PartitionFlowSpec { key: String, createdAt: Timestamp, context: KeyContext[IO] - )(implicit logOf: LogOf[IO]): Resource[IO, KeyState[IO, ConsumerRecord[String, ByteVector]]] = { + ): Resource[IO, KeyState[IO, ConsumerRecord[String, ByteVector]]] = { implicit val _context = context val fold0 = fold val kafkaKey = KafkaKey("test", "test", topicPartition, key) diff --git a/core/src/test/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOfSpec.scala b/core/src/test/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOfSpec.scala index 21a2573b5..c24d4875f 100644 --- a/core/src/test/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOfSpec.scala +++ b/core/src/test/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOfSpec.scala @@ -20,8 +20,6 @@ import scala.concurrent.duration.* class TimerFlowOfSpec extends FunSuite { - private implicit val log: LogOf[IO] = LogOf.empty - test("unloadOrphaned holds commits when started") { val f = new ConstFixture @@ -573,7 +571,8 @@ class TimerFlowOfSpec extends FunSuite { } object TimerFlowSpec { - implicit val log: Log[IO] = Log.empty[IO] + implicit val log: Log[IO] = Log.empty[IO] + implicit val logOf: LogOf[IO] = LogOf.empty case class Context( holding: Option[Offset] = None, @@ -603,7 +602,7 @@ object TimerFlowSpec { KeyContext( storage = contextRef.stateInstance.focus(Context.lens(_.holding)), removeFromCache = contextRef.update(ctx => ctx.copy(removed = ctx.removed + 1)), - _key = "test-key" + mdc = Log.Mdc.Eager("key" -> "test-key") ) implicit val timerContext: TimerContext[IO] = { diff --git a/metrics/src/main/scala/com/evolutiongaming/kafka/flow/KeyStateMetrics.scala b/metrics/src/main/scala/com/evolutiongaming/kafka/flow/KeyStateMetrics.scala index 7509084b8..ce13acaad 100644 --- a/metrics/src/main/scala/com/evolutiongaming/kafka/flow/KeyStateMetrics.scala +++ b/metrics/src/main/scala/com/evolutiongaming/kafka/flow/KeyStateMetrics.scala @@ -3,7 +3,6 @@ package com.evolutiongaming.kafka.flow import cats.Monad import cats.effect.Resource import cats.syntax.all.* -import com.evolutiongaming.catshelper.LogOf import com.evolutiongaming.kafka.flow.metrics.MetricsOf import com.evolutiongaming.kafka.flow.timer.Timestamp import com.evolutiongaming.skafka.TopicPartition @@ -29,7 +28,7 @@ object KeyStateMetrics { key: String, createdAt: Timestamp, context: KeyContext[F] - )(implicit logOf: LogOf[F]) = + ) = count(topicPartition.topic) *> keyStateOf(topicPartition, key, createdAt, context) def all(topicPartition: TopicPartition) =