Steve Niemitz created KAFKA-4523:
------------------------------------
Summary: Controlled shutdown fails if consumer group restabilizes
during shutdown
Key: KAFKA-4523
URL: https://issues.apache.org/jira/browse/KAFKA-4523
Project: Kafka
Issue Type: Bug
Affects Versions: 0.10.1.0
Reporter: Steve Niemitz
If I begin a controlled shutdown of a broker that is a coordinator for a
consumer group, often the shutdown will fail with the following error:
{code}
[2016-12-12 16:24:15,424] INFO [Replica Manager on Broker 10]: Shut down
completely (kafka.server.ReplicaManager)
[2016-12-12 16:24:15,424] INFO [ExpirationReaper-10], Shutting down
(kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2016-12-12 16:24:15,450] INFO [ExpirationReaper-10], Stopped
(kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2016-12-12 16:24:15,450] INFO [ExpirationReaper-10], Shutdown completed
(kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2016-12-12 16:24:15,451] INFO Shutting down. (kafka.log.LogManager)
[2016-12-12 16:24:31,241] INFO [GroupCoordinator 10]: Preparing to restabilize
group my-consumer-group with old generation 2673
(kafka.coordinator.GroupCoordinator)
[2016-12-12 16:24:32,499] INFO [GroupCoordinator 10]: Group my-consumer-group
with generation 2674 is now empty (kafka.coordinator.GroupCoordinator)
[2016-12-12 16:24:32,515] FATAL [Replica Manager on Broker 10]: Halting due to
unrecoverable I/O error while handling produce request:
(kafka.server.ReplicaManager)
kafka.common.KafkaStorageException: I/O exception in append to log
'__consumer_offsets-33'
at kafka.log.Log.append(Log.scala:349)
at kafka.cluster.Partition$$anonfun$10.apply(Partition.scala:443)
at kafka.cluster.Partition$$anonfun$10.apply(Partition.scala:429)
at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:234)
at kafka.utils.CoreUtils$.inReadLock(CoreUtils.scala:240)
at kafka.cluster.Partition.appendMessagesToLeader(Partition.scala:429)
at
kafka.server.ReplicaManager$$anonfun$appendToLocalLog$2.apply(ReplicaManager.scala:407)
at
kafka.server.ReplicaManager$$anonfun$appendToLocalLog$2.apply(ReplicaManager.scala:393)
at
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
at
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
at scala.collection.immutable.Map$Map1.foreach(Map.scala:109)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
at scala.collection.AbstractTraversable.map(Traversable.scala:105)
at
kafka.server.ReplicaManager.appendToLocalLog(ReplicaManager.scala:393)
at kafka.server.ReplicaManager.appendMessages(ReplicaManager.scala:330)
at
kafka.coordinator.GroupMetadataManager.store(GroupMetadataManager.scala:251)
at
kafka.coordinator.GroupCoordinator$$anonfun$onCompleteJoin$6.apply(GroupCoordinator.scala:726)
at
kafka.coordinator.GroupCoordinator$$anonfun$onCompleteJoin$6.apply(GroupCoordinator.scala:726)
at scala.Option.foreach(Option.scala:236)
at
kafka.coordinator.GroupCoordinator.onCompleteJoin(GroupCoordinator.scala:726)
at kafka.coordinator.DelayedJoin.onComplete(DelayedJoin.scala:39)
at
kafka.server.DelayedOperation.forceComplete(DelayedOperation.scala:70)
at
kafka.coordinator.DelayedJoin$$anonfun$tryComplete$1.apply$mcZ$sp(DelayedJoin.scala:37)
at
kafka.coordinator.GroupCoordinator.tryCompleteJoin(GroupCoordinator.scala:672)
at kafka.coordinator.DelayedJoin.tryComplete(DelayedJoin.scala:37)
at
kafka.server.DelayedOperationPurgatory$Watchers.tryCompleteWatched(DelayedOperation.scala:315)
at
kafka.server.DelayedOperationPurgatory.checkAndComplete(DelayedOperation.scala:234)
at
kafka.coordinator.GroupCoordinator.onMemberFailure(GroupCoordinator.scala:665)
at
kafka.coordinator.GroupCoordinator.onExpireHeartbeat(GroupCoordinator.scala:740)
at
kafka.coordinator.DelayedHeartbeat.onExpiration(DelayedHeartbeat.scala:33)
at kafka.server.DelayedOperation.run(DelayedOperation.scala:107)
at
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.nio.channels.ClosedChannelException
at sun.nio.ch.FileChannelImpl.ensureOpen(FileChannelImpl.java:110)
at sun.nio.ch.FileChannelImpl.size(FileChannelImpl.java:300)
at kafka.log.FileMessageSet.truncateTo(FileMessageSet.scala:405)
at kafka.log.FileMessageSet.trim(FileMessageSet.scala:378)
at kafka.log.Log.roll(Log.scala:773)
at kafka.log.Log.maybeRoll(Log.scala:742)
at kafka.log.Log.append(Log.scala:405)
... 35 more
{code}
This then causes the broker to attempt to run recovery on all log segments on
the next startup, which obviously is not ideal.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
