private RemotingCommand handleBrokerHeartbeat(ChannelHandlerContext ctx, RemotingCommand request) throws Exception {
    final BrokerHeartbeatRequestHeader requestHeader = (BrokerHeartbeatRequestHeader) request.decodeCommandCustomHeader(BrokerHeartbeatRequestHeader.class);
    if (requestHeader.getBrokerId() == null) {
        return RemotingCommand.createResponseCommand(ResponseCode.CONTROLLER_INVALID_REQUEST, "Heart beat with empty brokerId");
    }
    this.heartbeatManager.onBrokerHeartbeat(requestHeader.getClusterName(), requestHeader.getBrokerName(), requestHeader.getBrokerAddr(), requestHeader.getBrokerId(),
                                            requestHeader.getHeartbeatTimeoutMills(), ctx.channel(), requestHeader.getEpoch(), requestHeader.getMaxOffset(), requestHeader.getConfirmOffset(), requestHeader.getElectionPriority());
    return RemotingCommand.createResponseCommand(ResponseCode.SUCCESS, "Heart beat success");
}

public void onBrokerHeartbeat(String clusterName, String brokerName, String brokerAddr, Long brokerId,
                              Long timeoutMillis, Channel channel, Integer epoch, Long maxOffset, Long confirmOffset, Integer electionPriority) {
    BrokerIdentityInfo brokerIdentityInfo = new BrokerIdentityInfo(clusterName, brokerName, brokerId);
    BrokerLiveInfo prev = this.brokerLiveTable.get(brokerIdentityInfo);
    ......
    if (null == prev) {
        this.brokerLiveTable.put(...);
        log.info("new broker registered, {}, brokerId:{}", brokerIdentityInfo, realBrokerId);
    } else {
        prev.setXXX(......)
    }
}

public ControllerResult<ElectMasterResponseHeader> electMaster(final ElectMasterRequestHeader request,
                                                               final ElectPolicy electPolicy) {
    ...
    // 从request中取信息
    ...
    if (syncStateInfo.isFirstTimeForElect()) {
        // 从未注册，直接任命
        newMaster = brokerId;
    }

    // 按选举政策选主
    if (newMaster == null) {
        // we should assign this assignedBrokerId when the brokerAddress need to be elected by force
        Long assignedBrokerId = request.getDesignateElect() ? brokerId : null;
        newMaster = electPolicy.elect(brokerReplicaInfo.getClusterName(), brokerReplicaInfo.getBrokerName(), syncStateSet, allReplicaBrokers, oldMaster, assignedBrokerId);
    }

    if (newMaster != null && newMaster.equals(oldMaster)) {
        // 老主 == 新主
        // old master still valid, change nothing
        String err = String.format("The old master %s is still alive, not need to elect new master for broker %s", oldMaster, brokerReplicaInfo.getBrokerName());
        LOGGER.warn("{}", err);
        // the master still exist
        response.setXXX()

        result.setBody(new ElectMasterResponseBody(syncStateSet).encode());
        result.setCodeAndRemark(ResponseCode.CONTROLLER_MASTER_STILL_EXIST, err);
        return result;
    }

    // a new master is elected
    if (newMaster != null) {
        // 出现不一样的新主
        final int masterEpoch = syncStateInfo.getMasterEpoch();
        final int syncStateSetEpoch = syncStateInfo.getSyncStateSetEpoch();
        final HashSet<Long> newSyncStateSet = new HashSet<>();
        //设置新的syncStateSet
        newSyncStateSet.add(newMaster);
        response.setXXX()...
        ElectMasterResponseBody responseBody =
            new ElectMasterResponseBody(newSyncStateSet);
    }

        result.setBody(responseBody.encode());
        final ElectMasterEvent event = new ElectMasterEvent(brokerName, newMaster);
        result.addEvent(event);
        return result;
    }

    // 走到这里，说明没有主，选举失败
    // If elect failed and the electMaster is triggered by controller (we can figure it out by brokerAddress),
    // we still need to apply an ElectMasterEvent to tell the statemachine
    // that the master was shutdown and no new master was elected.
    if (request.getBrokerId() == null) {
        final ElectMasterEvent event = new ElectMasterEvent(false, brokerName);
        result.addEvent(event);
        result.setCodeAndRemark(ResponseCode.CONTROLLER_MASTER_NOT_AVAILABLE, "Old master has down and failed to elect a new broker master");
    } else {
        result.setCodeAndRemark(ResponseCode.CONTROLLER_ELECT_MASTER_FAILED, "Failed to elect a new master");
    }
        return result;
    }

private RemotingCommand handleAlterSyncStateSet(ChannelHandlerContext ctx,
                                                RemotingCommand request) throws Exception {
    final AlterSyncStateSetRequestHeader controllerRequest = (AlterSyncStateSetRequestHeader) request.decodeCommandCustomHeader(AlterSyncStateSetRequestHeader.class);
    final SyncStateSet syncStateSet = RemotingSerializable.decode(request.getBody(), SyncStateSet.class);
    final CompletableFuture<RemotingCommand> future = this.controllerManager.getController().alterSyncStateSet(controllerRequest, syncStateSet);
    if (future != null) {
        return future.get(WAIT_TIMEOUT_OUT, TimeUnit.SECONDS);
    }
    return RemotingCommand.createResponseCommand(null);
}

