Replica {

Replica {                    // a replica of a partition
  broker_id   : int
  partition   : Partition
  isLocal     : Boolean      // is this replica local to this broker
  log         : Log          // local log associated with this replica
  hw  // a replica of a partition
  broker_id : long : int
  partition     // offset of the last committed message: Partition
  leo isLocal        : longBoolean         // logis end offset
}

Partition {     this replica local to this broker
  log         : Log          //a partitionlocal inlog aassociated topic
with this topicreplica
  hw          : string
long  partition_id     : int
  leader    // offset of the last committed message
  leo         : Replicalong         // thelog leader replica of this partition
  commitQend offset
}

Partition {              : Queue          //a produce requests pending commit at the leader
  ARpartition in a topic
  topic            : string
  partition_id     : Set[Replica]int
  leader // replicas assigned to this partition
  ISR   : Replica        // the : Set[Replica]leader replica of this partition
  commitQ          : Queue          // In-syncproduce replicarequests set,pending maintainedcommit at the leader
  AR CUR              : Set[Replica]   // Catch-upreplicas replicaassigned set,to maintained at the leaderthis partition
  RARISR              : Set[Replica]   // ReassignedIn-sync replica set, maintained at the leader
}

brokerStartup()
{
  create /brokers/state/[broker_id] path if it doesn’t already exist
  register the state change handler to listen on child change ZK notifications on /brokers/state/[broker_id]
  register session expiration listener
  drain the state change queue 
  get replica info from ZK and compute AR, a list of replicas assigned to this broker
  for each r in AR
  {
      subscribe to leader changes for the r’s partition
      startReplica(r)
  }
  // broker startup procedure is complete. Register is broker id in ZK to announce the availability of this broker
  register its broker_id in /brokers/ids/[broker_id] in ZK
}

leaderElection(r: Replica)
  read the current ISR and AR for r.partition.partition_id from ZK
  if( (r in AR) && (ISR is empty || r in ISR) ) 
  {
     wait for PreferredReplicaTime if r is not the preferred replica
     if(successfully write r as the current leader of r.partition in ZK)
       becomeLeader(r, ISR, CUR)
     else
       becomeFollower(r)
  }
} 

onLeaderChange() 
{
   if(broker_id is registered under /brokers/topics/[topic]/[partition_id]/replicas)
     leaderElection(r)
}

