Repository: kafka Updated Branches: refs/heads/trunk 9180cb23d -> 2d154420c (forced update)
KAFKA-3123; Ensure cleaning is resumed if truncateTo throws Author: Mickael Maison <mickael.mai...@gmail.com> Reviewers: Ismael Juma <ism...@juma.me.uk> Closes #3296 from mimaison/KAFKA-3123 Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/2d154420 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/2d154420 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/2d154420 Branch: refs/heads/trunk Commit: 2d154420c54f725988a7ca696febe341f625049a Parents: 7d1ef63 Author: Mickael Maison <mickael.mai...@gmail.com> Authored: Mon Jun 12 22:36:47 2017 +0100 Committer: Ismael Juma <ism...@juma.me.uk> Committed: Mon Jun 12 22:38:08 2017 +0100 ---------------------------------------------------------------------- core/src/main/scala/kafka/log/LogManager.scala | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/2d154420/core/src/main/scala/kafka/log/LogManager.scala ---------------------------------------------------------------------- diff --git a/core/src/main/scala/kafka/log/LogManager.scala b/core/src/main/scala/kafka/log/LogManager.scala index d8cdf90..61879be 100755 --- a/core/src/main/scala/kafka/log/LogManager.scala +++ b/core/src/main/scala/kafka/log/LogManager.scala @@ -324,13 +324,16 @@ class LogManager(val logDirs: Array[File], // If the log does not exist, skip it if (log != null) { //May need to abort and pause the cleaning of the log, and resume after truncation is done. - val needToStopCleaner: Boolean = truncateOffset < log.activeSegment.baseOffset - if (needToStopCleaner && cleaner != null) + val needToStopCleaner = cleaner != null && truncateOffset < log.activeSegment.baseOffset + if (needToStopCleaner) cleaner.abortAndPauseCleaning(topicPartition) - log.truncateTo(truncateOffset) - if (needToStopCleaner && cleaner != null) { - cleaner.maybeTruncateCheckpoint(log.dir.getParentFile, topicPartition, log.activeSegment.baseOffset) - cleaner.resumeCleaning(topicPartition) + try { + log.truncateTo(truncateOffset) + if (needToStopCleaner) + cleaner.maybeTruncateCheckpoint(log.dir.getParentFile, topicPartition, log.activeSegment.baseOffset) + } finally { + if (needToStopCleaner) + cleaner.resumeCleaning(topicPartition) } } }
