handleStopReplica should not send a second response
`shutdownIdleFetcherThreads()` can throw InterruptedException
for example.
Author: Ismael Juma
Reviewers: Rajini Sivaram
Closes from ijuma/kafka-5289-stop-replica-should-not-send-two-responses
@@ -35,7 +35,7 @@ import kafka.coordinator.transaction.{InitProducerIdResult, TransactionCoordinat
import kafka.log.{Log, LogManager, TimestampOffset}
import kafka.network.{RequestChannel, RequestOrResponseSend}
import kafka.security.auth._
-import kafka.utils.{Exit, Logging, ZKGroupTopicDirs, ZkUtils}
+import kafka.utils.{CoreUtils, Exit, Logging, ZKGroupTopicDirs, ZkUtils}
import org.apache.kafka.common.errors._
import org.apache.kafka.common.internals.FatalExitError
import org.apache.kafka.common.internals.Topic.{GROUP_METADATA_TOPIC_NAME, TRANSACTION_STATE_TOPIC_NAME, isInternal}
@@ -195,7 +195,7 @@ class KafkaApis(val requestChannel: RequestChannel,
// request to become a follower due to which cache for groups that belong to an offsets topic partition for which broker 1 was the leader,
// is not cleared.
result.foreach { case (topicPartition, error) =>
- if (error == Errors.NONE && stopReplicaRequest.deletePartitions() && topicPartition.topic == GROUP_METADATA_TOPIC_NAME) {
+ if (error == Errors.NONE && stopReplicaRequest.deletePartitions && topicPartition.topic == GROUP_METADATA_TOPIC_NAME) {
groupCoordinator.handleGroupEmigration(topicPartition.partition)
}
}
@@ -207,7 +207,7 @@ class KafkaApis(val requestChannel: RequestChannel,
new StopReplicaResponse(Errors.CLUSTER_AUTHORIZATION_FAILED, result.asJava))
}
- replicaManager.replicaFetcherManager.shutdownIdleFetcherThreads()
+ CoreUtils.swallow(replicaManager.replicaFetcherManager.shutdownIdleFetcherThreads())
}
def handleUpdateMetadataRequest(request: RequestChannel.Request) {
