kafkaクラスタController選挙と責任設計構想アーキテクチャ詳細-kafkaビジネス環境実戦

  class KafkaController(val config : KafkaConfig, zkClient: ZkClient, 
  val brokerState: BrokerState) extends Logging with KafkaMetricsGroup {
      this.logIdent = "[Controller " + config.brokerId + "]: "
      private var isRunning = true
      private val stateChangeLogger = KafkaController.stateChangeLogger
      val controllerContext = new ControllerContext(zkClient, config.zkSessionTimeoutMs)

      val partitionStateMachine = new PartitionStateMachine(this)
      val replicaStateMachine = new ReplicaStateMachine(this)
      
      private val controllerElector = new ZookeeperLeaderElector(controllerContext, ZkUtils.ControllerPath, onControllerFailover,
      onControllerResignation, config.brokerId)
      // have a separate scheduler for the controller to be able to start and stop independently of the
      // kafka server
      private val autoRebalanceScheduler = new KafkaScheduler(1)
      var deleteTopicManager: TopicDeletionManager = null
      val offlinePartitionSelector = new OfflinePartitionLeaderSelector(controllerContext, config)
      private val reassignedPartitionLeaderSelector = new ReassignedPartitionLeaderSelector(controllerContext)
      private val preferredReplicaPartitionLeaderSelector = new PreferredReplicaPartitionLeaderSelector(controllerContext)
      private val controlledShutdownPartitionLeaderSelector = new ControlledShutdownLeaderSelector(controllerContext)
      private val brokerRequestBatch = new ControllerBrokerRequestBatch(this)
      
      private val partitionReassignedListener = new PartitionsReassignedListener(this)
      private val preferredReplicaElectionListener = new PreferredReplicaElectionListener(this)

  1、ReassignedPartitionLeaderSelector
      ISR        leader，    ISR    ISR，             LeaderAndIsr       。
  2、PreferredReplicaPartitionLeaderSelector
     assignedReplicas            leader  ，     ，             leader。
  3、ControlledShutdownLeaderSelector
   ISR                 ，       ISR  ，           leader，     AR    LeaderAndIsr     。
  4、NoOpLeaderSelector
           ，     leader isr。
  5、OfflinePartitionLeaderSelector
      ISR     broker  leader，  ISR        ，  assignedReplicas         leader，leader        Zookeeper ，        。

  ../bin/kafka-topics.sh --zookeeper 127.0.0.1:2181 --describe  --topic test1
  
  Topic:test1       PartitionCount:3        ReplicationFactor:2     Configs:
  Topic: test1      Partition: 0    Leader: 2       Replicas: 2,4   Isr: 2,4
  Topic: test1      Partition: 1    Leader: 3       Replicas: 3,5   Isr: 3,5
  Topic: test1      Partition: 2    Leader: 4       Replicas: 4,1   Isr: 4,1

  ./kafka-topics.sh --zookeeper 127.0.0.1:2181 -alter --partitions 4 --topic test1

  1 NewReplica:  partition reassignment  KafkaController  New replica
  
  2 OnlineReplica:    replica    parition assingned replicas 
       OnlineReplica,       OnlineReplica
  
  3 Online   parition     leader isr    
  
  4 OfflineReplica:    broker down ,    replica   die,       Onffline;
  ReplicaDeletionStarted:    replica        ,      ReplicaDeletionStarted
  
  5 ReplicaDeletionSuccessful: Replica     ,      ReplicaDeletionSuccessful
  
  6 ReplicaDeletionIneligible: Replica     ,      ReplicaDeletionIneligible
  
  7 NonExistentReplica:  Replica     ,  ReplicaDeletionSuccessful     NonExistentReplica  

  NonExistentPartition:   partition                 ;
  NewPartition:   parition   , replicas     , leader/isr    ;
  OnlinePartition:   partition leader  ;
  OfflinePartition:   partition leader  ,  parition   OfflinePartition;