Lucas Wang created KAFKA-14213:
----------------------------------
Summary: Reduce lock contention between
control-plane-kafka-request-handler and kafka-log-cleaner-thread
Key: KAFKA-14213
URL: https://issues.apache.org/jira/browse/KAFKA-14213
Project: Kafka
Issue Type: Improvement
Reporter: Lucas Wang
We found that the StopReplica request's local processing time may be quite long
due to the reasons explained below. The impact of a time-consuming StopReplica
request is that all subsequent requests from the controller are blocked,
causing slow convergence on the metadata plane.
The long local processing time is because the
control-plane-kafka-request-handler thread is blocked on a lock to abort
logCleaning with the following stack trace:
{code:java}
"control-plane-kafka-request-handler-0"
java.lang.Thread.State: WAITING at
java.base@11.0.13/jdk.internal.misc.Unsafe.park(Native Method) at
java.base@11.0.13/java.util.concurrent.locks.LockSupport.park(LockSupport.java:194)
at
java.base@11.0.13/java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:885)
at
java.base@11.0.13/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireQueued(AbstractQueuedSynchronizer.java:917)
at
java.base@11.0.13/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:1240)
at
java.base@11.0.13/java.util.concurrent.locks.ReentrantLock.lock(ReentrantLock.java:267)
at kafka.log.LogCleanerManager.abortCleaning(LogCleanerManager.scala:266) at
kafka.log.LogCleaner.abortCleaning(LogCleaner.scala:205) at
kafka.log.LogManager.asyncDelete(LogManager.scala:1039)
{code}
In the mean time, the kafka-log-cleaner-thread is holding the lock and busy
figuring out the filthiest compacted log, which can be a very time consuming
task. Our periodic thread dumps captured many snapshots with the
kafka-log-cleaner-thread in either of the following 2 stack traces:
{code:java}
"kafka-log-cleaner-thread-0"
java.lang.Thread.State: RUNNABLE at
kafka.log.TimeIndex.<init>(TimeIndex.scala:57) at
kafka.log.LazyIndex$.$anonfun$forTime$1(LazyIndex.scala:109) at
kafka.log.LazyIndex$$$Lambda$1871/0x0000000800f4d840.apply(Unknown Source) at
kafka.log.LazyIndex.$anonfun$get$1(LazyIndex.scala:63) at
kafka.log.LazyIndex.get(LazyIndex.scala:60) at
kafka.log.LogSegment.timeIndex(LogSegment.scala:66) at
kafka.log.LogSegment.maxTimestampAndOffsetSoFar(LogSegment.scala:107) at
kafka.log.LogSegment.maxTimestampSoFar(LogSegment.scala:113) at
kafka.log.LogSegment.largestTimestamp(LogSegment.scala:640) at
kafka.log.LogCleanerManager$.$anonfun$cleanableOffsets$4(LogCleanerManager.scala:617)
at
kafka.log.LogCleanerManager$.$anonfun$cleanableOffsets$4$adapted(LogCleanerManager.scala:616)
at kafka.log.LogCleanerManager$$$Lambda$2215/0x000000080104d040.apply(Unknown
Source) at scala.collection.Iterator.find(Iterator.scala:993) at
scala.collection.Iterator.find$(Iterator.scala:990) at
scala.collection.AbstractIterator.find(Iterator.scala:1429) at
scala.collection.IterableLike.find(IterableLike.scala:81) at
scala.collection.IterableLike.find$(IterableLike.scala:80) at
scala.collection.AbstractIterable.find(Iterable.scala:56) at
kafka.log.LogCleanerManager$.cleanableOffsets(LogCleanerManager.scala:616) at
kafka.log.LogCleanerManager.$anonfun$grabFilthiestCompactedLog$4(LogCleanerManager.scala:186)
at kafka.log.LogCleanerManager$$Lambda$2212/0x000000080104f040.apply(Unknown
Source) at
scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:238) at
scala.collection.TraversableLike$$Lambda$891/0x0000000800a89840.apply(Unknown
Source) at scala.collection.immutable.List.foreach(List.scala:392)
{code}
{code:java}
"kafka-log-cleaner-thread-0" java.lang.Thread.State: RUNNABLE at
java.base@11.0.13/sun.nio.ch.FileDispatcherImpl.pread0(Native Method) at
java.base@11.0.13/sun.nio.ch.FileDispatcherImpl.pread(FileDispatcherImpl.java:54)
at java.base@11.0.13/sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:274)
at java.base@11.0.13/sun.nio.ch.IOUtil.read(IOUtil.java:245) at
java.base@11.0.13/sun.nio.ch.FileChannelImpl.readInternal(FileChannelImpl.java:811)
at java.base@11.0.13/sun.nio.ch.FileChannelImpl.read(FileChannelImpl.java:796)
at org.apache.kafka.common.utils.Utils.readFully(Utils.java:1114) at
org.apache.kafka.common.utils.Utils.readFullyOrFail(Utils.java:1087) at
org.apache.kafka.common.record.FileLogInputStream.nextBatch(FileLogInputStream.java:69)
at
org.apache.kafka.common.record.FileLogInputStream.nextBatch(FileLogInputStream.java:42)
at
org.apache.kafka.common.record.RecordBatchIterator.makeNext(RecordBatchIterator.java:35)
at
org.apache.kafka.common.record.RecordBatchIterator.makeNext(RecordBatchIterator.java:24)
at
org.apache.kafka.common.utils.AbstractIterator.maybeComputeNext(AbstractIterator.java:79)
at
org.apache.kafka.common.utils.AbstractIterator.hasNext(AbstractIterator.java:45)
at kafka.log.LogSegment.loadFirstBatchTimestamp(LogSegment.scala:516) at
kafka.log.LogSegment.getFirstBatchTimestamp(LogSegment.scala:543) at
kafka.log.Log.$anonfun$getFirstBatchTimestampForSegments$1(Log.scala:2201) at
kafka.log.Log.$anonfun$getFirstBatchTimestampForSegments$1$adapted(Log.scala:2200)
at kafka.log.Log$$Lambda$2223/0x000000080105a040.apply(Unknown Source) at
scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:238) at
scala.collection.TraversableLike$$Lambda$891/0x0000000800a89840.apply(Unknown
Source) at scala.collection.Iterator.foreach(Iterator.scala:941) at
scala.collection.Iterator.foreach$(Iterator.scala:941) at
scala.collection.AbstractIterator.foreach(Iterator.scala:1429) at
scala.collection.IterableLike.foreach(IterableLike.scala:74) at
scala.collection.IterableLike.foreach$(IterableLike.scala:73) at
scala.collection.AbstractIterable.foreach(Iterable.scala:56) at
scala.collection.TraversableLike.map(TraversableLike.scala:238) at
scala.collection.TraversableLike.map$(TraversableLike.scala:231) at
scala.collection.AbstractTraversable.map(Traversable.scala:108) at
kafka.log.Log.getFirstBatchTimestampForSegments(Log.scala:2199) at
kafka.log.LogCleanerManager$.maxCompactionDelay(LogCleanerManager.scala:552) at
kafka.log.LogCleanerManager.$anonfun$grabFilthiestCompactedLog$4(LogCleanerManager.scala:190)
{code}
Upon inspection of the grabFilthiestCompactedLog method, it seems the most time
consuming part can be performed without holding the lock.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
