kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jun...@apache.org
Subject kafka git commit: Revert "KAFKA-5413; Log cleaner fails due to large offset in segment file"
Date Wed, 21 Jun 2017 04:55:15 GMT
Repository: kafka
Updated Branches:
  refs/heads/0.10.2 7647b97f3 -> 33727acf9


Revert "KAFKA-5413; Log cleaner fails due to large offset in segment file"

This reverts commit 7647b97f317ad8231a2b77c71ccf7e3ddb29a4cd.


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

Branch: refs/heads/0.10.2
Commit: 33727acf987bb4e4272140ccb6dace775138b9b0
Parents: 7647b97
Author: Jun Rao <junrao@gmail.com>
Authored: Tue Jun 20 21:47:06 2017 -0700
Committer: Jun Rao <junrao@gmail.com>
Committed: Tue Jun 20 21:47:06 2017 -0700

----------------------------------------------------------------------
 core/src/main/scala/kafka/log/LogCleaner.scala  | 29 +---------
 .../scala/unit/kafka/log/LogCleanerTest.scala   | 58 +++-----------------
 2 files changed, 11 insertions(+), 76 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/33727acf/core/src/main/scala/kafka/log/LogCleaner.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/log/LogCleaner.scala b/core/src/main/scala/kafka/log/LogCleaner.scala
