Group 状态机
在 0.9.0.0 之后的 Kafka，出现了几个新变动，一个是在 Server 端增加了 GroupCoordinator 这个角色，另一个较大的变动是将 topic 的 offset 信息由之前存储在 zookeeper 上改为存储到一个特殊的 topic 中（__consumer_offsets）。
offset 那些事
在 Kafka 中，无论是写入 topic，还是从 topic 读取数据，都免不了与 offset 打交道，关于 Kafka 的 offset 主要有以下几个概念，如下图。
其中，Last Committed Offset 和 Current Position 是与 Consumer Client 有关，High Watermark 和 Log End Offset 与 Producer Client 数据写入和 replica 之间的数据同步有关。
- Last Committed Offset：这是 group 最新一次 commit 的 offset，表示这个 group 已经把 Last Committed Offset 之前的数据都消费成功了；
- Current Position：group 当前消费数据的 offset，也就是说，Last Committed Offset 到 Current Position 之间的数据已经拉取成功，可能正在处理，但是还未 commit；
- Log End Offset：Producer 写入到 Kafka 中的最新一条数据的 offset；
- High Watermark：已经成功备份到其他 replicas 中的最新一条数据的 offset，也就是说 Log End Offset 与 High Watermark 之间的数据已经写入到该 partition 的 leader 中，但是还未成功备份到其他的 replicas 中，这部分数据被认为是不安全的，是不允许 Consumer 消费的
Topic __consumer_offsets
__consumer_offsets 是 Kafka 内部使用的一个 topic，专门用来存储 group 消费的情况，默认情况下有50个 partition，每个 partition 三副本，而具体 group 的消费情况要存储到哪一个 partition 上，是根据 abs(GroupId.hashCode()) % NumPartitions 来计算（其中，NumPartitions 是__consumer_offsets 的 partition 数，默认是50个）的。
GroupCoordinator
根据上面所述，一个具体的 group，是根据其 group 名进行 hash 并计算得到其具对应的 partition 值，该 partition leader 所在 Broker 即为该 Group 所对应的 GroupCoordinator，GroupCoordinator 会存储与该 group 相关的所有的 Meta 信息。
在 Broker 启动时，每个 Broker 都会启动一个 GroupCoordinator 服务，但只有 __consumer_offsets 的 partition 的 leader 才会直接与 Consumer Client 进行交互，也就是其 group 的 GroupCoordinator，其他的 GroupCoordinator 只是作为备份，一旦作为 leader 的 Broker 挂掉之后及时进行替代。
状态转移
Server 端，Consumer 的 Group 共定义了五个状态
- Empty：Group 没有任何成员，如果所有的 offsets 都过期的话就会变成 Dead，一般当 Group 新创建时是这个状态，也有可能这个 Group 仅仅用于 offset commits 并没有任何成员（Group has no more members, but lingers until all offsets have expired. This state also represents groups which use Kafka only for offset commits and have no members.）；
- PreparingRebalance：Group 正在准备进行 Rebalance（Group is preparing to rebalance）；
- AwaitingSync：Group 正在等待来 group leader 的 assignment（Group is awaiting state assignment from the leader）；
- Stable：稳定的状态（Group is stable）；
- Dead：Group 内已经没有成员，并且它的 Meta 已经被移除（Group has no more members and its metadata is being removed）。
其各个状态的定义及转换都在 GroupMetadata 中定义
Consumer 初始化时 group 状态变化
这里详述一下 Client 进行以上操作时，Server 端 Group 状态的变化情况。当 Consumer Client 首次进行拉取数据，如果该其所属 Group 并不存在时，Group 的状态变化过程如下：
- Consumer Client 发送 join-group 请求，如果 Group 不存在，创建该 Group，Group 的状态为 Empty；
- 由于 Group 的 member 为空，将该 member 加入到 Group 中，并将当前 member （client）设置为 Group 的 leader，进行 rebalance 操作，Group 的状态变为 preparingRebalance，等待
rebalance.timeout.ms之后（为了等待其他 member 重新发送 join-group，如果 Group 的状态变为preparingRebalance，Consumer Client 在进行 poll 操作时，needRejoin()方法结果就会返回 true，也就意味着当前 Consumer Client 需要重新加入 Group），Group 的 member 更新已经完成，此时 Group 的状态变为 AwaitingSync，并向 Group 的所有 member 返回 join-group 响应； - client 在收到 join-group 结果之后，如果发现自己的角色是 Group 的 leader，就进行 assignment，该 leader 将 assignment 的结果通过 sync-group 请求发送给 GroupCoordinator，而 follower 也会向 GroupCoordinator 发送一个 sync-group 请求（只不过对应的字段为空）；
- 当 GroupCoordinator 收到这个 Group leader 的请求之后，获取 assignment 的结果，将各个 member 对应的 assignment 发送给各个 member，而如果该 Client 是 follower 的话就不做任何处理，此时 group 的状态变为 Stable（也就是说，只有当收到的 Leader 的请求之后，才会向所有 member 返回 sync-group 的结果，这个是只发送一次的，由 leader 请求来触发）。
Consumer Rebalance
根据上图，当 group 在 Empty、AwaitSync 或 Stable 状态时，group 可能会进行 rebalance；
rebalance 的过程就是：等待所有 member 发送 join-group（上述过程的第2步），然后设置 Group 的 leader，进行 reassignment，各个 client 发送 sync-group 来同步 server 的 assignment 结果。
