MINOR: Log controller id/epoch when LeaderAndIsr, StopReplica and UpdateMetadata requests are fenced
Reviewers: Luke Chen
@@ -265,8 +265,9 @@ class KafkaApis(val requestChannel: RequestChannel,
if (isBrokerEpochStale(zkSupport, leaderAndIsrRequest.brokerEpoch)) {
// When the broker restarts very quickly, it is possible for this broker to receive request intended
// for its previous generation so the broker should skip the stale request.
- info("Received LeaderAndIsr request with broker epoch " +
- s"${leaderAndIsrRequest.brokerEpoch} smaller than the current broker epoch ${zkSupport.controller.brokerEpoch}")
+ info(s"Received LeaderAndIsr request with broker epoch ${leaderAndIsrRequest.brokerEpoch} " +
+ s"smaller than the current broker epoch ${zkSupport.controller.brokerEpoch} from " +
+ s"controller ${leaderAndIsrRequest.controllerId} with epoch ${leaderAndIsrRequest.controllerEpoch}.")
requestHelper.sendResponseExemptThrottle(request, leaderAndIsrRequest.getErrorResponse(0, Errors.STALE_BROKER_EPOCH.exception))
} else {
val response = replicaManager.becomeLeaderOrFollower(correlationId, leaderAndIsrRequest,
@@ -285,8 +286,9 @@ class KafkaApis(val requestChannel: RequestChannel,
if (isBrokerEpochStale(zkSupport, stopReplicaRequest.brokerEpoch)) {
// When the broker restarts very quickly, it is possible for this broker to receive request intended
// for its previous generation so the broker should skip the stale request.
- info("Received StopReplica request with broker epoch " +
- s"${stopReplicaRequest.brokerEpoch} smaller than the current broker epoch ${zkSupport.controller.brokerEpoch}")
+ info(s"Received StopReplica request with broker epoch ${stopReplicaRequest.brokerEpoch} " +
+ s"smaller than the current broker epoch ${zkSupport.controller.brokerEpoch} from " +
+ s"controller ${stopReplicaRequest.controllerId} with epoch ${stopReplicaRequest.controllerEpoch}.")
requestHelper.sendResponseExemptThrottle(request, new StopReplicaResponse(
new StopReplicaResponseData().setErrorCode(Errors.STALE_BROKER_EPOCH.code)))
} else {
@@ -344,8 +346,9 @@ class KafkaApis(val requestChannel: RequestChannel,
if (isBrokerEpochStale(zkSupport, updateMetadataRequest.brokerEpoch)) {
// When the broker restarts very quickly, it is possible for this broker to receive request intended
// for its previous generation so the broker should skip the stale request.
- info("Received update metadata request with broker epoch " +
- s"${updateMetadataRequest.brokerEpoch} smaller than the current broker epoch ${zkSupport.controller.brokerEpoch}")
+ info(s"Received UpdateMetadata request with broker epoch ${updateMetadataRequest.brokerEpoch} " +
+ s"smaller than the current broker epoch ${zkSupport.controller.brokerEpoch} from " +
+ s"controller ${updateMetadataRequest.controllerId} with epoch ${updateMetadataRequest.controllerEpoch}.")
requestHelper.sendResponseExemptThrottle(request,
new UpdateMetadataResponse(new UpdateMetadataResponseData().setErrorCode(Errors.STALE_BROKER_EPOCH.code)))
} else {
