log topic-partition and key in KeyDatabase/Keys logging when deleting…

… a key (#670)
evolution-gaming · Jan 21, 2025 · d3dfc30 · d3dfc30
1 parent c86104f
commit d3dfc30
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 9 deletions.
diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/key/KeyDatabase.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/key/KeyDatabase.scala
@@ -5,6 +5,7 @@ import cats.mtl.Stateful
 import cats.syntax.all._
 import cats.{Applicative, Monad}
 import com.evolutiongaming.catshelper.LogOf
+import com.evolutiongaming.kafka.flow.LogPrefix
 import com.evolutiongaming.kafka.flow.effect.CatsEffectMtlInstances._
 import com.evolutiongaming.skafka.TopicPartition
 import com.evolutiongaming.sstream.Stream
@@ -19,7 +20,7 @@ trait KeyDatabase[F[_], K] {
 
   def all(applicationId: String, groupId: String, topicPartition: TopicPartition): Stream[F, K]
 
-  def keysOf(implicit F: Monad[F], logOf: LogOf[F]): F[KeysOf[F, K]] =
+  def keysOf(implicit F: Monad[F], logOf: LogOf[F], logPrefix: LogPrefix[K]): F[KeysOf[F, K]] =
     logOf(KeyDatabase.getClass) map { implicit log => KeysOf(this) }
 
 }

diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/key/Keys.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/key/Keys.scala
@@ -1,9 +1,9 @@
 package com.evolutiongaming.kafka.flow.key
 
-import cats.Applicative
-import cats.Monad
 import cats.syntax.all._
+import cats.{Applicative, Monad}
 import com.evolutiongaming.catshelper.Log
+import com.evolutiongaming.kafka.flow.LogPrefix
 
 trait Keys[F[_]] extends KeyWriter[F]
 
@@ -24,17 +24,18 @@ trait KeyWriter[F[_]] {
 object Keys {
 
   /** Creates a buffer for a given writer */
-  private[key] def apply[F[_]: Monad: Log, K](
+  private[key] def apply[F[_]: Monad: Log, K: LogPrefix](
     key: K,
     database: KeyDatabase[F, K]
   ): Keys[F] = new Keys[F] {
 
+    private val prefixedLog = Log[F].prefixed(s"[${LogPrefix[K].extract(key)}]")
+
     def flush: F[Unit] = database.persist(key)
 
     def delete(persist: Boolean): F[Unit] =
       if (persist) {
-        database.delete(key) *>
-          Log[F].info("deleted key")
+        database.delete(key) *> prefixedLog.info("deleted key")
       } else {
         ().pure[F]
       }

diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/key/KeysOf.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/key/KeysOf.scala
@@ -1,9 +1,10 @@
 package com.evolutiongaming.kafka.flow.key
 
+import cats.Monad
 import cats.effect.Sync
 import cats.syntax.all._
-import cats.Monad
 import com.evolutiongaming.catshelper.Log
+import com.evolutiongaming.kafka.flow.LogPrefix
 import com.evolutiongaming.skafka.TopicPartition
 import com.evolutiongaming.sstream.Stream
 
@@ -16,13 +17,13 @@ trait KeysOf[F[_], K] {
 }
 object KeysOf {
 
-  def memory[F[_]: Sync: Log, K]: F[KeysOf[F, K]] =
+  def memory[F[_]: Sync: Log, K: LogPrefix]: F[KeysOf[F, K]] =
     KeyDatabase.memory[F, K] map { database =>
       KeysOf(database)
     }
 
   /** Creates `KeysOf` with a passed logger */
-  def apply[F[_]: Monad: Log, K](
+  def apply[F[_]: Monad: Log, K: LogPrefix](
     database: KeyDatabase[F, K]
   ): KeysOf[F, K] = new KeysOf[F, K] {
     def apply(key: K) = Keys(key, database)