KAFKA-1614: Partition log directory and segments info exposed via JMX

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

@@ -95,6 +95,8 @@ class LogSegment private[log] (val log: FileRecords,
 
   private var created = time.milliseconds
 
+  def createdMs: Long = created
+
   /* the number of bytes since we last added an entry in the offset index */
   private var bytesSinceLastIndexEntry = 0
 
@@ -123,6 +125,9 @@ class LogSegment private[log] (val log: FileRecords,
   /* Return the size in bytes of this log segment */
   def size: Int = log.sizeInBytes()
 
+  /* Return the index size in bytes of this log segment */
+  def indexSize: Int = offsetIndex.sizeInBytes + timeIndex.sizeInBytes
+
   /**
    * checks that the argument offset can be represented as an integer offset relative to the baseOffset.
    */

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

@@ -47,6 +47,7 @@ import java.nio.file.Files
 import java.util
 import java.util.concurrent.{ConcurrentHashMap, ConcurrentMap}
 import java.util.{Collections, Optional, OptionalInt, OptionalLong}
+import java.util.{List => JList, ArrayList => JArrayList}
 import scala.annotation.nowarn
 import scala.collection.mutable.ListBuffer
 import scala.collection.{Seq, immutable, mutable}
@@ -435,10 +436,12 @@ class UnifiedLog(@volatile var logStartOffset: Long,
     val tags = (Map("topic" -> topicPartition.topic, "partition" -> topicPartition.partition.toString) ++
       (if (isFuture) Map("is-future" -> "true") else Map.empty)).asJava
     metricsGroup.newGauge(LogMetricNames.NumLogSegments, () => numberOfSegments, tags)
+    metricsGroup.newGauge(LogMetricNames.LogSegments, () => logSegmentsDetail, tags)
     metricsGroup.newGauge(LogMetricNames.LogStartOffset, () => logStartOffset, tags)
     metricsGroup.newGauge(LogMetricNames.LogEndOffset, () => logEndOffset, tags)
     metricsGroup.newGauge(LogMetricNames.Size, () => size, tags)
     metricNames = Map(LogMetricNames.NumLogSegments -> tags,
+      LogMetricNames.LogSegments -> tags,
       LogMetricNames.LogStartOffset -> tags,
       LogMetricNames.LogEndOffset -> tags,
       LogMetricNames.Size -> tags)
@@ -588,6 +591,16 @@ class UnifiedLog(@volatile var logStartOffset: Long,
    */
   def numberOfSegments: Int = localLog.segments.numberOfSegments
 
+  /**
+   * The detailed metrics for log segments
+   */
+  def logSegmentsDetail: JList[String] = {
+    val list = logSegments.toSeq.map { seg =>
+      s"baseOffset=${seg.baseOffset}; created=${seg.createdMs}; logSize=${seg.size}; indexSize=${seg.indexSize}"
+    }
+    new JArrayList[String](list.asJava)
+  }
+
   /**
    * Close this log.
    * The memory mapped buffer for index files of this log will be left open until the log is deleted.
@@ -2114,12 +2127,13 @@ object UnifiedLog extends Logging {
 
 object LogMetricNames {
   val NumLogSegments: String = "NumLogSegments"
+  val LogSegments: String = "LogSegments"
   val LogStartOffset: String = "LogStartOffset"
   val LogEndOffset: String = "LogEndOffset"
   val Size: String = "Size"
 
   def allMetricNames: List[String] = {
-    List(NumLogSegments, LogStartOffset, LogEndOffset, Size)
+    List(NumLogSegments, LogSegments, LogStartOffset, LogEndOffset, Size)
   }
 }
 

docs/ops.html

@@ -1847,6 +1847,11 @@ <h4 class="anchor-heading"><a id="remote_jmx" class="anchor-link"></a><a href="#
         <td>kafka.log:type=Log,name=NumLogSegments,topic=([-.\w]+),partition=([0-9]+)</td>
         <td>The number of log segments in a partition.</td>
       </tr>
+      <tr>
+        <td>Detailed log segments info in a partition</td>
+        <td>kafka.log:type=Log,name=LogSegments,topic=([-.\w]+),partition=([0-9]+)</td>
+        <td>The detailed metrics of log segments in a partition.</td>
+      </tr>
       <tr>
         <td>First offset in a partition</td>
         <td>kafka.log:type=Log,name=LogStartOffset,topic=([-.\w]+),partition=([0-9]+)</td>