stateChangeListener() {  
// listens to state change requests issued by the leader and acts on those

    drain the state change queue
    read next state change request
    Let r be the replica that the state change request is sent for.
    // this should not happen
    Throw an error if r is not hosted on this broker                
    Let requestEpoch be the epoch of the state change request
    if(closeReplicaRequest) 
    { 
       // we don’t need to check epoch here to be able to handle delete topic/delete partition for dead brokers.
       closeReplica(r)
    }
    if(startReplicaRequest) 
    { 
       Let latestPartitionEpoch be the latest epoch for this partition, got by reading /brokers/topics/[topic]/[partition]/ISR
       if(leader for r.partition doesn’t exist) {
	   // this can only happen for new topics or new partitions for existing topics
         startReplica(r)         
       }else if(requestEpoch == latestPartitionEpoch) {
         // this is to ensure that if a follower is slow, and reads a state change request queued up by a previous leader, it ignores the request
	    startReplica(r

leaderElection(r: Replica)
  read the current ISR and AR for r.partition.partition_id from ZK
  if( (r in AR) && (ISR is empty || r in ISR) ) 
  {
     wait for PreferredReplicaTime if r is not the preferred replica
     if(successfully write r as the current leader of r.partition in ZK)
       becomeLeader(r, ISR, CUR)
}
    }
} 

sendStateChange(stateChange, followerBrokerId, leaderEpoch)
{
   stateChangeQ = new StateChangeQueue(“/brokers/state/followerBrokerId”)
   stateChangeRequest = new StateChangeRequest(stateChange, leaderEpoch)
   // check if the state change Q is full. This can happen if a broker is offline for a long time
   if(stateChangeQ.isFull) {
      // wethis don’toperation needretains toonly checkone epochclose-replica hererequest tofor bea ablepartition, tothe handleone deletewith topic/deletethe partition for dead brokers.
    }
    if(startReplicaRequest) 
    { 
       Let latestPartitionEpoch be the latest epoch for this partition, got by reading /brokers/topics/[topic]/[partition]/leader=[broker_id],[epoch] latest epoch. This is to ensure that an offline broker, on startup, will delete old topics and partitions, which it hosted before going offline. You don’t have to retain any start-replica requests for a partition
       if(leader for r.partition doesn’t exist) {
	stateChangeQ.shrink
      // this can only happen for new topics or new partitions for existing topics
         startReplica(r)         
       }else if(requestEpoch == latestPartitionEpochif the queue is still full, log an error
      throw new FollowerStateChangeQueueFull
   }
   stateChangeQ.put(stateChangeRequest)
} 

startReplica(r: Replica) {
  if(broker_id not in /brokers/topics/[r.topic]/[r.partition]/replicas)
    throw NotReplicaForPartitionException()
  if( r's log is not already started) {
     do local recovery of // this is to ensure that if a follower is slow, and reads a state change request queued up by a previous leader, it ignores the request
	    startReplica(r)
       }
    }
}

becomeFollower(r: Replica)
{
  // this is required if this replica was the last leader
  stop the commit thread, if any                  
  stop the current ReplicaFetcherThread, if any
  truncate the log to r.hw
  start a new ReplicaFetcherThread to the current leader of r, from offset r.leo
  start HW checkpoint thread for r
}

becomeLeader(r: Replica, ISR: Set[Replica], AR: Set[Replica])
{
   // get a new epoch value and write it to the leader path
   epoch = getNewEpoch()
   /brokers/topics/[r.partition.topic]/[r.partition.pid]/leader=broker_id, epoch
  

startReplica(r: Replica) {
  if(broker_id not in /brokers/topics/[r.partition.topic]/[r.partition.pid]/replicas)ISR=ISR;epoch
   stop throw NotReplicaForPartitionException()
  if( r's log is not already started) {HW checkpoint thread for r
   r.hw = do local recovery of r's log
     r.leo // TODO: check if this should actually be r.hw = min(last checkpointed HW for r, r.leo)
     register a leader-change listener on partition r.partition.partition_id
  }
   if( a leader does not exist for partition r.partition.partition_id in ZK)
      leaderElection(r)
   else {
     if(this broker is not the leader, then it is a follower since it is in the AR list for this partition)
      becomeFollower(r)
   }
}

createTopic(topic, numPartitions, replicationFactor, replicaAssignmentStr) 
{
   if(!cleanFailedTopicCreationAttempt(topic))
   {
      error(“Topic topic exists with live partitions”)
      exit
   }  
   if(replicaAssignmentStr == “”) {
     // assignReplicas will always assign partitions only to online brokers
     replicaAssignment = assignReplicas(topic, numPartitions, replicationFactor)
   }

   // create topic path in ZK
   create /brokers/topics/topic
   for(partition <- replicaAssignment) {
     addPartition(topic, partition.id, partition.replicas) 
   }
   // report successfully started partitions for this topic
}
waitTillStateChangeRequestConsumed(partition.replicas, timeout)
{
   register watch on state change path for each replica
   In the listener, use a condition variable to await(timeout). If it doesn’t fire return false, else return true
}

cleanFailedTopicCreationAttempts(topic) 
{
   topicsForPartitionsReassignment = ls /brokers/partitions_reassigned
   for(topic <- topicsForPartitionsReassignment)
   {
      partitionsCreated = ls /brokers/partitions_reassigned/topic
      cleanupFailed = false
      for(partition <- partitionsCreated) 
      {
          if(/brokers/topics/topic/partition/replicas path exists)
          {

becomeLeader(r: Replica, ISR: Set[Replica], AR: Set[Replica])
{
   // get a new epoch value and write it to the leader path
   epoch = getNewEpoch()
   /brokers/topics/[r.partition.topic]/[r.partition.pid]/leader=broker_id, epoch
   stop HW checkpoint thread for r
   r.hw = r.leo // TODO: check if this should actually be r.hw = last checkpointed HW for r
   wait until every live replica in AR catches up (i.e. its leo == r.hw) or a KeepInSyncTime has passed
   r.partition.ISR = the current set of replicas in sync with r
   r.partition.CUR = AR - ISR
   write r.partition.ISR & r.partition.CUR in ZK
   r.partition.RAR = replicas in /brokers/partitions_reassigned/[topic]/[partition_id] in ZK
   r.partition.leader = r                 delete /brokers/ this enables reads/writes to this partition on this broker
   start a commit thread on r.partition
   start HW checkpoint thread for r
    get replica info from ZK and compute AR, a list of replicas assigned to this broker
    for each r in ARpartitions_reassigned/topic/partition
             error(“Cannot cleanup. Topic exists with live partition”) 
             cleanupFailed = true
          }
      }
      if(cleanupFailed) {
        if(r.broker_id is registered under /brokers/ids/brokers/partitions_reassigned/topic has no children)
          sendStateChange(“start-replica”, r.broker_id, epoch) delete /brokers/partitions_reassigned/topic
        return false
    }
}

deleteTopic(topic) 
{
   partitionsForTopic = ls /brokers/topics/topic
   for(partition <- replicaAssignmentpartitionsForTopic) {
     addPartition(topic, partition.id if(!deletePartition(topic, partition.replicas))
 
     }{
     // report successfully started partitionserror(“Failed forto this topic
}
waitTillStateChangeRequestConsumed(partition.replicas, timeout)
{
delete partition for topic”)
   register watch on state change path forexit each
 replica
   In the listener,}
 use a condition}
 variable  // delete topic path in ZK
   delete /brokers/topics/topic
}

addPartition(topic, partition, replicasto await(timeout). If it doesn’t fire return false, else return true
}

cleanFailedTopicCreationAttempts(topic) 
{
   topicsForPartitionsReassignment = ls /brokers/ write the partitions_ reassigned
 path  for(topic <- topicsForPartitionsReassignment)
   {
  this create topic command
    partitionsCreated = ls /brokers/topicspartitions_reassigned/topic/partition=replicas
	// start replicas    for(partition <-this partitionsCreated)new partition
      {
		if(/brokers/topics/topic/partition/replicas path existsfor(replica <- replicas)
        sendStateChange(“start-replica”,  {replica.brokerId, -1)
     // wait till state change request is consumed error(“Cannot cleanup. Topic exists with live partition”)
   by all replicas
     if(!waitTillStateChangeRequestConsumed(partition.replicas, timeout)) 
     {
     return false 
 error(“Failed to  create topic partition partitionId for timeout }ms”)
      }
	 // partitionexit
 paths can be safely deleted}
     // create for(partition <- partitionsCreated)
 paths in ZK
        delete /brokers/topics/topic/partitionpartitionId/replicas=replicas
      delete /brokers/topicspartitions_reassigned/topic
   } /partitionId
}

produceRequestHandler(pr: ProduceRequest)
{
   if( the request partition pr.partition doesn't have leader replica on this broker)
      throw  sendStateChange(“start-replica”, replica.brokerId, -1)
     // wait till state change request is consumed by all replicas
     if(!waitTillStateChangeRequestConsumed(partition.replicas, timeout)) 
    NotLeaderException
   log = r.partition.leader.log
   append pr.messages to log
   pr.offset = log.LEO
   add pr to pr.partition.commitQ
}

while(true) {
  pr = commitQ.dequeue
  canCommit = false
  while(!canCommit) {
    canCommit = true
  error(“Failed to createfor topiceach partitionr partitionId for timeout ms”)in ISR
        exit
  if(!offsetReached(r, pr.offset)) {
	canCommit = false
	break
   }
    }
 // create partition paths in ZK
     /brokers/topics/topic/partitionId/replicas=replicasif(!canCommit) {
      p.CUR.add(r)
     delete /brokers/partitions_reassigned/topic/partitionId
}

ReplicaFetchReqeust {
   topic: String
   partition_id: 
}

completePartitionsReassignment(RAR: Set[Replica], ISR: Set[Replica], AR: Set[Replica], stopCommitThread: Boolean)
Int
   replica_id: Int
   offset: Long
}

ReplicaFetchResponse {
   hw: Long			//newly the assignedoffset replicasof are in-sync, switch over tothe last message committed at the new replicasleader
   messages: MessageSet	// fetched messages
}

replicaFetch (f: ReplicaFetchRequest) {	// handler for ReplicaFetchRequest at leader
  leader = getLeaderReplica(f.topic, f.partition_id)

  if(leader == null) throw NotLeaderException
   response = new ReplicaFetcherResponse
   getReplica(f.topic, f.partition_id, f.replica_id).leo = f.offset
    response.messages = fetch messages starting from f.offset from leader.log
    response.hw = leader.hw
    send response back
}

while(true) {
   send ReplicaFetchRequest to leader and get response:ReplicaFetcherResponse back
   append response.messages to r's log
   r.hw = response.hw
   advance offset in ReplicaFetchRequestneed (RAR + ISR) in case we fail right after here

    write (RAR + ISR) as the new ISR in ZK     
    update /brokers/topics/[topic]/[partition_id]/replicas in ZK with the new replicas in RAR
//triggers leader election
    delete /brokers/partitions_reassigned/[topic]/[partition_id] in ZK
    if(stopCommitThread || (broker_id is not preferred replica))
    {
      delete /brokers/topics/[topic]/[partition_id]/leader in ZK 
      stop this commit thread
    }
}