index a1d3b67..6c1f13d 100644
--- a/core/src/main/scala/kafka/log/LogCleaner.scala
+++ b/core/src/main/scala/kafka/log/LogCleaner.scala
@@ -357,10 +357,9 @@ private[log] class Cleaner(val id: Int,
     // this is the lower of the last active segment and the compaction lag
     val cleanableHorizonMs = log.logSegments(0, cleanable.firstUncleanableOffset).lastOption.map(_.lastModified).getOrElse(0L)
 
-
     // group the segments and clean the groups
     info("Cleaning log %s (cleaning prior to %s, discarding tombstones prior to %s)...".format(log.name,
new Date(cleanableHorizonMs), new Date(deleteHorizonMs)))
-    for (group <- groupSegmentsBySize(log.logSegments(0, endOffset), log.config.segmentSize,
log.config.maxIndexSize, cleanable.firstUncleanableOffset))
+    for (group <- groupSegmentsBySize(log.logSegments(0, endOffset), log.config.segmentSize,
log.config.maxIndexSize))
       cleanSegments(log, group, offsetMap, deleteHorizonMs, stats)
 
     // record buffer utilization
@@ -552,7 +551,7 @@ private[log] class Cleaner(val id: Int,
    *
    * @return A list of grouped segments
    */
-  private[log] def groupSegmentsBySize(segments: Iterable[LogSegment], maxSize: Int, maxIndexSize:
Int, firstUncleanableOffset: Long): List[Seq[LogSegment]] = {
+  private[log] def groupSegmentsBySize(segments: Iterable[LogSegment], maxSize: Int, maxIndexSize:
Int): List[Seq[LogSegment]] = {
     var grouped = List[List[LogSegment]]()
     var segs = segments.toList
     while(segs.nonEmpty) {
@@ -565,7 +564,7 @@ private[log] class Cleaner(val id: Int,
             logSize + segs.head.size <= maxSize &&
             indexSize + segs.head.index.sizeInBytes <= maxIndexSize &&
             timeIndexSize + segs.head.timeIndex.sizeInBytes <= maxIndexSize &&
-            lastOffsetForFirstSegment(segs, firstUncleanableOffset) - group.last.baseOffset
<= Int.MaxValue) {
+            segs.head.index.lastOffset - group.last.index.baseOffset <= Int.MaxValue)
{
         group = segs.head :: group
         logSize += segs.head.size
         indexSize += segs.head.index.sizeInBytes
@@ -578,28 +577,6 @@ private[log] class Cleaner(val id: Int,
   }
 
   /**
-    * We want to get the last offset in the first log segment in segs.
-    * LogSegment.nextOffset() gives the exact last offset in a segment, but can be expensive
since it requires
-    * scanning the segment from the last index entry.
-    * Therefore, we estimate the last offset of the first log segment by using
-    * the base offset of the next segment in the list.
-    * If the next segment doesn't exist, first Uncleanable Offset will be used.
-    *
-    * @param segs - remaining segments to group.
-    * @return The estimated last offset for the first segment in segs
-    */
-  private def lastOffsetForFirstSegment(segs: List[LogSegment], firstUncleanableOffset: Long):
Long = {
-    if (segs.size > 1) {
-      /* if there is a next segment, use its base offset as the bounding offset to guarantee
we know
-       * the worst case offset */
-      segs(1).baseOffset - 1
-    } else {
-      //for the last segment in the list, use the first uncleanable offset.
-      firstUncleanableOffset - 1
-    }
-  }
-
-  /**
    * Build a map of key_hash => offset for the keys in the cleanable dirty portion of
the log to use in cleaning.
    * @param log The log to use
    * @param start The offset at which dirty messages begin

http://git-wip-us.apache.org/repos/asf/kafka/blob/33727acf/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
index 529503a..40691b9 100755
--- a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
@@ -381,17 +381,17 @@ class LogCleanerTest extends JUnitSuite {
     }
 
     // grouping by very large values should result in a single group with all the segments
in it
-    var groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize
= Int.MaxValue, log.logEndOffset)
+    var groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize
= Int.MaxValue)
     assertEquals(1, groups.size)
     assertEquals(log.numberOfSegments, groups.head.size)
     checkSegmentOrder(groups)
 
     // grouping by very small values should result in all groups having one entry
-    groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = 1, maxIndexSize = Int.MaxValue,
log.logEndOffset)
+    groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = 1, maxIndexSize = Int.MaxValue)
     assertEquals(log.numberOfSegments, groups.size)
     assertTrue("All groups should be singletons.", groups.forall(_.size == 1))
     checkSegmentOrder(groups)
-    groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize
= 1, log.logEndOffset)
+    groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize
= 1)
     assertEquals(log.numberOfSegments, groups.size)
     assertTrue("All groups should be singletons.", groups.forall(_.size == 1))
     checkSegmentOrder(groups)
@@ -400,13 +400,13 @@ class LogCleanerTest extends JUnitSuite {
 
     // check grouping by log size
     val logSize = log.logSegments.take(groupSize).map(_.size).sum.toInt + 1
-    groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = logSize, maxIndexSize
= Int.MaxValue, log.logEndOffset)
+    groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = logSize, maxIndexSize
= Int.MaxValue)
     checkSegmentOrder(groups)
     assertTrue("All but the last group should be the target size.", groups.dropRight(1).forall(_.size
== groupSize))
 
     // check grouping by index size
     val indexSize = log.logSegments.take(groupSize).map(_.index.sizeInBytes).sum + 1
-    groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize
= indexSize, log.logEndOffset)
+    groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize
= indexSize)
     checkSegmentOrder(groups)
     assertTrue("All but the last group should be the target size.", groups.dropRight(1).forall(_.size
== groupSize))
   }
@@ -438,14 +438,14 @@ class LogCleanerTest extends JUnitSuite {
     assertEquals(Int.MaxValue, log.activeSegment.index.lastOffset)
 
     // grouping should result in a single group with maximum relative offset of Int.MaxValue
-    var groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize
= Int.MaxValue, log.logEndOffset)
+    var groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize
= Int.MaxValue)
     assertEquals(1, groups.size)
 
     // append another message, making last offset of second segment > Int.MaxValue
     log.append(TestUtils.singletonRecords(value = "hello".getBytes, key = "hello".getBytes))
 
     // grouping should not group the two segments to ensure that maximum relative offset
in each group <= Int.MaxValue
-    groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize
= Int.MaxValue, log.logEndOffset)
+    groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize
= Int.MaxValue)
     assertEquals(2, groups.size)
     checkSegmentOrder(groups)
 
@@ -453,55 +453,13 @@ class LogCleanerTest extends JUnitSuite {
     while (log.numberOfSegments < 4)
       log.append(TestUtils.singletonRecords(value = "hello".getBytes, key = "hello".getBytes))
 
-    groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize
= Int.MaxValue, log.logEndOffset)
+    groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize
= Int.MaxValue)
     assertEquals(log.numberOfSegments - 1, groups.size)
     for (group <- groups)
       assertTrue("Relative offset greater than Int.MaxValue", group.last.index.lastOffset
- group.head.index.baseOffset <= Int.MaxValue)
     checkSegmentOrder(groups)
   }
 
-  /** 
-   * Following the loading of a log segment where the index file is zero sized,
-   * the index returned would be the base offset.  Sometimes the log file would
-   * contain data with offsets in excess of the baseOffset which would cause
-   * the log cleaner to group together segments with a range of > Int.MaxValue
-   * this test replicates that scenario to ensure that the segments are grouped
-   * correctly.
-   */
-  @Test
-  def testSegmentGroupingFollowingLoadOfZeroIndex(): Unit = {
-    val cleaner = makeCleaner(Int.MaxValue)
-
-    val logProps = new Properties()
-    logProps.put(LogConfig.SegmentBytesProp, 400: java.lang.Integer)
-
-    //mimic the effect of loading an empty index file
-    logProps.put(LogConfig.IndexIntervalBytesProp, 400: java.lang.Integer)
-
-    val log = makeLog(config = LogConfig.fromProps(logConfig.originals, logProps))
-
-    val record1 = messageWithOffset("hello".getBytes, "hello".getBytes, 0)
-    log.appendAsFollower(record1)
-    val record2 = messageWithOffset("hello".getBytes, "hello".getBytes, 1)
-    log.appendAsFollower(record2)
-    log.roll(Int.MaxValue/2) // starting a new log segment at offset Int.MaxValue/2
-    val record3 = messageWithOffset("hello".getBytes, "hello".getBytes, Int.MaxValue/2)
-    log.appendAsFollower(record3)
-    val record4 = messageWithOffset("hello".getBytes, "hello".getBytes, Int.MaxValue.toLong
+ 1)
-    log.appendAsFollower(record4)
-
-    assertTrue("Actual offset range should be > Int.MaxValue", log.logEndOffset - 1 -
log.logStartOffset > Int.MaxValue)
-    assertTrue("index.lastOffset is reporting the wrong last offset", log.logSegments.last.index.lastOffset
- log.logStartOffset <= Int.MaxValue)
-
-    // grouping should result in two groups because the second segment takes the offset range
> MaxInt
-    val groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize
= Int.MaxValue, log.logEndOffset)
-    assertEquals(2, groups.size)
-
-    for (group <- groups)
-      assertTrue("Relative offset greater than Int.MaxValue", group.last.nextOffset() - 1
- group.head.baseOffset <= Int.MaxValue)
-    checkSegmentOrder(groups)
-  }
-
   private def checkSegmentOrder(groups: Seq[Seq[LogSegment]]): Unit = {
     val offsets = groups.flatMap(_.map(_.baseOffset))
     assertEquals("Offsets should be in increasing order.", offsets.sorted, offsets)


Mime
View raw message