[
https://issues.apache.org/jira/browse/KAFKA-1414?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14063742#comment-14063742
]
Jay Kreps commented on KAFKA-1414:
----------------------------------
Great patch!
I second Jun's comments.
Is there a way we can make this a little better out-of-the-box? Our experience
is that people don't set every config optimally out of the box. So a couple
ideas:
It may be that having more threads than disks doesn't hurt as the I/O scheduler
can figure it out, in which case we could default to something higher than 1.
Is there really a need for two configurations, one for recovery and one for
shutdown? It may be that the ideal setting is always just the number of disks.
Or it may be that the recovery is actually CPU bound whereas flush isn't. If
the right answer is just the number of disks then we may be able to simplify
things by just asking for "num.disks" or "io.parallelism" or something like
that and we set the number of threads appropriately for these cases. This would
let us default num.disks to the number of data directories, but would let users
with RAID arrays or other logical volume mgmt things override this. It would
also allow us to use this config for other things related to I/O parallelism as
they arise without creating to many configs.
This is obviously a minor thing but a little research here would give a nicer
user experience.
Also, if you have a chance to run a basic before/after test on the speedup on a
machine with multiple disks that would be great (presumably it is a linear
speedup). This would let us report the improvement in the release in concrete
terms as well as validating that it actually works as we expect.
> Speedup broker startup after hard reset
> ---------------------------------------
>
> Key: KAFKA-1414
> URL: https://issues.apache.org/jira/browse/KAFKA-1414
> Project: Kafka
> Issue Type: Improvement
> Components: log
> Affects Versions: 0.8.2, 0.9.0, 0.8.1.1
> Reporter: Dmitry Bugaychenko
> Assignee: Jay Kreps
> Attachments:
> 0001-KAFKA-1414-Speedup-broker-startup-after-hard-reset-a.patch,
> parallel-dir-loading-0.8.patch,
> parallel-dir-loading-trunk-fixed-threadpool.patch,
> parallel-dir-loading-trunk-threadpool.patch, parallel-dir-loading-trunk.patch
>
>
> After hard reset due to power failure broker takes way too much time
> recovering unflushed segments in a single thread. This could be easiliy
> improved launching multiple threads (one per data dirrectory, assuming that
> typically each data directory is on a dedicated drive). Localy we trie this
> simple patch to LogManager.loadLogs and it seems to work, however I'm too new
> to scala, so do not take it literally:
> {code}
> /**
> * Recover and load all logs in the given data directories
> */
> private def loadLogs(dirs: Seq[File]) {
> val threads : Array[Thread] = new Array[Thread](dirs.size)
> var i: Int = 0
> val me = this
> for(dir <- dirs) {
> val thread = new Thread( new Runnable {
> def run()
> {
> val recoveryPoints = me.recoveryPointCheckpoints(dir).read
> /* load the logs */
> val subDirs = dir.listFiles()
> if(subDirs != null) {
> val cleanShutDownFile = new File(dir, Log.CleanShutdownFile)
> if(cleanShutDownFile.exists())
> info("Found clean shutdown file. Skipping recovery for all logs
> in data directory '%s'".format(dir.getAbsolutePath))
> for(dir <- subDirs) {
> if(dir.isDirectory) {
> info("Loading log '" + dir.getName + "'")
> val topicPartition = Log.parseTopicPartitionName(dir.getName)
> val config = topicConfigs.getOrElse(topicPartition.topic,
> defaultConfig)
> val log = new Log(dir,
> config,
> recoveryPoints.getOrElse(topicPartition, 0L),
> scheduler,
> time)
> val previous = addLogWithLock(topicPartition, log)
> if(previous != null)
> throw new IllegalArgumentException("Duplicate log
> directories found: %s, %s!".format(log.dir.getAbsolutePath,
> previous.dir.getAbsolutePath))
> }
> }
> cleanShutDownFile.delete()
> }
> }
> })
> thread.start()
> threads(i) = thread
> i = i + 1
> }
> for(thread <- threads) {
> thread.join()
> }
> }
> def addLogWithLock(topicPartition: TopicAndPartition, log: Log): Log = {
> logCreationOrDeletionLock synchronized {
> this.logs.put(topicPartition, log)
> }
> }
> {code}
--
This message was sent by Atlassian JIRA
(v6.2#6252)
