trunk/core/src/main/scala/kafka/log/LogManager.scala - kafka - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *    http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package kafka.log

 import java.io._
 import kafka.utils._
 import scala.actors.Actor
 import scala.collection._
 import java.util.concurrent.CountDownLatch
 import kafka.server.{KafkaConfig, KafkaZooKeeper}
 import kafka.common.{InvalidTopicException, InvalidPartitionException}
 import kafka.api.OffsetRequest

 /**
  * The guy who creates and hands out logs
  */
 @threadsafe
 private[kafka] class LogManager(val config: KafkaConfig,
                                 private val scheduler: KafkaScheduler,
                                 private val time: Time,
                                 val logRollDefaultIntervalMs: Long,
                                 val logCleanupIntervalMs: Long,
                                 val logCleanupDefaultAgeMs: Long,
                                 needRecovery: Boolean) extends Logging {

   val logDir: File = new File(config.logDir)
   private val numPartitions = config.numPartitions
   private val logFileSizeMap = config.logFileSizeMap
   private val flushInterval = config.flushInterval
   private val topicPartitionsMap = config.topicPartitionsMap
   private val logCreationLock = new Object
   private val random = new java.util.Random
   private var kafkaZookeeper: KafkaZooKeeper = null
   private var zkActor: Actor = null
   private val startupLatch: CountDownLatch = if (config.enableZookeeper) new CountDownLatch(1) else null
   private val logFlusherScheduler = new KafkaScheduler(1, "kafka-logflusher-", false)
   private val topicNameValidator = new TopicNameValidator(config)
   private val logFlushIntervalMap = config.flushIntervalMap
   private val logRetentionSizeMap = config.logRetentionSizeMap
   private val logRetentionMsMap = getMsMap(config.logRetentionHoursMap)
   private val logRollMsMap = getMsMap(config.logRollHoursMap)

   /* Initialize a log for each subdirectory of the main log directory */
   private val logs = new Pool[String, Pool[Int, Log]]()
   if(!logDir.exists()) {
     info("No log directory found, creating '" + logDir.getAbsolutePath() + "'")
     logDir.mkdirs()
   }
   if(!logDir.isDirectory() || !logDir.canRead())
     throw new IllegalArgumentException(logDir.getAbsolutePath() + " is not a readable log directory.")
   val subDirs = logDir.listFiles()
   if(subDirs != null) {
     for(dir <- subDirs) {
       if(!dir.isDirectory()) {
         warn("Skipping unexplainable file '" + dir.getAbsolutePath() + "'--should it be there?")
       } else {
         info("Loading log '" + dir.getName() + "'")
         val topic = Utils.getTopicPartition(dir.getName)._1
         val rollIntervalMs = logRollMsMap.get(topic).getOrElse(this.logRollDefaultIntervalMs)
         val maxLogFileSize = logFileSizeMap.get(topic).getOrElse(config.logFileSize)
         val log = new Log(dir, time, maxLogFileSize, config.maxMessageSize, flushInterval, rollIntervalMs, needRecovery)
         val topicPartion = Utils.getTopicPartition(dir.getName)
         logs.putIfNotExists(topicPartion._1, new Pool[Int, Log]())
         val parts = logs.get(topicPartion._1)
         parts.put(topicPartion._2, log)
       }
     }
   }

   /* Schedule the cleanup task to delete old logs */
   if(scheduler != null) {
     info("starting log cleaner every " + logCleanupIntervalMs + " ms")
     scheduler.scheduleWithRate(cleanupLogs, 60 * 1000, logCleanupIntervalMs)
   }

   if(config.enableZookeeper) {
     kafkaZookeeper = new KafkaZooKeeper(config, this)
     kafkaZookeeper.startup
     zkActor = new Actor {
       def act() {
         loop {
           receive {
             case topic: String =>
               try {
                 kafkaZookeeper.registerTopicInZk(topic)
               }
               catch {
                 case e => error(e) // log it and let it go
               }
             case StopActor =>
               info("zkActor stopped")
               exit
           }
         }
       }
     }
     zkActor.start
   }

   case object StopActor

   private def getMsMap(hoursMap: Map[String, Int]) : Map[String, Long] = {
     var ret = new mutable.HashMap[String, Long]
     for ( (topic, hour) <- hoursMap ) {
       ret.put(topic, hour * 60 * 60 * 1000L)
     }
     ret
   }

   /**
    *  Register this broker in ZK for the first time.
    */
   def startup() {
     if(config.enableZookeeper) {
       kafkaZookeeper.registerBrokerInZk()
       for (topic <- getAllTopics)
         kafkaZookeeper.registerTopicInZk(topic)
       startupLatch.countDown
     }
     info("Starting log flusher every " + config.flushSchedulerThreadRate + " ms with the following overrides " + logFlushIntervalMap)
     logFlusherScheduler.scheduleWithRate(flushAllLogs, config.flushSchedulerThreadRate, config.flushSchedulerThreadRate)
   }

   private def awaitStartup() {
     if (config.enableZookeeper)
       startupLatch.await
   }

   private def registerNewTopicInZK(topic: String) {
     if (config.enableZookeeper)
       zkActor ! topic
   }

   /**
    * Create a log for the given topic and the given partition
    */
   private def createLog(topic: String, partition: Int): Log = {
     logCreationLock synchronized {
       val d = new File(logDir, topic + "-" + partition)
       d.mkdirs()
       val rollIntervalMs = logRollMsMap.get(topic).getOrElse(this.logRollDefaultIntervalMs)
       val maxLogFileSize = logFileSizeMap.get(topic).getOrElse(config.logFileSize)
       new Log(d, time, maxLogFileSize, config.maxMessageSize, flushInterval, rollIntervalMs, false)
     }
   }

   /**
    * Return the Pool (partitions) for a specific log
    */
   private def getLogPool(topic: String, partition: Int): Pool[Int, Log] = {
     awaitStartup
     topicNameValidator.validate(topic)
     if (partition < 0 || partition >= topicPartitionsMap.getOrElse(topic, numPartitions)) {
       warn("Wrong partition " + partition + " valid partitions (0," +
               (topicPartitionsMap.getOrElse(topic, numPartitions) - 1) + ")")
       throw new InvalidPartitionException("wrong partition " + partition)
     }
     logs.get(topic)
   }

   /**
    * Pick a random partition from the given topic
    */
   def chooseRandomPartition(topic: String): Int = {
     random.nextInt(topicPartitionsMap.getOrElse(topic, numPartitions))
   }

   def getOffsets(offsetRequest: OffsetRequest): Array[Long] = {
     val log = getLog(offsetRequest.topic, offsetRequest.partition)
     if (log != null) return log.getOffsetsBefore(offsetRequest)
     Log.getEmptyOffsets(offsetRequest)
   }

   /**
    * Get the log if exists
    */
   def getLog(topic: String, partition: Int): Log = {
     val parts = getLogPool(topic, partition)
     if (parts == null) return null
     parts.get(partition)
   }

   /**
    * Create the log if it does not exist, if it exists just return it
    */
   def getOrCreateLog(topic: String, partition: Int): Log = {
     var hasNewTopic = false
     var parts = getLogPool(topic, partition)
     if (parts == null) {
       val found = logs.putIfNotExists(topic, new Pool[Int, Log])
       if (found == null)
         hasNewTopic = true
       parts = logs.get(topic)
     }
     var log = parts.get(partition)
     if(log == null) {
       log = createLog(topic, partition)
       val found = parts.putIfNotExists(partition, log)
       if(found != null) {
         // there was already somebody there
         log.close()
         log = found
       }
       else
         info("Created log for '" + topic + "'-" + partition)
     }

     if (hasNewTopic)
       registerNewTopicInZK(topic)
     log
   }

   /* Attemps to delete all provided segments from a log and returns how many it was able to */
   private def deleteSegments(log: Log, segments: Seq[LogSegment]): Int = {
     var total = 0
     for(segment <- segments) {
       info("Deleting log segment " + segment.file.getName() + " from " + log.name)
       Utils.swallow(logger.warn, segment.messageSet.close())
       if(!segment.file.delete()) {
         warn("Delete failed.")
       } else {
         total += 1
       }
     }
     total
   }

   /* Runs through the log removing segments older than a certain age */
   private def cleanupExpiredSegments(log: Log): Int = {
     val startMs = time.milliseconds
     val topic = Utils.getTopicPartition(log.dir.getName)._1
     val logCleanupThresholdMS = logRetentionMsMap.get(topic).getOrElse(this.logCleanupDefaultAgeMs)
     val toBeDeleted = log.markDeletedWhile(startMs - _.file.lastModified > logCleanupThresholdMS)
     val total = deleteSegments(log, toBeDeleted)
     total
   }

   /**
    *  Runs through the log removing segments until the size of the log
    *  is at least logRetentionSize bytes in size
    */
   private def cleanupSegmentsToMaintainSize(log: Log): Int = {
     val topic = Utils.getTopicPartition(log.dir.getName)._1
     val maxLogRetentionSize = logRetentionSizeMap.get(topic).getOrElse(config.logRetentionSize)
     if(maxLogRetentionSize < 0 || log.size < maxLogRetentionSize) return 0
     var diff = log.size - maxLogRetentionSize
     def shouldDelete(segment: LogSegment) = {
       if(diff - segment.size >= 0) {
         diff -= segment.size
         true
       } else {
         false
       }
     }
     val toBeDeleted = log.markDeletedWhile( shouldDelete )
     val total = deleteSegments(log, toBeDeleted)
     total
   }

   /**
    * Delete any eligible logs. Return the number of segments deleted.
    */
   def cleanupLogs() {
     debug("Beginning log cleanup...")
     val iter = getLogIterator
     var total = 0
     val startMs = time.milliseconds
     while(iter.hasNext) {
       val log = iter.next
       debug("Garbage collecting '" + log.name + "'")
       total += cleanupExpiredSegments(log) + cleanupSegmentsToMaintainSize(log)
     }
     debug("Log cleanup completed. " + total + " files deleted in " +
                  (time.milliseconds - startMs) / 1000 + " seconds")
   }

   /**
    * Close all the logs
    */
   def close() {
     logFlusherScheduler.shutdown()
     val iter = getLogIterator
     while(iter.hasNext)
       iter.next.close()
     if (config.enableZookeeper) {
       zkActor ! StopActor
       kafkaZookeeper.close
     }
   }

   private def getLogIterator(): Iterator[Log] = {
     new IteratorTemplate[Log] {
       val partsIter = logs.values.iterator
       var logIter: Iterator[Log] = null

       override def makeNext(): Log = {
         while (true) {
           if (logIter != null && logIter.hasNext)
             return logIter.next
           if (!partsIter.hasNext)
             return allDone
           logIter = partsIter.next.values.iterator
         }
         // should never reach here
         assert(false)
         return allDone
       }
     }
   }

   private def flushAllLogs() = {
     debug("flushing the high watermark of all logs")

     for (log <- getLogIterator)
     {
       try{
         val timeSinceLastFlush = System.currentTimeMillis - log.getLastFlushedTime
         var logFlushInterval = config.defaultFlushIntervalMs
         if(logFlushIntervalMap.contains(log.getTopicName))
           logFlushInterval = logFlushIntervalMap(log.getTopicName)
         debug(log.getTopicName + " flush interval  " + logFlushInterval +
             " last flushed " + log.getLastFlushedTime + " timesincelastFlush: " + timeSinceLastFlush)
         if(timeSinceLastFlush >= logFlushInterval)
           log.flush
       }
       catch {
         case e =>
           error("Error flushing topic " + log.getTopicName, e)
           e match {
             case _: IOException =>
               fatal("Halting due to unrecoverable I/O error while flushing logs: " + e.getMessage, e)
               Runtime.getRuntime.halt(1)
             case _ =>
           }
       }
     }
   }


   def getAllTopics(): Iterator[String] = logs.keys.iterator
   def getTopicPartitionsMap() = topicPartitionsMap
 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package kafka.log

	import java.io._
	import kafka.utils._
	import scala.actors.Actor
	import scala.collection._
	import java.util.concurrent.CountDownLatch
	import kafka.server.{KafkaConfig, KafkaZooKeeper}
	import kafka.common.{InvalidTopicException, InvalidPartitionException}
	import kafka.api.OffsetRequest

	/**
	* The guy who creates and hands out logs
	*/
	@threadsafe
	private[kafka] class LogManager(val config: KafkaConfig,
	private val scheduler: KafkaScheduler,
	private val time: Time,
	val logRollDefaultIntervalMs: Long,
	val logCleanupIntervalMs: Long,
	val logCleanupDefaultAgeMs: Long,
	needRecovery: Boolean) extends Logging {

	val logDir: File = new File(config.logDir)
	private val numPartitions = config.numPartitions
	private val logFileSizeMap = config.logFileSizeMap
	private val flushInterval = config.flushInterval
	private val topicPartitionsMap = config.topicPartitionsMap
	private val logCreationLock = new Object
	private val random = new java.util.Random
	private var kafkaZookeeper: KafkaZooKeeper = null
	private var zkActor: Actor = null
	private val startupLatch: CountDownLatch = if (config.enableZookeeper) new CountDownLatch(1) else null
	private val logFlusherScheduler = new KafkaScheduler(1, "kafka-logflusher-", false)
	private val topicNameValidator = new TopicNameValidator(config)
	private val logFlushIntervalMap = config.flushIntervalMap
	private val logRetentionSizeMap = config.logRetentionSizeMap
	private val logRetentionMsMap = getMsMap(config.logRetentionHoursMap)
	private val logRollMsMap = getMsMap(config.logRollHoursMap)

	/* Initialize a log for each subdirectory of the main log directory */
	private val logs = new Pool[String, Pool[Int, Log]]()
	if(!logDir.exists()) {
	info("No log directory found, creating '" + logDir.getAbsolutePath() + "'")
	logDir.mkdirs()
	}
	if(!logDir.isDirectory() \|\| !logDir.canRead())
	throw new IllegalArgumentException(logDir.getAbsolutePath() + " is not a readable log directory.")
	val subDirs = logDir.listFiles()
	if(subDirs != null) {
	for(dir <- subDirs) {
	if(!dir.isDirectory()) {
	warn("Skipping unexplainable file '" + dir.getAbsolutePath() + "'--should it be there?")
	} else {
	info("Loading log '" + dir.getName() + "'")
	val topic = Utils.getTopicPartition(dir.getName)._1
	val rollIntervalMs = logRollMsMap.get(topic).getOrElse(this.logRollDefaultIntervalMs)
	val maxLogFileSize = logFileSizeMap.get(topic).getOrElse(config.logFileSize)
	val log = new Log(dir, time, maxLogFileSize, config.maxMessageSize, flushInterval, rollIntervalMs, needRecovery)
	val topicPartion = Utils.getTopicPartition(dir.getName)
	logs.putIfNotExists(topicPartion._1, new Pool[Int, Log]())
	val parts = logs.get(topicPartion._1)
	parts.put(topicPartion._2, log)
	}
	}
	}

	/* Schedule the cleanup task to delete old logs */
	if(scheduler != null) {
	info("starting log cleaner every " + logCleanupIntervalMs + " ms")
	scheduler.scheduleWithRate(cleanupLogs, 60 * 1000, logCleanupIntervalMs)
	}

	if(config.enableZookeeper) {
	kafkaZookeeper = new KafkaZooKeeper(config, this)
	kafkaZookeeper.startup
	zkActor = new Actor {
	def act() {
	loop {
	receive {
	case topic: String =>
	try {
	kafkaZookeeper.registerTopicInZk(topic)
	}
	catch {
	case e => error(e) // log it and let it go
	}
	case StopActor =>
	info("zkActor stopped")
	exit
	}
	}
	}
	}
	zkActor.start
	}

	case object StopActor

	private def getMsMap(hoursMap: Map[String, Int]) : Map[String, Long] = {
	var ret = new mutable.HashMap[String, Long]
	for ( (topic, hour) <- hoursMap ) {
	ret.put(topic, hour * 60 * 60 * 1000L)
	}
	ret
	}

	/**
	* Register this broker in ZK for the first time.
	*/
	def startup() {
	if(config.enableZookeeper) {
	kafkaZookeeper.registerBrokerInZk()
	for (topic <- getAllTopics)
	kafkaZookeeper.registerTopicInZk(topic)
	startupLatch.countDown
	}
	info("Starting log flusher every " + config.flushSchedulerThreadRate + " ms with the following overrides " + logFlushIntervalMap)
	logFlusherScheduler.scheduleWithRate(flushAllLogs, config.flushSchedulerThreadRate, config.flushSchedulerThreadRate)
	}

	private def awaitStartup() {
	if (config.enableZookeeper)
	startupLatch.await
	}

	private def registerNewTopicInZK(topic: String) {
	if (config.enableZookeeper)
	zkActor ! topic
	}

	/**
	* Create a log for the given topic and the given partition
	*/
	private def createLog(topic: String, partition: Int): Log = {
	logCreationLock synchronized {
	val d = new File(logDir, topic + "-" + partition)
	d.mkdirs()
	val rollIntervalMs = logRollMsMap.get(topic).getOrElse(this.logRollDefaultIntervalMs)
	val maxLogFileSize = logFileSizeMap.get(topic).getOrElse(config.logFileSize)
	new Log(d, time, maxLogFileSize, config.maxMessageSize, flushInterval, rollIntervalMs, false)
	}
	}

	/**
	* Return the Pool (partitions) for a specific log
	*/
	private def getLogPool(topic: String, partition: Int): Pool[Int, Log] = {
	awaitStartup
	topicNameValidator.validate(topic)
	if (partition < 0 \|\| partition >= topicPartitionsMap.getOrElse(topic, numPartitions)) {
	warn("Wrong partition " + partition + " valid partitions (0," +
	(topicPartitionsMap.getOrElse(topic, numPartitions) - 1) + ")")
	throw new InvalidPartitionException("wrong partition " + partition)
	}
	logs.get(topic)
	}

	/**
	* Pick a random partition from the given topic
	*/
	def chooseRandomPartition(topic: String): Int = {
	random.nextInt(topicPartitionsMap.getOrElse(topic, numPartitions))
	}

	def getOffsets(offsetRequest: OffsetRequest): Array[Long] = {
	val log = getLog(offsetRequest.topic, offsetRequest.partition)
	if (log != null) return log.getOffsetsBefore(offsetRequest)
	Log.getEmptyOffsets(offsetRequest)
	}

	/**
	* Get the log if exists
	*/
	def getLog(topic: String, partition: Int): Log = {
	val parts = getLogPool(topic, partition)
	if (parts == null) return null
	parts.get(partition)
	}

	/**
	* Create the log if it does not exist, if it exists just return it
	*/
	def getOrCreateLog(topic: String, partition: Int): Log = {
	var hasNewTopic = false
	var parts = getLogPool(topic, partition)
	if (parts == null) {
	val found = logs.putIfNotExists(topic, new Pool[Int, Log])
	if (found == null)
	hasNewTopic = true
	parts = logs.get(topic)
	}
	var log = parts.get(partition)
	if(log == null) {
	log = createLog(topic, partition)
	val found = parts.putIfNotExists(partition, log)
	if(found != null) {
	// there was already somebody there
	log.close()
	log = found
	}
	else
	info("Created log for '" + topic + "'-" + partition)
	}

	if (hasNewTopic)
	registerNewTopicInZK(topic)
	log
	}

	/* Attemps to delete all provided segments from a log and returns how many it was able to */
	private def deleteSegments(log: Log, segments: Seq[LogSegment]): Int = {
	var total = 0
	for(segment <- segments) {
	info("Deleting log segment " + segment.file.getName() + " from " + log.name)
	Utils.swallow(logger.warn, segment.messageSet.close())
	if(!segment.file.delete()) {
	warn("Delete failed.")
	} else {
	total += 1
	}
	}
	total
	}

	/* Runs through the log removing segments older than a certain age */
	private def cleanupExpiredSegments(log: Log): Int = {
	val startMs = time.milliseconds
	val topic = Utils.getTopicPartition(log.dir.getName)._1
	val logCleanupThresholdMS = logRetentionMsMap.get(topic).getOrElse(this.logCleanupDefaultAgeMs)
	val toBeDeleted = log.markDeletedWhile(startMs - _.file.lastModified > logCleanupThresholdMS)
	val total = deleteSegments(log, toBeDeleted)
	total
	}

	/**
	* Runs through the log removing segments until the size of the log
	* is at least logRetentionSize bytes in size
	*/
	private def cleanupSegmentsToMaintainSize(log: Log): Int = {
	val topic = Utils.getTopicPartition(log.dir.getName)._1
	val maxLogRetentionSize = logRetentionSizeMap.get(topic).getOrElse(config.logRetentionSize)
	if(maxLogRetentionSize < 0 \|\| log.size < maxLogRetentionSize) return 0
	var diff = log.size - maxLogRetentionSize
	def shouldDelete(segment: LogSegment) = {
	if(diff - segment.size >= 0) {
	diff -= segment.size
	true
	} else {
	false
	}
	}
	val toBeDeleted = log.markDeletedWhile( shouldDelete )
	val total = deleteSegments(log, toBeDeleted)
	total
	}

	/**
	* Delete any eligible logs. Return the number of segments deleted.
	*/
	def cleanupLogs() {
	debug("Beginning log cleanup...")
	val iter = getLogIterator
	var total = 0
	val startMs = time.milliseconds
	while(iter.hasNext) {
	val log = iter.next
	debug("Garbage collecting '" + log.name + "'")
	total += cleanupExpiredSegments(log) + cleanupSegmentsToMaintainSize(log)
	}
	debug("Log cleanup completed. " + total + " files deleted in " +
	(time.milliseconds - startMs) / 1000 + " seconds")
	}

	/**
	* Close all the logs
	*/
	def close() {
	logFlusherScheduler.shutdown()
	val iter = getLogIterator
	while(iter.hasNext)
	iter.next.close()
	if (config.enableZookeeper) {
	zkActor ! StopActor
	kafkaZookeeper.close
	}
	}

	private def getLogIterator(): Iterator[Log] = {
	new IteratorTemplate[Log] {
	val partsIter = logs.values.iterator
	var logIter: Iterator[Log] = null

	override def makeNext(): Log = {
	while (true) {
	if (logIter != null && logIter.hasNext)
	return logIter.next
	if (!partsIter.hasNext)
	return allDone
	logIter = partsIter.next.values.iterator
	}
	// should never reach here
	assert(false)
	return allDone
	}
	}
	}

	private def flushAllLogs() = {
	debug("flushing the high watermark of all logs")

	for (log <- getLogIterator)
	{
	try{
	val timeSinceLastFlush = System.currentTimeMillis - log.getLastFlushedTime
	var logFlushInterval = config.defaultFlushIntervalMs
	if(logFlushIntervalMap.contains(log.getTopicName))
	logFlushInterval = logFlushIntervalMap(log.getTopicName)
	debug(log.getTopicName + " flush interval " + logFlushInterval +
	" last flushed " + log.getLastFlushedTime + " timesincelastFlush: " + timeSinceLastFlush)
	if(timeSinceLastFlush >= logFlushInterval)
	log.flush
	}
	catch {
	case e =>
	error("Error flushing topic " + log.getTopicName, e)
	e match {
	case _: IOException =>
	fatal("Halting due to unrecoverable I/O error while flushing logs: " + e.getMessage, e)
	Runtime.getRuntime.halt(1)
	case _ =>
	}
	}
	}
	}


	def getAllTopics(): Iterator[String] = logs.keys.iterator
	def getTopicPartitionsMap() = topicPartitionsMap
	}