kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jun...@apache.org
Subject kafka git commit: KAFKA-4734; Trim the time index on old segments
Date Tue, 07 Feb 2017 16:43:30 GMT
Repository: kafka
Updated Branches:
  refs/heads/0.10.2 6bdc0da9a -> aca81877f


KAFKA-4734; Trim the time index on old segments

Author: Jiangjie Qin <becket.qin@gmail.com>

Reviewers: Dong Lin <lindong28@gmail.com>, Jun Rao <junrao@gmail.com>

Closes #2501 from becketqin/KAFKA-4734

(cherry picked from commit 6cef4f5e4c3a73911a43378caf5cfc2736814333)
Signed-off-by: Jun Rao <junrao@gmail.com>


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/aca81877
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/aca81877
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/aca81877

Branch: refs/heads/0.10.2
Commit: aca81877fbdaa17649ada327370526a713e751e4
Parents: 6bdc0da
Author: Jiangjie Qin <becket.qin@gmail.com>
Authored: Tue Feb 7 08:43:02 2017 -0800
Committer: Jun Rao <junrao@gmail.com>
Committed: Tue Feb 7 08:43:19 2017 -0800

----------------------------------------------------------------------
 core/src/main/scala/kafka/log/Log.scala          | 4 ++++
 core/src/test/scala/unit/kafka/log/LogTest.scala | 4 +++-
 2 files changed, 7 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/aca81877/core/src/main/scala/kafka/log/Log.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/log/Log.scala b/core/src/main/scala/kafka/log/Log.scala
index 25bb83d..417122c 100644
--- a/core/src/main/scala/kafka/log/Log.scala
+++ b/core/src/main/scala/kafka/log/Log.scala
@@ -206,6 +206,7 @@ class Log(@volatile var dir: File,
         val timeIndexFile = Log.timeIndexFilename(dir, start)
 
         val indexFileExists = indexFile.exists()
+        val timeIndexFileExists = timeIndexFile.exists()
         val segment = new LogSegment(dir = dir,
                                      startOffset = start,
                                      indexIntervalBytes = config.indexInterval,
@@ -217,6 +218,9 @@ class Log(@volatile var dir: File,
         if (indexFileExists) {
           try {
             segment.index.sanityCheck()
+            // Resize the time index file to 0 if it is newly created.
+            if (!timeIndexFileExists)
+              segment.timeIndex.resize(0)
             segment.timeIndex.sanityCheck()
           } catch {
             case e: java.lang.IllegalArgumentException =>

http://git-wip-us.apache.org/repos/asf/kafka/blob/aca81877/core/src/test/scala/unit/kafka/log/LogTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/log/LogTest.scala b/core/src/test/scala/unit/kafka/log/LogTest.scala
index 08cdac5..9e0deb2 100755
--- a/core/src/test/scala/unit/kafka/log/LogTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogTest.scala
@@ -659,8 +659,10 @@ class LogTest extends JUnitSuite {
     // The rebuilt time index should be empty
     log = new Log(logDir, config, recoveryPoint = numMessages + 1, time.scheduler, time)
     val segArray = log.logSegments.toArray
-    for (i <- 0 until segArray.size - 1)
+    for (i <- 0 until segArray.size - 1) {
       assertEquals("The time index should be empty", 0, segArray(i).timeIndex.entries)
+      assertEquals("The time index file size should be 0", 0, segArray(i).timeIndex.file.length)
+    }
 
   }
 


Mime
View raw message