Kyle Ambroff created KAFKA-5297:
-----------------------------------
Summary: Broker can take a long time to shut down if there are
many active log segments
Key: KAFKA-5297
URL: https://issues.apache.org/jira/browse/KAFKA-5297
Project: Kafka
Issue Type: Improvement
Reporter: Kyle Ambroff
Priority: Minor
Attachments: shutdown-flame-graph.png
After the changes for KIP-33 were merged, we started noticing that our cluster
restart times were quite a bit longer. In some cases it was taking four times
as long as expected to do a rolling restart of every broker in the cluster.
This meant that doing a deploy to one of our Kafka clusters went from taking
about 3 hours to more than 12 hours!
We looked into this and we have some data from a couple of runs with a sampling
profiler. It turns out that it isn't unusual for us to have a broker sit in
kafka.log.Log#close for up to 30 minutes if it has been running for several
weeks. There are just so many active log segments that it just takes a long
time to truncate all of the indexes.
I've attached a flame graph that was generated from 10 minutes of stack samples
collected during shutdown of a broker that took about 30 minutes total to shut
down cleanly.
* About 60% of the time was spent in kafka.log.AbstractIndex#resize, where
every index and timeindex file is truncated to the size of the number of
entries in that index.
* Another big chunk of time is spent reading the last entry from the index,
which is used to make any final updates to the timeindex file. This is
something that can be cached. For a broker that's been running for a long time
the bulk of these indexes are not likely to be in the page cache anymore. We
cache the largestTimestamp and offsetOfLargestTimestamp in LogSegment, so we
could add a cache for this as well.
Looking at these changes and considering KIP-33, it isn't surprising that the
broker shutdown time has increased so dramatically. The extra index plus the
extra reads have increased the amount of work performed by kafka.log.Log#close
by about 4x (in terms of system calls and potential page faults). Breaking down
what this function does:
# Read the max timestamp from the timeindex. Could lead to a disk seek.
# Read the max offset from the index. Could lead to a disk seek.
# Append the timestamp and offset of the most recently written message to the
timeindex if it hasn't been written there for some reason.
# Truncate the index file
## Get the position in the index of the last entry written
## If on Windows then unmap and close the index
## reopen
## truncate to the number of entries * entry size. (ftruncate() system call)
## mmap()
## Set the position back to where it was before the original. Leads to lseek()
system call.
## Close the newly reopenned and mapped index
# Same thing as #4 but for the timeindex.
## Get the position in the timeindex of the last entry written
## If on Windows then unmap and close the timeindex
## reopen
## truncate to the number of entries * entry size. (ftruncate() system call)
## mmap()
## Set the position back to where it was before the original. Leads to lseek()
system call.
## Close the newly reopenned and mapped timeindex
# Finalize the log segment
## Invoke java.nio.channels.FileChannel#force, which leads to a fsync() for
that log segment.
## Truncate the log segment if it doesn't have enough messages written to fill
up the whole thing. Potentially leads to a ftruncate() system call.
## Set the position to the end of the segment after truncation. Leads to a
lseek() system call.
## Close and unmap the channel.
Looking in to the current implementation of kafka.log.AbstractIndex#resize, it
appears to do quite a bit of extra work to avoid keeping an instance of
RandomAccessFile around. It has to reopen the file, truncate, mmap(),
potentially perform an additional disk seek, all before imediately closing the
file.
You wouldn't think this would amount to much, but I put together a benchmark
using jmh to measure the difference between the current code and a new
implementation that didn't have to recreate the page mapping during resize(),
and the difference is pretty dramatic.
{noformat}
Result "currentImplementation":
2063.386 ±(99.9%) 81.758 ops/s [Average]
(min, avg, max) = (1685.574, 2063.386, 2338.945), stdev = 182.863
CI (99.9%): [1981.628, 2145.144] (assumes normal distribution)
Result "optimizedImplementation":
3497.354 ±(99.9%) 31.575 ops/s [Average]
(min, avg, max) = (3261.232, 3497.354, 3605.527), stdev = 70.623
CI (99.9%): [3465.778, 3528.929] (assumes normal distribution)
# Run complete. Total time: 00:03:37
Benchmark Mode Cnt Score Error
Units
LogSegmentBenchmark.currentImplementation thrpt 60 2063.386 ± 81.758
ops/s
LogSegmentBenchmark.optimizedImplementation thrpt 60 3497.354 ± 31.575
ops/s
{noformat}
I ran this benchmark on a Linux workstation. It just measures the throughput of
Log#close after 20 segments have been created. Not having to reopen the file
amounts to a 70% increase in throughput.
I think there are two totally valid approaches to making this better:
* Premptively truncate index files when log rotation happens. Once a log is
rotated, jobs could be added to an ExecutorService which truncates indexes so
that they don't all have to be truncated on shutdown. The new shutdown code
would enqueue all remaining active indexes and then drain the queue.
* Alternatively we could just add a RandomAccessFile instance variable to
AbstractIndex so that it doesn't have to recreate the page mapping on resize().
This means an extra file handle for each segment but that doesn't seem like a
big deal to me.
No matter what we should add a cache for kafka.log.OffsetIndex#lastEntry.
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
