KAFKA-6569: Move OffsetIndex/TimeIndex logger to companion object

core/src/main/scala/kafka/log/AbstractIndex.scala

@@ -39,7 +39,8 @@ import scala.math.ceil
  * @param maxIndexSize The maximum index size in bytes.
  */
 abstract class AbstractIndex[K, V](@volatile var file: File, val baseOffset: Long,
-                                   val maxIndexSize: Int = -1, val writable: Boolean) extends Closeable with Logging {
+                                   val maxIndexSize: Int = -1, val writable: Boolean) extends Closeable {
+  import AbstractIndex._
 
   // Length of the index file
   @volatile
@@ -135,7 +136,7 @@ abstract class AbstractIndex[K, V](@volatile var file: File, val baseOffset: Lon
         idx.position(roundDownToExactMultiple(idx.limit(), entrySize))
       idx
     } finally {
-      CoreUtils.swallow(raf.close(), this)
+      CoreUtils.swallow(raf.close(), AbstractIndex)
     }
   }
 
@@ -190,7 +191,7 @@ abstract class AbstractIndex[K, V](@volatile var file: File, val baseOffset: Lon
           mmap.position(position)
           true
         } finally {
-          CoreUtils.swallow(raf.close(), this)
+          CoreUtils.swallow(raf.close(), AbstractIndex)
         }
       }
     }
@@ -422,6 +423,10 @@ abstract class AbstractIndex[K, V](@volatile var file: File, val baseOffset: Lon
 
 }
 
+object AbstractIndex extends Logging {
+  override val loggerName: String = classOf[AbstractIndex[_, _]].getName
+}
+
 object IndexSearchType extends Enumeration {
   type IndexSearchEntity = Value
   val KEY, VALUE = Value

core/src/main/scala/kafka/log/OffsetIndex.scala

@@ -21,6 +21,7 @@ import java.io.File
 import java.nio.ByteBuffer
 
 import kafka.utils.CoreUtils.inLock
+import kafka.utils.Logging
 import org.apache.kafka.common.errors.InvalidOffsetException
 
 /**
@@ -51,6 +52,7 @@ import org.apache.kafka.common.errors.InvalidOffsetException
 // Avoid shadowing mutable `file` in AbstractIndex
 class OffsetIndex(_file: File, baseOffset: Long, maxIndexSize: Int = -1, writable: Boolean = true)
     extends AbstractIndex[Long, Int](_file, baseOffset, maxIndexSize, writable) {
+  import OffsetIndex._
 
   override def entrySize = 8
 
@@ -197,3 +199,7 @@ class OffsetIndex(_file: File, baseOffset: Long, maxIndexSize: Int = -1, writabl
   }
 
 }
+
+object OffsetIndex extends Logging {
+  override val loggerName: String = classOf[OffsetIndex].getName
+}

core/src/main/scala/kafka/log/TimeIndex.scala

@@ -20,7 +20,7 @@ package kafka.log
 import java.io.File
 import java.nio.ByteBuffer
 
-import kafka.utils.CoreUtils._
+import kafka.utils.CoreUtils.inLock
 import kafka.utils.Logging
 import org.apache.kafka.common.errors.InvalidOffsetException
 import org.apache.kafka.common.record.RecordBatch
@@ -51,7 +51,8 @@ import org.apache.kafka.common.record.RecordBatch
  */
 // Avoid shadowing mutable file in AbstractIndex
 class TimeIndex(_file: File, baseOffset: Long, maxIndexSize: Int = -1, writable: Boolean = true)
-    extends AbstractIndex[Long, Long](_file, baseOffset, maxIndexSize, writable) with Logging {
+    extends AbstractIndex[Long, Long](_file, baseOffset, maxIndexSize, writable) {
+  import TimeIndex._
 
   @volatile private var _lastEntry = lastEntryFromIndexFile
 
@@ -220,3 +221,7 @@ class TimeIndex(_file: File, baseOffset: Long, maxIndexSize: Int = -1, writable:
         s"which is neither positive nor a multiple of $entrySize.")
   }
 }
+
+object TimeIndex extends Logging {
+  override val loggerName: String = classOf[TimeIndex].getName
+}

core/src/main/scala/kafka/utils/CoreUtils.scala

@@ -24,6 +24,7 @@ import java.util.concurrent.locks.{Lock, ReadWriteLock}
 import java.lang.management._
 import java.util.{Base64, Properties, UUID}
 
+import com.typesafe.scalalogging.Logger
 import javax.management._
 
 import scala.collection._
@@ -45,7 +46,8 @@ import org.slf4j.event.Level
  * 2. It is the most general possible utility, not just the thing you needed in one particular place
  * 3. You have tests for it if it is nontrivial in any way
  */
-object CoreUtils extends Logging {
+object CoreUtils {
+  private val logger = Logger(getClass)
 
   /**
    * Return the smallest element in `traversable` if it is not empty. Otherwise return `ifEmpty`.
@@ -147,7 +149,7 @@ object CoreUtils extends Logging {
       }
     } catch {
       case e: Exception => {
-        error(s"Failed to register Mbean $name", e)
+        logger.error(s"Failed to register Mbean $name", e)
         false
       }
     }