[jira] [Updated] (KAFKA-4358) Following a hung broker, newly elected leader is unnecessarily slow assuming leadership because of ReplicaFetcherThread

["Nelson Elhage (JIRA)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/KAFKA-4358?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Nelson Elhage updated KAFKA-4358:
---------------------------------
    Status: Open  (was: Patch Available)

> Following a hung broker, newly elected leader is unnecessarily slow assuming leadership because of ReplicaFetcherThread
> -----------------------------------------------------------------------------------------------------------------------
>
>                 Key: KAFKA-4358
>                 URL: https://issues.apache.org/jira/browse/KAFKA-4358
>             Project: Kafka
>          Issue Type: Bug
>          Components: replication
>    Affects Versions: 0.10.0.1
>            Reporter: Nelson Elhage
>            Priority: Minor
>
> When a broker handles a `LeaderAndIsr` request, the replica manager blocks waiting for idle replication fetcher threads to die before responding to the message and being able to service new produce requests.
> If requests to a broker start blackholing (e.g. due to network failure, or due to the broker hanging), shutting down the `ReplicaFetcherThread` can take a long time (around 30s in my testing), blocking recovery of any partitions previously lead by that broker.
> This is a very similar issue to KAFKA-612.
> Instructions to reproduce/demonstrate:
> Stand up three brokers and create a replicated topic:
> {code}
> bin/zookeeper-server-start.sh config/zookeeper.properties &
> bin/kafka-server-start.sh config/server1.properties &
> bin/kafka-server-start.sh config/server2.properties &
> bin/kafka-server-start.sh config/server3.properties &
> bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic replicated.topic
> {code}
> Identify the leader, and (for simplicity in interpreting the event) make sure it's not the same as the cluster controller:
> {code}
> bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic replicated.topic
> {code}
> Start a stream of produce events (with a shortened timeout so we get faster visibility into when the cluster recovers):
> {code}
> echo request.timeout.ms=1000 >> config/producer.properties
> bin/kafka-verifiable-producer.sh --throughput 2 --topic replicated.topic --broker-list localhost:9092 --producer.config $(pwd)/config/producer.properties
> {code}
> Now SIGSTOP the leader (3, in my example):
> {code}
> kill -STOP $(pgrep -f server3.properties)
> {code}
> The producer will log errors for about 30 seconds, and then recover. However, if we read logs, we'll see (excerpting key log lines from `state-change.log`):
> {code}
> [2016-10-28 20:36:03,128] TRACE Controller 2 epoch 8 sending become-leader LeaderAndIsr request (Leader:2,ISR:2,1,LeaderEpoch:22,ControllerEpoch:8) to broker 2 for partition [replicated.topic,0] (state.change.logger)
> [2016-10-28 20:36:03,131] TRACE Broker 2 handling LeaderAndIsr request correlationId 2 from controller 2 epoch 8 starting the become-leader transition for partition [replicated.topic,0] (state.change.logger)
> [2016-10-28 20:48:17,741] TRACE Controller 1 epoch 11 changed partition [replicated.topic,0] from OnlinePartition to OnlinePartition with leader 3 (state.change.logger)
> [2016-10-28 20:48:33,012] TRACE Controller 1 epoch 11 changed partition [replicated.topic,0] state from OnlinePartition to OfflinePartition (state.change.logger)
> [2016-10-28 20:48:33,012] TRACE Controller 1 epoch 11 started leader election for partition [replicated.topic,0] (state.change.logger)
> [2016-10-28 20:48:33,016] TRACE Controller 1 epoch 11 elected leader 2 for Offline partition [replicated.topic,0] (state.change.logger)
> [2016-10-28 20:48:33,017] TRACE Controller 1 epoch 11 changed partition [replicated.topic,0] from OfflinePartition to OnlinePartition with leader 2 (state.change.logger)
> [2016-10-28 20:48:33,017] TRACE Controller 1 epoch 11 sending become-leader LeaderAndIsr request (Leader:2,ISR:1,2,LeaderEpoch:30,ControllerEpoch:11) to broker 2 for partition [replicated.topic,0] (state.change.logger)
> [2016-10-28 20:48:33,023] TRACE Broker 2 received LeaderAndIsr request PartitionState(controllerEpoch=11, leader=2, leaderEpoch=30, isr=[1, 2], zkVersion=46, replicas=[1, 2, 3]) correlation id 18 from controller 1 epoch 11 for partition [replicated.topic,0] (state.change.logger)
> [2016-10-28 20:48:33,024] TRACE Broker 2 handling LeaderAndIsr request correlationId 18 from controller 1 epoch 11 starting the become-leader transition for partition [replicated.topic,0] (state.change.logger)
> [2016-10-28 20:48:33,026] TRACE Broker 2 stopped fetchers as part of become-leader request from controller 1 epoch 11 with correlation id 18 for partition [replicated.topic,0] (state.change.logger)
> [2016-10-28 20:48:33,026] TRACE Broker 2 completed LeaderAndIsr request correlationId 18 from controller 1 epoch 11 for the become-leader transition for partition [replicated.topic,0] (state.change.logger)
> [2016-10-28 20:48:56,058] TRACE Controller 1 epoch 11 received response {error_code=0,partitions=[{topic=replicated.topic,partition=0,error_code=0}]} for a request sent to broker qa-dev1.northwest.stripe.io:9093 (id: 2 rack: null) (state.change.logger)
> {code}
> Note the ~23s pause between broker 2 logging completion of the LeaderAndIsr request, and the controller logging receipt. If we look at broker 2 specifically now, we find
> {code}
> [2016-10-28 20:48:33,025] INFO [ReplicaFetcherManager on broker 2] Removed fetcher for partitions replicated.topic-0 (kafka.server.ReplicaFetcherManager)
> [2016-10-28 20:48:33,026] INFO [ReplicaFetcherThread-0-3], Shutting down (kafka.server.ReplicaFetcherThread)
> [2016-10-28 20:48:56,055] INFO [ReplicaFetcherThread-0-3], Stopped  (kafka.server.ReplicaFetcherThread)
> [2016-10-28 20:48:56,055] INFO [ReplicaFetcherThread-0-3], Shutdown completed (kafka.server.ReplicaFetcherThread)
> {code}
> Which timestamps exactly match the "missing" seconds. It also aligns with the first successful publish from my publish job, which happened at `2016-10-28 20:48:57.555`
> Aggressively dropping `request.timeout.ms` speeds up this failover process, but it seems that we should be able to recover those ~25s of unavailability without having to drop any additional timeouts, by either interrupting that thread somehow or not having to block on its shutdown.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)