[kafka] branch trunk updated: MINOR: Fix log cleaner offset range log message (#8435)

[jgus]

This is an automated email from the ASF dual-hosted git repository.

jgus pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 9d96523  MINOR: Fix log cleaner offset range log message (#8435)
9d96523 is described below

commit 9d965236a4459ddfaf333203cc4465902247926a
Author: Jason Gustafson <ja...@confluent.io>
AuthorDate: Mon Apr 6 17:11:12 2020 -0700

    MINOR: Fix log cleaner offset range log message (#8435)
    
    The upper limit offset is displayed incorrectly in the log cleaner summary 
message. For example:
    ```
    Log cleaner thread 0 cleaned log __consumer_offsets-47 (dirty section = 
[358800359, 358800359])
    ```
    We should be using the next dirty offset as the upper limit.
    
    Reviewers: David Arthur <mum...@gmail.com>
---
 core/src/main/scala/kafka/log/LogCleaner.scala | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/core/src/main/scala/kafka/log/LogCleaner.scala 
b/core/src/main/scala/kafka/log/LogCleaner.scala
index 8252151..2a20292 100644
--- a/core/src/main/scala/kafka/log/LogCleaner.scala
+++ b/core/src/main/scala/kafka/log/LogCleaner.scala
@@ -356,17 +356,18 @@ class LogCleaner(initialConfig: CleanerConfig,
     }
 
     private def cleanLog(cleanable: LogToClean): Unit = {
-      var endOffset = cleanable.firstDirtyOffset
+      val startOffset = cleanable.firstDirtyOffset
+      var endOffset = startOffset
       try {
         val (nextDirtyOffset, cleanerStats) = cleaner.clean(cleanable)
-        recordStats(cleaner.id, cleanable.log.name, 
cleanable.firstDirtyOffset, endOffset, cleanerStats)
         endOffset = nextDirtyOffset
+        recordStats(cleaner.id, cleanable.log.name, startOffset, endOffset, 
cleanerStats)
       } catch {
         case _: LogCleaningAbortedException => // task can be aborted, let it 
go.
         case _: KafkaStorageException => // partition is already offline. let 
it go.
         case e: IOException =>
           val logDirectory = cleanable.log.parentDir
-          val msg = s"Failed to clean up log for ${cleanable.topicPartition} 
in dir ${logDirectory} due to IOException"
+          val msg = s"Failed to clean up log for ${cleanable.topicPartition} 
in dir $logDirectory due to IOException"
           logDirFailureChannel.maybeAddOfflineLogDir(logDirectory, msg, e)
       } finally {
         cleanerManager.doneCleaning(cleanable.topicPartition, 
cleanable.log.parentDirFile, endOffset)