public ControllerResult<AlterSyncStateSetResponseHeader> alterSyncStateSet(
    final AlterSyncStateSetRequestHeader request, final SyncStateSet syncStateSet,
    final BrokerValidPredicate brokerAlivePredicate) {
    final String brokerName = request.getBrokerName();
  ...
    final Set<Long> newSyncStateSet = syncStateSet.getSyncStateSet();
    final SyncStateInfo syncStateInfo = this.syncStateSetInfoTable.get(brokerName);
    final BrokerReplicaInfo brokerReplicaInfo = this.replicaInfoTable.get(brokerName);

    // 检查syncStateSet是否有变化
    final Set<Long> oldSyncStateSet = syncStateInfo.getSyncStateSet();
    if (oldSyncStateSet.size() == newSyncStateSet.size() && oldSyncStateSet.containsAll(newSyncStateSet)) {
        String err = "The newSyncStateSet is equal with oldSyncStateSet, no needed to update syncStateSet";
        ...
    }

    // 检查是否是master发起的
    if (!syncStateInfo.getMasterBrokerId().equals(request.getMasterBrokerId())) {
        String err = String.format("Rejecting alter syncStateSet request because the current leader is:{%s}, not {%s}",
                                   syncStateInfo.getMasterBrokerId(), request.getMasterBrokerId());
        ...
    }

    // 检查master的任期epoch是否一致
    if (request.getMasterEpoch() != syncStateInfo.getMasterEpoch()) {
        String err = String.format("Rejecting alter syncStateSet request because the current master epoch is:{%d}, not {%d}",
                                   syncStateInfo.getMasterEpoch(), request.getMasterEpoch());
        ...
    }

    // 检查syncStateSet的epoch
    if (syncStateSet.getSyncStateSetEpoch() != syncStateInfo.getSyncStateSetEpoch()) {
        String err = String.format("Rejecting alter syncStateSet request because the current syncStateSet epoch is:{%d}, not {%d}",
                                   syncStateInfo.getSyncStateSetEpoch(), syncStateSet.getSyncStateSetEpoch());
        ...
    }

    // 检查新的syncStateSet的合理性
    for (Long replica : newSyncStateSet) {
        // 检查replica是否存在
        if (!brokerReplicaInfo.isBrokerExist(replica)) {
            String err = String.format("Rejecting alter syncStateSet request because the replicas {%s} don't exist", replica);
            ...
        }
        // 检查broker是否存活
        if (!brokerAlivePredicate.check(brokerReplicaInfo.getClusterName(), brokerReplicaInfo.getBrokerName(), replica)) {
            String err = String.format("Rejecting alter syncStateSet request because the replicas {%s} don't alive", replica);
            ...
        }
    }

    // 检查是否包含master
    if (!newSyncStateSet.contains(syncStateInfo.getMasterBrokerId())) {
        String err = String.format("Rejecting alter syncStateSet request because the newSyncStateSet don't contains origin leader {%s}", syncStateInfo.getMasterBrokerId());
        ...
    }

    // 更新epoch
    int epoch = syncStateInfo.getSyncStateSetEpoch() + 1;
    ...
    // 生成事件，替换syncStateSet
    final AlterSyncStateSetEvent event = new AlterSyncStateSetEvent(brokerName, newSyncStateSet);
    ...
}

protected abstract class AcceptSocketService extends ServiceThread {
    ...

    /**
     * Starts listening to slave connections.
     *
     * @throws Exception If fails.
     */
    public void beginAccept() throws Exception {
        ...
    }

    @Override
    public void shutdown(final boolean interrupt) {
        ...
    }

