kafka 心跳和 reblance

参数名                --> MemberMetadata 字段
session.timeout.ms   --> MemberMetadata.sessionTimeoutMs
max.poll.interval.ms --> MemberMetadata.rebalanceTimeoutMs

 1   private def completeAndScheduleNextHeartbeatExpiration(group: GroupMetadata, member: MemberMetadata) {
 2     // complete current heartbeat expectation
 3     member.latestHeartbeat = time.milliseconds()
 4     val memberKey = MemberKey(member.groupId, member.memberId)
 5     heartbeatPurgatory.checkAndComplete(memberKey)
 6 
 7     // reschedule the next heartbeat expiration deadline
 8     // 计算心跳截止时刻
 9     val newHeartbeatDeadline = member.latestHeartbeat + member.sessionTimeoutMs
10     val delayedHeartbeat = new DelayedHeartbeat(this, group, member, newHeartbeatDeadline, member.sessionTimeoutMs)
11     heartbeatPurgatory.tryCompleteElseWatch(delayedHeartbeat, Seq(memberKey))
12   } 
13   
14   // 心跳过期
15   def onExpireHeartbeat(group: GroupMetadata, member: MemberMetadata, heartbeatDeadline: Long) {
16     group.inLock {
17       if (!shouldKeepMemberAlive(member, heartbeatDeadline)) {
18         info(s"Member ${member.memberId} in group ${group.groupId} has failed, removing it from the group")
19         removeMemberAndUpdateGroup(group, member)
20       }
21     }
22   }
23   
24   private def shouldKeepMemberAlive(member: MemberMetadata, heartbeatDeadline: Long) =
25     member.awaitingJoinCallback != null ||
26       member.awaitingSyncCallback != null ||
27       member.latestHeartbeat + member.sessionTimeoutMs > heartbeatDeadline

if (coordinatorUnknown()) {
    if (findCoordinatorFuture != null || lookupCoordinator().failed())
        // the immediate future check ensures that we backoff properly in the case that no
        // brokers are available to connect to.
        AbstractCoordinator.this.wait(retryBackoffMs);
} else if (heartbeat.sessionTimeoutExpired(now)) {
    // the session timeout has expired without seeing a successful heartbeat, so we should
    // probably make sure the coordinator is still healthy.
    markCoordinatorUnknown();
} else if (heartbeat.pollTimeoutExpired(now)) {
    // the poll timeout has expired, which means that the foreground thread has stalled
    // in between calls to poll(), so we explicitly leave the group.
    maybeLeaveGroup();
} else if (!heartbeat.shouldHeartbeat(now)) {
    // poll again after waiting for the retry backoff in case the heartbeat failed or the
    // coordinator disconnected
    AbstractCoordinator.this.wait(retryBackoffMs);
} else {
    heartbeat.sentHeartbeat(now);

    sendHeartbeatRequest().addListener(new RequestFutureListener<Void>() {
        @Override
        public void onSuccess(Void value) {
            synchronized (AbstractCoordinator.this) {
                heartbeat.receiveHeartbeat(time.milliseconds());
            }
        }

        @Override
        public void onFailure(RuntimeException e) {
            synchronized (AbstractCoordinator.this) {
                if (e instanceof RebalanceInProgressException) {
                    // it is valid to continue heartbeating while the group is rebalancing. This
                    // ensures that the coordinator keeps the member in the group for as long
                    // as the duration of the rebalance timeout. If we stop sending heartbeats,
                    // however, then the session timeout may expire before we can rejoin.
                    heartbeat.receiveHeartbeat(time.milliseconds());
                } else {
                    heartbeat.failHeartbeat();

                    // wake up the thread if it‘s sleeping to reschedule the heartbeat
                    AbstractCoordinator.this.notify();
                }
            }
        }
    });
}

//maxPollInterval 即 rebalanceTimeoutMs 
public boolean pollTimeoutExpired(long now) {
    return now - lastPoll > maxPollInterval;
}