Skip to content

KAFKA-3076: BrokerChangeListener should log the brokers in order #749

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
konradkalita wants to merge 7 commits into from
Closed

KAFKA-3076: BrokerChangeListener should log the brokers in order #749

Changes from 5 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
acbc58b
BrokerChangeListener logging the brokers in order
konradkalita Jan 9, 2016
f220ab0
BrokerChangeListener logging refactor
konradkalita Jan 11, 2016
0e73989
BrokerChangeListener now passes sorted seq to handlers
konradkalita Jan 12, 2016
5005acb
Merge branch 'trunk' into kafka-3076
konradkalita Jan 12, 2016
2d81ab6
Merge conflicts resolved
konradkalita Jan 12, 2016
57a264b
Merge branch 'trunk' into kafka-3076
konradkalita Jan 25, 2016
4ccf612
liveBrokerIds inconsistency fix
konradkalita Jan 25, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 7 additions & 4 deletions core/src/main/scala/kafka/controller/ReplicaStateMachine.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -352,7 +352,7 @@ class ReplicaStateMachine(controller: KafkaController) extends Logging {
class BrokerChangeListener() extends IZkChildListener with Logging {
this.logIdent = "[BrokerChangeListener on Controller " + controller.config.brokerId + "]: "
def handleChildChange(parentPath : String, currentBrokerList : java.util.List[String]) {
info("Broker change listener fired for path %s with children %s".format(parentPath, currentBrokerList.mkString(",")))
info("Broker change listener fired for path %s with children %s".format(parentPath, currentBrokerList.sorted.mkString(",")))
inLock(controllerContext.controllerLock) {
if (hasStarted.get) {
ControllerStats.leaderElectionTimer.time {
Expand All @@ -365,14 +365,17 @@ class ReplicaStateMachine(controller: KafkaController) extends Logging {
val brokerById = curBrokers.map(broker => broker.id -> broker).toMap
val newBrokers = newBrokerIds.flatMap(brokerById.get)
controllerContext.liveBrokers = curBrokers
val newBrokerIdsSorted = newBrokerIds.toSeq.sorted
val deadBrokerIdsSorted = deadBrokerIds.toSeq.sorted
val liveBrokerIdsSorted = controllerContext.liveBrokerIds.toSeq.sorted

@ijuma ijuma Jan 25, 2016

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should use curBrokers instead of controllerContext.liveBrokerIds here.

info("Newly added brokers: %s, deleted brokers: %s, all live brokers: %s"
.format(newBrokerIds.mkString(","), deadBrokerIds.mkString(","), controllerContext.liveBrokerIds.mkString(",")))
.format(newBrokerIdsSorted.mkString(","), deadBrokerIdsSorted.mkString(","), liveBrokerIdsSorted.mkString(",")))
newBrokers.foreach(controllerContext.controllerChannelManager.addBroker)
deadBrokerIds.foreach(controllerContext.controllerChannelManager.removeBroker)
if(newBrokerIds.size > 0)
controller.onBrokerStartup(newBrokerIds.toSeq)
controller.onBrokerStartup(newBrokerIdsSorted)
if(deadBrokerIds.size > 0)
controller.onBrokerFailure(deadBrokerIds.toSeq)
controller.onBrokerFailure(deadBrokerIdsSorted)
} catch {
case e: Throwable => error("Error while handling broker changes", e)
}
Expand Down