    @Override
    public void run() {
        log.info(this.getServiceName() + " service started");

        while (!this.isStopped()) {
            try {
                this.selector.select(1000);
                Set<SelectionKey> selected = this.selector.selectedKeys();

                if (selected != null) {
                    for (SelectionKey k : selected) {
                        if (k.isAcceptable()) {
                            SocketChannel sc = ((ServerSocketChannel) k.channel()).accept();

                            if (sc != null) {
                                DefaultHAService.log.info("HAService receive new connection, "
                                                          + sc.socket().getRemoteSocketAddress());
                                try {
                                    HAConnection conn = createConnection(sc);
                                    conn.start();
                                    DefaultHAService.this.addConnection(conn);
                                } catch (Exception e) {
                                    log.error("new HAConnection exception", e);
                                    sc.close();
                                }
                            }
                        }
                        ...
    }
}

public class AutoSwitchHAClient extends ServiceThread implements HAClient {
    ...
}

public interface HAClient {
    void start();

    void shutdown();

    void wakeup();

    void updateMasterAddress(String newAddress);

    void updateHaMasterAddress(String newAddress);

    String getMasterAddress();

    String getHaMasterAddress();

    long getLastReadTimestamp();

    long getLastWriteTimestamp();

    HAConnectionState getCurrentState();

    void changeCurrentState(HAConnectionState haConnectionState);

    void closeMaster();

    long getTransferredByteInSecond();
}

public interface HAConnection {
    void start();

    void shutdown();

    void close();

    SocketChannel getSocketChannel();

    HAConnectionState getCurrentState();

    String getClientAddress();

    long getTransferredByteInSecond();

    long getTransferFromWhere();

    long getSlaveAckOffset();
}

public abstract class AbstractHAReader {
    private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
    protected final List<HAReadHook> readHookList = new ArrayList<>();

    public boolean read(SocketChannel socketChannel, ByteBuffer byteBufferRead) {
        int readSizeZeroTimes = 0;
        while (byteBufferRead.hasRemaining()) {
            ...
            boolean result = processReadResult(byteBufferRead);
            ...
        }

    }

  ...

    protected abstract boolean processReadResult(ByteBuffer byteBufferRead);
}

@Override
protected boolean processReadResult(ByteBuffer byteBufferRead) {
int readSocketPos = byteBufferRead.position();
try {
    while (true) {
        ...
            switch (AutoSwitchHAClient.this.currentState) {
                case HANDSHAKE: {
                    ...
                    // 握手阶段，先检查commitlog完整性，截断
                }
                    break;
                case TRANSFER: {
                    // 传输阶段，将body写入commitlog
                    ...
                    byte[] bodyData = new byte[bodySize];
                    ...

                    if (bodySize > 0) {
                        // 传输阶段，将body写入commitlog
                        AutoSwitchHAClient.this.messageStore.appendToCommitLog(masterOffset, bodyData, 0, bodyData.length);
                    }
                    haService.updateConfirmOffset(Math.min(confirmOffset, messageStore.getMaxPhyOffset()));
                    ...
                    break;
                }
                default:
                    break;
            }
            if (isComplete) {
                continue;
            }

        }
      // 检查buffer中是否还有数据, 如果有, compact()
        ...
        break;
    }
}
...
}

abstract class AbstractWriteSocketService extends ServiceThread {
    ...

    private void transferToSlave() throws Exception {
        ...
        int size = this.getNextTransferDataSize();
        if (size > 0) {
            ...
            buildTransferHeaderBuffer(this.transferOffset, size);
            this.lastWriteOver = this.transferData(size);
        } else {
            // 无需传输，直接更新caught up的时间
            AutoSwitchHAConnection.this.haService.updateConnectionLastCaughtUpTime(AutoSwitchHAConnection.this.slaveId, System.currentTimeMillis());
            haService.getWaitNotifyObject().allWaitForRunning(100);
        }
    }

    @Override
    public void run() {
        AutoSwitchHAConnection.LOGGER.info(this.getServiceName() + " service started");

        while (!this.isStopped()) {
            try {
                this.selector.select(1000);

                switch (currentState) {
                    case HANDSHAKE:
                        // Wait until the slave send it handshake msg to master.
                        // 等待slave的握手请求，并进行回复
                        break;
                    case TRANSFER:
                        ...
                        transferToSlave();
                        break;
                    default: ...
                }
            } catch (Exception e) {
                ...
            }
        }
    ...
        // 在service结束后的一些事情
    }
    ...
}

@Override
protected boolean processReadResult(ByteBuffer byteBufferRead) {
while (true) {
    ...
    HAConnectionState slaveState = HAConnectionState.values()[byteBufferRead.getInt(readPosition)];
    switch (slaveState) {
        case HANDSHAKE:
            // 收到了client的握手
            ...
            LOGGER.info("Receive slave handshake, slaveBrokerId:{}, isSyncFromLastFile:{}, isAsyncLearner:{}",
                        AutoSwitchHAConnection.this.slaveId, AutoSwitchHAConnection.this.isSyncFromLastFile, AutoSwitchHAConnection.this.isAsyncLearner);
            break;
        case TRANSFER:
            // 收到了client的transfer状态
            ...
            // 更新client状态信息
            break;
        default:
            ...
    }
...
}

class RoleChangeHandler implements DLedgerLeaderElector.RoleChangeHandler {

    private final String selfId;
    private final ExecutorService executorService = Executors.newSingleThreadExecutor(new ThreadFactoryImpl("DLedgerControllerRoleChangeHandler_"));
    private volatile MemberState.Role currentRole = MemberState.Role.FOLLOWER;

    public RoleChangeHandler(final String selfId) {
        this.selfId = selfId;
    }

    @Override
    public void handle(long term, MemberState.Role role) {
        Runnable runnable = () -> {
            switch (role) {
                case CANDIDATE:
                    this.currentRole = MemberState.Role.CANDIDATE;
                    // 停止扫描inactive broker任务
                    ...
                case FOLLOWER:
                    this.currentRole = MemberState.Role.FOLLOWER;
                    // 停止扫描inactive broker任务
                    ...
                case LEADER: {
                    log.info("Controller {} change role to leader, try process a initial proposal", this.selfId);
                    int tryTimes = 0;
                    while (true) {
                        // 将会开始扫描inactive brokers
                        ...
                    break;
                }
            }
        };
        this.executorService.submit(runnable);
    }
  ...
}