Kafka笔记

1 介绍
2 设计思想
3 实现思路

1 介绍

Apache Kafka是一个分布式流处理平台。这到底意味着什么呢？

流处理平台有以下三种特性：

可以发布和订阅流式的记录，这一方面与消息队列或者企业消息系统类似；
可以存储流式的记录，并且有较好的容错性；
可以在流式记录产生时就进行处理。

Kafka适合什么样的场景？它可以用于两大类别的应用：

构造实时流数据管道，它可以在系统或应用之间可靠地获取数据。（相当于message queue）
构建实时流式应用程序，对这些流数据进行转换或者影响。（就是流处理，通过kafka stream topic和topic之间内部进行变化）

为了理解kafka是如何做到以上所说的功能，从下面开始，我们将深入探索Kafka的特性。

首先是一些概念：

Kafka作为一个集群，运行在一台或者多台服务器上。
Kafka通过topic对存储的流数据进行分类。
每条记录中包含一个key、一个value和一个timestamp（时间戳）。

Kafka有四个核心的API：

The Producer API允许一个应用程序发布一串流式的数据到一个或者多个Kafka topic。
The Consumer API允许一个应用程序订阅一个或者多个topic，并且对发布给他们的流式数据进行处理。
The Streams API允许一个应用程序作为一个流处理器，消费一个或者多个topic产生的输入流，然后生产一个输出流到一个或者多个topic中去，在输入输出流中进行有效的转换。
The Connector API允许构建并运行可重用的生产者或者消费者，将Kafka topics连接到已存在的应用程序或者数据系统。比如，连接到一个关系型数据库，捕捉表（table）的所有变更内容。

在Kafka中，客户端和服务器使用一个简单、高性能、支持多语言的TCP协议。此协议版本化并且向下兼容老版本，我们为Kafka提供了Java客户端，也支持许多其他语言的客户端。

1.1 Topics和日志

让我们首先深入了解下Kafka的核心概念：提供一串流式的记录-topic。

Topic就是数据主题，是数据记录发布的地方，可以用来区分业务系统。Kafka中的Topics总是多订阅者模式，一个topic可以拥有一个或者多个消费者来订阅它的数据。

对于每一个topic，Kafka集群都会维持一个分区日志。每个分区都是有序且顺序不可变的记录集，并且不断地追加到架构话的commit log文件。分区中的每一个记录都会分配一个id号来表示顺序，我们称之为offset，offset用来唯一的表时分区中每一条记录。

Kafka集群保留所有发布的记录，无论他们是否已被消费，并通过一个可配置的参数【保留期限】来控制。举个例子，如果保留策略设置为2天，一条记录发布后两天内，可以随时被消费，两天过后这条记录会被抛弃并释放磁盘空间。

Kafka的性能和数据大小无关，所以长时间存储数据没有什么问题。

事实上，在每一个消费者中保存的源数据就是offset（偏移量）即消费在log中的位置。偏移量由消费者所控制：通常在读取记录后，消费者会以线性的方式增加偏移量，但是实际上，由于这个位置由消费者控制，所以消费者可以采用任何顺序来消费记录。例如，一个消费者可以重置到一个旧的偏移量，从而重新处理过去的数据；也可以跳过最近的记录，从“现在”开始消费。

这些细节说明Kafka消费者是非常廉价的，消费者的增加和减少，对集群或者其他消费者没有多大的影响。比如，可以使用命令行工具，对一些topic内容执行tail操作，并不会影响已存在的消费者消费数据。

日志中的Partition（分区）有以下几个用途。第一，当日志大小超过了单台服务器的限制，允许日志进行扩展。每个单独的分区都必须受限于主机的文件限制，不过一个主题可能有多个分区，因此可以处理无限量的数据。第二，可以作为并行的单元集。

1.2 分布式

日志的分区Partition（分布）在Kafka集群的服务器上。每个服务器在处理数据和请求时，共享这些分区。每一个分区都会在以配置的服务器上进行备份，确保容错性。

每个分区都有一台server作为leader，零台或者多台server作为followers。leader server处理一切对partition（分区）的读写请求，而followers只需被动的同步leader上的数据。当leader宕机时，followers中的一台服务器回自动成为新的leader。每台server都会成为某些分区的leader和某些分区的follower，因此集群的负载是平衡的。

1.3 生产者

生产者可以将数据发布到所选择的topic（主题）中。生产者负责将记录分配到topic的哪一个partition（分区）中，可以使用循环的方式来简单地实现负载均衡，也可以根据某些予以分区函数（例如：记录中的key）来完成。下面会介绍更多关于分区的使用。

1.4 消费者

消费者使用一个消费组名称来进行标识，发布到topic中的每条记录被分配给订阅消费组中的一个消费者实例。消费者实例可以分布在多个进程中或者多个机器上。

如果所有的消费者实例在同一个消费组中，消息记录会负载平衡到每一个消费者实例。

如果所有的消费者实例在不同的消费组中，每条消息记录会广播到所有的消费者进程。

通常情况下，每个topic都会有一些消费组，一个消费组对应一个逻辑订阅者。一个消费组由许多消费者实例组成，便于扩展和容错。这就是发布和订阅的概念，只不过订阅者是一组消费者而不是单个的进程。

在Kafka中实现消费的方式是将日志中的分区划分到每一个消费者实例上，以便在任何时间，每个实例都是分区唯一的消费者。维护消费组中的消费关系由Kafka协议动态处理。如果新的实例加入组，他们将从组中其他成员处接管一些Partition分区；如果一个实例消失，拥有的分区将被分发到剩余的实例。

Kafka只保证分区内的记录是有序的，而不保证主题中不同分区的顺序。每个Partition分区按照Key值排序足以满足大多数应用程序的需求。但如果你需要总记录在所有记录的上面，可使用仅有一个分区的主题来实现，这意味着每个消费者组只有一个消费者进程。

1.5 高可用

hight-level Kafka给予以下保证：

生产者发送到特定topic partition的消息将按照发送的顺序处理。也就是说，如果记录M1和记录M2由相同的生产者发送，并先发送M1记录，那么M1的便宜比M2小，并在日志中较早出现；
一个消费者实例按照日志中的顺序查看记录；
对于具有N个副本的主题，最多容忍N-1个服务器故障，从而保证不会丢失任何提交到日志中的记录。

1.6 Kafka作为消息系统

Kafka streams的概念与传统的企业消息系统相比如何？

传统的消息系统有两个模块：队列和发布-订阅。在队列中，消费者池从server读取数据，每条记录被赤字中的一个消费者消费；在发布订阅中，记录被广播到所有的消费者。两者均有优缺点。

队列的优点在于它允许你将处理数据的过程分给多个消费者实例，使你可以扩展处理过程。不好的是，队列不是多订阅者模式的，一旦一个进程读取了数据，数据就会被丢弃。而发布-订阅系统允许广播数据到多个进程，但是无法进行扩展处理，因为每条消息都会发送给所有的订阅者。

消费组在Kafka有两层概念。在队列中，消费组允许你将处理过程分发给一系列进程（消费组中的成员）。在发布订阅中，Kafka允许你将消息广播给多个消费组。

Kafka的优势在于每个Topic都有以下特性，可以扩展处理并且允许多订阅者模式，不需要只选择其中一个。

Kafka相比于传统消息队列还具有更严格的顺序保证。

传统队列在服务器上保存有序的记录，如果多个消费者消费队列中的数据，服务器按照存储顺序输出记录。虽然服务器按顺序输出记录，但是记录被异步传递给消费者，因此记录可能会无序的到达不同的消费者。这意味着在并行消耗的情况下，记录的顺序是丢失的。因此消息系统通常使用“唯一消费者”的概念，即只让一个进程从队列中消费，但这就意味着不能够并行地处理数据。

Kafka设计的更好。Topic中的Partition是一个并行的概念。Kafka能够为一个消费者池提供顺序保证和负载平衡，是通过将Topic中的Partition分配给消费者组中的消费者来实现的，以便每个分区由消费组中的一个消费者消耗。通过这样，我们能够确保消费者是该分区的唯一读者，并按顺序消费数据。众多分区保证了多个消费者实例间的负载均衡，但请注意，消费者组中的消费者实例个数不能超过分区的数量。

1.7 Kafka作为存储系统

许多消息队列可以发布消息，处理消费消息之外还可以充当中间数据的存储系统。那么Kafka作为一个优秀的存储系统有什么不同呢？

数据写入Kafka后被写到磁盘，并且进行备份以便容错。直到完全备份，Kafka才让生产者认为完成写入，即使写入失败，Kafka也会确保继续写入。

Kafka使用磁盘结构，具有很好的扩展性，50kb和50TB的数据在server上表现一致。

可以存储大量数据，并且可通过客户端控制它读取数据的位置，可认为Kafka是一种高性能、低延迟、具备日志存储、备份和传播功能的分布式文件系统。

1.8 Kafka用做流处理

Kafka流处理不仅仅用来读写和存储流式数据，它最终的目的是为了能够进行实时的流处理。

在Kafka中，流处理器不断地从输入的topic获取流数据，处理数据后，再不断生产流数据到输出的topic中去。

例如，零售应用程序可能会接收销售和出货的输入流，经过价格调整计算后，再输出一串流式数据。

简单的数据处理可以直接用生产者和消费者的API，对于复杂的数据交换，Kafka提供了Streams API。Stream API允许应用做一些复杂的处理，比如将流数据聚合或者join。

这一功能有助于解决以下这种应用程序所面临的问题：处理无序数据，当消费端代码变更后重新处理输入，执行有状态计算等。

Streams API建立在Kafka的核心之上：它使用Producer和Consumer API作为输入，使用Kafka进行有状态的存储，并在流处理器实例之间使用相同的消费组机制来实现容错。

1.9 批处理

将消息、存储和流处理结合起来，使得Kafka看上去不一般，但这是它作为流平台所备的。

像HDFS这样的分布式文件系统可以存储用于批处理的静态文件，一个系统如果可以存储和处理历史数据是非常不错的。

传统的企业消息系统允许处理订阅后到达的数据。以这种方式来构建应用程序，并用它来处理即将到达的数据。

Kafka结合了上面所说的两种特性。作为一个流应用程序平台或者流数据管道，这两个特性，对于Kafka来说是至关重要的。

通过组合存储和低延迟订阅，流式应用程序可以以同样的方式处理过去和未来的数据。一个单一的应用程序可以处理历史记录的数据，并且可以持续不断地处理以后到达的数据，而不是在达到最后一条记录时结束进程。这是一个广泛的流处理概念，其中包含批处理一级消息驱动应用程序。

同样，作为流数据管道，能够订阅实时时间使得Kafka具有非常低的延迟；同时Kafka还具有可靠存储数据的特性，可用来存储重要的支付数据，或者与离线系统进行交互，系统可间歇性地加载数据，也可在停机维护后再次加载数据，流处理功能使得数据可以在到达时转换数据。

1.10 使用案例

1.10.1 消息

Kafka很好地替代了传统的message broker（消息代理）。Message brokers可用于各种场合（如将数据生成器与数据处理解耦，缓冲未处理的消息等）。与大多数消息系统相比，Kafka拥有更好的吞吐量、内置分区、具有复制和容错的功能，这使它成为一个非常理想的大型消息处理应用。

根据我们的经验，通常消息传递使用较低的吞吐量，但可能要求较低的端到端延迟，Kafka提供强大的持久性来满足这一要求。

在这方面，Kafka可以与传统的消息传递系统（ActiveMQ和RobbitMQ）相媲美。

1.10.2 跟踪网站活动

Kafka的初始用例是将用户活动跟踪管道重建为一组实时发布-订阅源。这意味着网站活动（浏览网页、搜索或者其他的用户操作）将被发布到中心topic，其中每个活动类型有一个topic。这些订阅源提供一系列用力，包括实时处理、实时监视、对加载到Hadoop或离线数据仓库系统的数据进行离线处理和报告等。

每个用户浏览网页时都生成了许多活动信息，因此活动跟踪的数据量通常非常大。

1.10.3 度量

Kafka通常用于监控数据。这设计到从分布式应用程序中汇总数据，然后生成可操作的集中数据源。

1.10.4 日志聚合

许多人使用Kafka来替代日志聚合解决方案。日志聚合系统通常从服务器收集物理日志文件，并将其置于一个中心系统（可能时文件服务器或HDFS）进行处理。Kafka从这些日志文件中提取信息，并将其抽象为一个更加清晰的消息流。这样可以实现耕地的延迟处理且易于支持多个数据源及分布式数据的消耗。与Scribe或Flume等以日志为中心的系统相比，Kafka具备同样出色的性能、更强的耐用性（因为复制功能）和更低的端到端延迟。

1.10.5 流处理

许多Kafka用户通过管道来处理数据，有多个阶段：从Kafka topic中消费原始输入数据，然后聚合、修饰或通过其他方式转化为新的topic，以供进一步消费或处理。例如，一个推荐新闻文章的处理管道可以从RSS订阅源抓取文章内容并将其发布到“文章”topic；然后对这个内容进行标准化或者重复的内容，并将处理完的文章内容发布到新的topic；最终它会尝试将这些内容推荐给用户。这种处理管道基于各个topic创建实时数据流图。从0.10.0.0开始，在Apache Kafka中，Kafka Streams可以用来执行上述的数据处理，它是一个轻量但功能强大的流处理库。除Kafka Streams外，可供替代的开源流处理工具还包括Apache Storm和Apache Samza。

1.10.6 采集日志

Event sourcing是一种应用程序设计风格，按时间来记录状态的更改。Kafka可以存储非常多的日志数据，为基于event sourcing的应用程序提供强有力的支持。

1.10.7 提交日志

Kafka可以从外部为分布式系统提供日志提交功能。日志有助于记录节点和行为间的数据，采用重新同步机制可以从失败节点恢复数据。Kafka的日志压缩功能支持这一用法。这一点与Apache BookKeeper项目类似。

2 设计思想

2.1 动机

我们设计的Kafka能够作为一个统一的平台来处理大公司可能拥有的所有实时数据推送，要做到这一点，我们必须考虑相当广泛的用例。

Kafka必须具有高吞吐量来支持高容量事件流，例如日志聚合。

Kafka需要能够正常处理大量的数据挤压，以便能够支持来自离线系统的周期性数据加载。

这也意味着系统必须处理低延迟分发，来处理更传统的消息传递用例。

我们希望支持对这些推送进行分区、分布式、以及实时处理来创建新的分发推送等特性。由此产生了我们的分区模式和消费者模式。

最后，在数据流被推送到其他数据系统进行服务的情况下，我们要求系统在出现机器故障时必须能够保证容错。

为支持这些使用场景，我们设计了一些独特的元素，使得Kafka相比传统的消息系统更像是数据库日志。

2.2 持久化

2.2.1 不要害怕文件系统！

Kafka对消息的存储和缓存严重依赖于文件系统，人们对与“磁盘速度慢”的普遍印象使得人们对于持久化的架构能够提供强有力的性能产生怀疑。事实上，磁盘的速度比人们于其的要慢得多，也快得多，这取决于人们使用磁盘的方式。而且设计合理的磁盘结构通常可以和网络一样快。

关于磁盘性能的关键事实是，磁盘的吞吐量和过去十年里磁盘的寻址延迟不同。因此，使用6个7200rpm、SATA接口、RAID-5的磁盘阵列在JBOD配置下的顺序写入的性能约为600MB/秒，但随即写入的性能约为100k/秒，相差6000倍以上。因为现象的读取和写入是磁盘使用模式中最有规律的，并且由操作系统进行了大量的优化。现代操作系统提供了read-ahead和write-behind技术，read-ahead是以大的data block为单位预先读取数据，而write-behind是将多个小型的逻辑写合并成一次大型的物理磁盘写入。关于该问题的进一步讨论可以参考ACM Queue article，他们发现实际上顺序磁盘访问在某些情况下比随机内存还要快！

为了弥补这种性能差异，现代操作系统在越来越注重使用内存对磁盘进行cache。现代操作系统主动将所有空闲内存用作disk caching，代价是内存回收时性能会有所降低。所有对磁盘的读写操作都会通过这个统一的cache。如果不适用直接I/O，该功能不能轻易关闭，因此即使进程维护了inprocess cache，该数据也可能会被复制到操作系统的pagecache中，事实上所有内容都被存储了两份。

此外，Kafka建立在JVM之上，任何了解Java内存使用的人都知道两点：

对象的内存开销非常高，通常是所存储的数据的两倍（甚至更多）
随着队中数据的增加，Java的垃圾回收变得越来越复杂和缓慢

受这些因素影响，相比于维护in-memory cache或者其他结构，使用文件系统和pagecache显得更有优势，我们可以通过自动访问所有空闲内存将可用缓存的容量至少翻倍，并且通过存储紧凑的字节结构而不是独立的对象，有望将缓存容量再翻已被，这样使得32GB的机器缓存容量可以达到28-31GB，并且不会产生额外的GC负担。此外，即使服务重新启动，缓存依旧可用，而in-process cache则需要在内存中重建（重建一个10GB的缓存可能需要10分钟），否则进程就要从cold cache的状态开始（这意味着进程最初的性能表现十分糟糕）。这同时也加大的简化了代码，因为所有保持cache和文件系统之间一致性的逻辑现在都被放到了OS中，这样做比一次性的进程内缓存更准确、更高效。如果磁盘使用更倾向于顺序读取，那么read-ahead可以有效的使用每次从磁盘中读取到的有用数据预先填充cache。

这里给出了一个非常简单的设计：相比于维护尽可能多的in-memory cache，并且在空间不足的时候匆忙将数据flush到文件系统，我们把这个过程倒过来。所有数据一开始就被写入到文件系统的持久化日志中，而不用在cache空间不足的时候flush到磁盘。实际上，这表明数据被转移到了内核的pagecache中。

2.2.2 常量时间就足够了

消息系统使用的持久化数据结构通常是和BTree相关联的消费者队列或者其他用于存储消息源数据的通用随机访问数据结构。BTree是最通用的数据结构，可以在消息系统能够支持各种事务性和非事务性语义。虽然BTree的操作复杂度是O(log N)，但成本也相当高。通常我们认为O(log N)基本等同于常数时间，但这条在磁盘操作中不成立。磁盘寻址是每10ms一跳，并且每个磁盘同时只能执行一次寻址，因此并行性受到了限制。因此即使是少量的磁盘寻址也会很高的开销。由于存储系统将非常快的cache操作和非常慢的物理磁盘操作混合在一起，当数据随着fixed cache增加时，可以看到树的性能通常是非线性的，比如数据翻倍时性能下降不只两倍。

所以直观来看，持久化队列可以建立在简单的读取和像文件追加两种操作之上，这和日志解决方案相同。这种架构的有点在于所有的操作复杂度都是O(1)，而且读操作不会阻塞写操作，读操作之间也不会互相影响。这有着明显的性能有事，由于性能和数据大小完全分离开来，服务器现在可以充分利用大量廉价、低转速的1TB SATA硬盘。虽然这些硬盘的寻址性能很差，但他们在大规模写方面的性能是可以接受的，而且价格是原来的三分之一、容量是原来的三倍。

在不产生任何性能损失的情况下能够访问几乎无限的硬盘空间，这意味着我们可以提供一些其他消息系统不常见的特性。例如：在Kafka中，我们可以让消息保留相对较长的一段时间（比如一周），而不是视图在被消费后立即删除。正如我们后面将要提到的，这给消费者带来了很大的灵活性。

2.3 Efficiency

我们在性能上以经做了很大的努力，我们主要的使用场景是处理web活动数据，这个数据量非常大，因为每个页面都有可能大量的写入，此外我们假设每个发布message至少被一个consumer（通常很多个consumer）消费，因此我们尽可能的去降低消费的代价。

我们还发现，从构建和运行许多相似系统的经验上来看，性能是多租户运营的关键。如果下游的基础设施服务很轻易被应用层冲击形成瓶颈，那么一些小的改变也会造成问题。通过非常快的（缓存）技术，我们能确保应用层冲击基础设施之前，将负载稳定下来。当尝试去运行支持集中式集群上成百上千个应用程序的集中式服务时，这一点很重要，因为应用层使用方式几乎每天都会发生变化。

我们在上一节讨论了磁盘性能，一旦消除了磁盘访问模式不佳的情况，该类系统性能低下的主要原因就剩下了两个：大量的小型I/O操作，以及过多的字节拷贝。

小型的I/O操作发生在客户端和服务端之间以及服务端的持久化操作中。

为了避免这种情况，我们的协议是建立在一个“消息块“的抽象基础上，合理将消息分组。这使得网络请求将多个消息打包成一组，而不是每次发送一条消息，从而使整组消息分担网络中往返的开销。Consumer每次获取多个大型有序的消息快，并由服务端一次将消息块一次加载到它的日志中。

这个简单的优化对速度有着数量级的提升，批处理允许更大的网络数据包，更大的顺序读写磁盘操作，连续的内存块等等，所有这些都使Kafka将随机流消息顺序写入到磁盘，再由consumers进行消费。

另一个低效率的操作是字节拷贝，在消息量少时，这不是什么问题，但是在高负载的情况下，影响就不容忽视。为了避免这种情况，我们使用producer、broker和consumer都共享的标准化的二进制消息格式，这样数据块不用修改就能在他们之间传递。

broker维护的消息日志本身就是一个文件目录，每个文件都由一系列以相同格式写入到磁盘的消息集合组成，这种写入格式被producer和consumer公用。保持这种通用格式可以对一些很重要的操作进行优化：持久化日志块的网络传输，现代的unix操作系统提供了一个高度优化的编码方式，用于将数据从pagecache转移到socket网络连接中，在linux中系统调用sendfile做到这一点。

为了理解sendfile的意义，了解数据从文件到套接字的创建数据传输路径就非常重要：

操作系统从磁盘读取数据到内核空间的pagecache；
应用程序读取内核空间的数据到用户空间的缓冲区；
应用程序将数据（用户空间的缓冲区）写回内核空间到套接字缓冲区（内核空间）；
操作系统将数据从套接字缓冲区（内核空间）复制到通过网络发送的NIC缓冲区。

这显然是低效的，有四次copy操作和两次系统调用，使用sendfile方法，可以允许操作系统将数据从pagecache直接发送到网络，这样避免重新复制数据，所以这种优化方式，只需要最后以部的copy操作，将数据复制到NIC缓冲区。

我们期望一个普遍的应用场景，一个topic被多消费者消费。使用上面提交的zero-copy（零拷贝）优化，数据在使用时只会被复制到pagecache中一次，节省了每次拷贝到用户空间内存中，再从用户空间进行读取到消耗。这使得消息能够以接近网络连接速度的上线进行消费。

pagecache和sendfile的组合使用意味着，在一个kafka集群中，大多数consumer消费时，将看不到磁盘上的读取活动，因为数据将完全由缓存提供。

Java中更多有关sendfile方法和zero-copy（零拷贝）相关的资料，可以参考网站http://www.ibm.com/developerworks/linux/library/j-zerocopy.

端到端的批量压缩

在某些情况下，数据传输的平静不是CPU，也不是磁盘，二十网络带宽。对于需要通过广域网在数据中心之间发送消息的数据管道尤其如此。当然，用户可以在不需要Kafka支持下一次一个的压缩消息。但是这样会造成非常差的压缩比和消息重复类型的荣誉，比如JSON中的字段名称或者是Web日志中的用户代理或公共字符串值。高性能的压缩是一次压缩多个消息，而不是压缩单个消息。

Kafka以高效的批处理格式支持一批消息可以压缩在一起发送到服务器。这批消息将以压缩格式写入，并且在日志中保持压缩，只会在consumer消费时解压缩。

Kafka支持GZIP、Snappy和LZ4压缩协议，更多有关压缩的资料参看地址：https://cwiki.apache.org/confluence/display/KAFKA/Compression.

2.4 The Producer

2.4.1 Load balancing

生产者直接发送数据到主分区的服务器上，不需要经过任何中间路由。为了让生产者实现这个功能，所有的kafka服务器节点都能响应这样的元数据请求：哪些服务器是活着的，主题的哪些分区是主分区，分配在哪个服务器上，这样生产者就能适当地直接发送它的请求到服务器上。

客户端控制消息发送数据到哪个分区，这个可以实现随机的负载均衡方式或者使用一些特定语义的分区函数。我们有提供特定分区的接口让用于根据指定的键值进行hash分区（当然也有选项可以重写分区函数），例如，如果使用用户ID作为key，则用户相关的所有数据都会被分发到同一个分区上。这允许消费者在消费数据时做一些特定的本地化处理，这样的分区风格经常被设计用于一些本地处理比较敏感的消费者。

2.4.2 Asynchronous send

批处理是提升性能的一个主要驱动，为了允许批处理，kafka生产者会尝试在内存中汇总数据，并用一次请求批次提交信息。批处理，不仅仅可以配置指定的消息数量，也可以指定等待特定的延迟时间（如64k或10ms），这允许汇总更多的数据后再发送，在服务器端也会减少更多的IO操作。该缓冲是可配置的，并给出了一个机制，通过权衡少量额外的延迟时间获取更好的吞吐量。

2.5 消费者

Kafka consumer通过向broker发出一个fetch请求来获取它想要消费的partition。consumer的每个请求都在log中指定了对应的offset，并接收从该位置开始的一大块数据。因此，consumer对于该位置的控制就显得极为重要，并且可以在需要的时候通过回退到该位置再次消费对应的数据。

2.5.1 Push vs. pull

最初我们考虑的问题是：究竟是由consumer从broker那里pull数据，还是由broker将数据push到consumer。Kafka在这方面采取了一种较为传统的设计方式，也是大多数的消息系统所共享的方式：即producer把数据push到broker，然后consumer从broker中pull数据。也有一些logging-centric的系统，比如Scribe和Apache Flume，沿着一条完全不同的push-based的系统中，当消费速率地域生产速率时，consumer往往会不堪重负（本质上类似于拒绝服务攻击）。pull-based系统有一个很好的特性，那就是当consumer速率落后于producer时，可以在适当的时间赶上来。还可以通过使用某种backoff协议来减少这种现象：即consumer可以通过backoff表时它以经不堪重负了，然而通过获得负载情况来充分使用consumer（但永远不超载）这一方式实现起来比它看起来更棘手。前面以这种方式构建系统的尝试，引导着Kafka走向了更传统的pull模型。

另一个pull-based系统的优点在于：它可以大批量生产要发送给consumer的数据。而push-based系统必须选择立即发送请求或者积累更多的数据，然后在不知道下游的consumer能否立即处理它的情况下发送这些数据。如果系统调整为低延迟状态，这就会导致一次只发送一条消息，以至于传输的数据不再被缓冲，这种方式是极度浪费的。而pull-based得设计修复了该问题，因为consumer总是将所有可用得（或者达到配置得最大长度）消息pull到log当前位置的后面，从而使得数据能够得到最佳的处理而不会引入不必要的延迟。

简单的pull-based系统的不足之处在于：如果broker中没有数据，consumer可能会在一个紧密的循环中结束轮询，实际上busy-waiting直到数据到来。为了避免busy-waiting，我们在pull请求中加入参数，使得consumer在一个“long pull”中阻塞等待，直到数据到来（还可以选择等待给定字节长度的数据来确保传输长度）。

你可以想象其他可能的只基于pull的，end-to-end的设计。例如producer直接将数据写入一个本地的log，然后broker从producer那里pull数据，最后consumer从broker中pull数据。通常提到的还有“store-and-forward”式producer，这是一种很有趣的设计，但我们觉得它跟我们设定的有数以千计的生产者的应用场景不太相符。我们在运行大规模持久化数据系统方面的经验使我们感觉到，横跨多个应用、涉及数千磁盘的系统事实上并不会让事情更可靠，反而会成为操作时的噩梦。在实践中，我们发现可以通过大规模运行的带有强大的SLAs的pipeline，而省略Producer的持久化过程。

2.5.2 消费者的位置

令人惊讶的是，持续追踪以经被消费的内容是消息系统的关键性能点之一。

大多数消息系统都在broker上保存被消费消息的元数据，也就是说，当消息被传递给consumer，broker要么立即在本地记录该事件，要么等待consumer的确认后再记录。这是一种相当直接的选择，而且事实上对于单机服务器来说，也没与其他地方能够存储这些状态信息。由于大多数消息系统用于存储的数据结构规模都很小，所以这也是一个很实用的选择，因为只要broker知道哪些消息被消费了，就可以在本地立即进行删除，一致保持较小的数据量。

也许不太明显，但要让broker和consumer就被消费的数据保持一致性也不是一个小问题。如果broker在每条消息被发送到网络的时候，立即将其标记为consumed，那么一旦consumer无法处理该消息（可能由consumer崩溃或者请求超时或者其他原因导致），该消息就会丢失。为了解决消息丢失的问题，许多消息系统增加了确认机制：即当消息被发送出去的时候，消息仅被发送出去的时候，消息仅被标记为sent而不是consumed；然后broker会等待一个来自consumer的特定确认，再将消息标记为consumed。这个策略修复了消息丢失的问题，但也产生了新问题。首先，如果consumer处理了消息但在发送确认之前出钱了，那么该消息就会被消费两次。第二个是关于性能的，现在broker必须为每条消息保存多个状态（首先对其枷锁，确保该消息只被发送一次，然后将其永久的标记为consumed，以便将其移除）。还有更棘手的问题要处理，比如如何处理以经发送但一直得不到确认的消息。

Kafka使用完全不同的方式解决消息丢失问题。Kafka的topic被分割成了一组完全有序的partition，其中每一个partition在任意给定的时间内只能被每个订阅了这个topic的consumer组中的一个consumer消费。这意味着partition中每一个consumer的位置仅仅是一个数字，即下一条要消费的消息的offset。这使得被消费的消息的状态信息相当少，每个partition只需要一个数字。这个状态信息还可以作为周期性的checkpoint。这以非常低的代价实现了和消息确认机制等同的效果。

这种方式还有一个附加的好处。consumer可以回退到之前的offset来再次消费之前的数据，这个操作违反了队列的基本原则，但事实证明对大多数consumer来说这是一个必不可少的特性。例如，如果consumer的代码有bug，并且在bug被发现前以经有一部分数据被消费了，那么consumer可以在bug修复后通过回退到之前的offset来再次消费这些数据。

2.5.3 离线数据加载

可伸缩的持久化特性允许consumer只进行周期性的消费，例如批量数据加载，周期性将数据加载到褚如Hadoop和关系型数据库之类的离线系统中。

在Hadoop的应用场景中，我们通过将数据加载分配到多个独立的map任务来实现并行化，每一个map任务负责一个node/topic/partition，从而达到充分并行化。Hadoop提供了任务管理机制，失败的任务可以重新启动而不会有重复数据的风险，只需要简单的从原来的位置重启即可。

2.6 消息交付语义

现在我们对于producer和consumer的工作原理以经有了一点了解，让我们接着讨论Kafka在producer和consumer之间提供的语义保证。显然，Kafka可以提供的消息交付语义保证有多种：

At most once：消息可能会丢失但绝不重传；
At least once：消息可以重传但绝不丢失；
Exactly once：这正是人们想要的，每一条消息只被传递一次。

值得注意的是，这个问题被分成了两部分：发布消息的持久性保证和消费消息的保证。

很多系统声称提供了“Exactly once”的消息交付语义，然而阅读他们的细则很重要，因为这些声称大多数都是误导性的，即他们没有考虑consumer或producer可能失败的情况，以及存在多个consumer进行处理的情况，或者写入磁盘的数据可能丢失的情况。

Kafka的语义是直截了当的。发布消息时，我们会有一个消息的概念被"committed"到log中。一旦消息被提交，只要有一个broker备份了该消息写入的Partition，并且保持“alive”装填，该消息就不会丢失。有关commited message和alive partition的定义，以及我们试图解决的故障类型都将在下一节进行细致描述。现在让我们假设存在完美无缺的broker，然后来试着理解Kafka对producer和consumer的语义保证。如果一个producer在试图发送消息的时候发生了网络故障，则不确定网络错误发生在消息提交之前还是之后，这与使用自动生成的键插入到数据库表中的语义场景很相似。

在0.11.0.0之前的版本中，如果producer没有收到表明消息以经被提交的相应，那么producer除了将消息重传之外别无选择。这里提供的时at-least-once的消息交付语义，因为如果最初的请求事实上执行成功了，那么重传过程中该消息就会被再次写入到log当中。从0.11.0.0版本开始，Kafka producer新增了幂等性的传递选项，该选项保证重传不会在log中产生重复条目。为实现这个目的，broker给每个producer都发分配了一个ID，并且producer给每条被发送的消息分配了一个序列号来避免重复的消息。同样也是从0.11.0.0版本开始，producer新增了使用类似事务性的语义将消息发送到多个topic partition的功能：也就是说，要么所有的消息都被成功的写入到了Log，要么一个都没写进去。这种语义的主要应用场景就是Kafka topic之间的exactly-once的数据传递（如下所述）。

并非所有的使用场景都需要这么强的保证，对于延迟敏感的应用场景，我们允许生产者指定它需要的持久性级别。如果producer指定了它想要等待消息被提交，则可以使用10ms的量级。然而，producer也可以指定它想要完全异步地执行发送，或者它只想等待知道Leader节点拥有该消息（follower节点有没有无所谓）。

现在让我们从consumer的视角来描述语义。所有的副本都有相同的log和相同的offset，consumer负责控制它在log中的位置。如果consumer永远不崩溃，那么它可以将这个位置信息只存储在内存中。但如果consumer发生了故障，我们希望这个topic partition被另一个进程接管，那么新进程需要选择一个合适的位置开始进行处理。假设consumer要读取一些消息，它有几个处理消息和更新位置的选项：

Consumer可以先读取消息，然后将它的位置保存到log中，然后再对消息进行处理。在这种情况下，消费者进程可能会在保存其位置之后，还没有处理消息之前发生崩溃。而在这种情况下，即使在此位置之前的一些消息没有被处理，接管处理的进程将从保存的位置开始，在consumer发生故障的情况下，这对应于“at-most-once”的语义，可能会有消息得不到处理；
Consumer可以先读取消息，然后处理消息，最后再保存它的位置。在这种情况下，消费者进程可能会在处理了消息之后，但还没有保存位置之前发生崩溃。而在这种情况下，当新的进程接管后，它最初收到的一部分消息都已经被处理过了。在consumer发生故障的情况下，这对应于”at-least-once“的语义。在许多应用场景中，消息都设有一个主键，所以更新操作时幂等的（相同的消息接收两次时，第二次写入会覆盖掉第一次写入的记录）。

那么exactly once语义（即真正想要的东西）呢？当从一个kafka topic中消费到另一个topic时（正如在一个Kafka Streames应用中所做的那样），我们可以使用上文提到的0.11.0.0版本中的新事务型producer，并将consumer的位置存储为一个topic中的消息，所以我们可以在输出topic接收以经被处理的数据的时候，在同一个事务中向kafka写入offset。如果事务被中断，则消费者的位置将恢复到原来的值，而输出topic上产生的数据对其他消费者是否可见，取决于事务的”隔离级别“。在默认的”read-uncommitted“隔离级别中，所有消息对consumer都是可见的，即使他们是中止的事务的一部分，但是在”read-committed“的隔离级别中，消费者只能访问已提交的事务中的消息（以及任何不属于事务的消息）。

在写入外部系统的应用场景中，限制在于需要在consumer的offset与实际存储为输出的内容间进行协调。解决这一问题的经典方法是在consumer offset的存储于其输出相同的位置。这也是一种更好的方式，因为大多数consumer想写入的输出系统都不支持two-phase commit。举个例子，Kafka Connect连接器，它将所读取的数据和数据的offset一起写入到HDFS，以保证数据和offset都被更新，或者两者都不被更新，对于其他很多需要这些较强语义，并且没有主键来避免消息重复的数据系统，我们也遵循类似的模式。

因此，事实上Kafka在Kafka Streams中支持了exactly-once的消息交付功能，并且在topic之间进行数据传递和处理时，通常使用事务型producer/consumer提供exactly-once的消息交付功能。到其他目标系统的exactly-once的消息交付通常需要与该类系统写作，但Kafka提供了offset，使得这种应用场景的实现变得可行。（详见Kafka Connect）。否则，Kafka默认保证at-least-once的消息交付，并且kafka允许用户通过禁用producer的重传功能和让consumer在处理一批消息之前提交offset，来实现at-most-once的消息交付。

2.7 Replication

Kafka允许topic的partition拥有若干副本，你可以在server端配置partition的副本数量。当集群中的节点出现故障时，能自动进行故障转移，保证数据的可用性。

其他的消息系统也提供了副本相关的特性，但是在我们（带有偏见）看来，他们的副本功能不常用，而且有很大缺点：slaves处于非活动状态，导致吞吐量收到验证影响，并且还要手动配置副本机制。Kafka默认使用本分机制，事实上，我i们将没有设置副本数的topic时限为副本数为1的topic。

创建副本的单位是topic的partition，正常情况下，每个分区都有一个Leader和零或多个followers。总的副本数是包含leader的综合。所有的读写操作都由leader处理，一般partition的数量都比broker的数量多很多，个分区的leader均匀的分布在brokers中。所有的followers节点都同步leader节点的日志，日志中的消息和偏移量都和leader中的一致。（当然，在任何给定的时间，leader节点的日志末尾可能有几个消息尚未被备份完成）。

Follows节点就像普通的consumer那样从leader节点那里拉取消息并保存在自己的日志文件中。Follows节点可以从leader节点那里批量拉取消息日志到自己的日志文件中。

与大多数分布式系统一样，自动处理故障需要精确定义节点”alive“的概念。Kafka判断节点是否存活有两种方式：

节点必须可以维护和ZooKeeper的连接，ZooKeeper通过心跳机制检查每个节点的连接；
如果节点是个follower，它必须能及时的同步leader的写操作，并且延时不能太久。

我们认为满足这两个条件的节点处于”in sync“状态，区别与”alive“和”failed“。Leader会追踪左右”in sync“的节点。如果有节点挂掉了，或是写超时，或是心跳超时，leader就会把它从同步副本列表中移除。同步超时和写超时的时间由replica.lag.time.max.ms配置确定。

分布式系统中，我们只尝试处理”fail/recover“模式的故障，即节点突然停止工作，然后又恢复（节点可能不知道自己曾经挂掉）的状况。Kafka没有处理所谓的”Byzantine“故障，即一个节点出现了随意相应和恶意相应（可能由于bug或非法操作导致）。

现在，我们可以更精确地定义，只有当消息被所有的副本节点加入到日志中时，才算是提交，只有提交的消息才会被consumer消费，这样就不用担心一旦leader挂掉了消息会丢失。另一方面，producer也可以选择是否等待消息被提交，这取决于他们的设置在延迟时间和持久性之间的权衡，这个选项是由producer使用的acks设置控制。请注意，topic可以设置同步备份的最小数量，producer请求确认消息是否写入到所有的备份时，可以用最小同步数量判断，如果producer对同步的备份数没有严格的要求，即使同步的备份数量低于最小同步数量（例如，仅仅只有leader同步了数据），消息也会被提交，然后被消费。

在所有时间里，Kafka保证只要有至少一个同步中的节点存活，提交的消息就不会丢失。

节点挂掉后，经过短暂的故障转移后，Kafka将仍然保持可用性，但在网络分区（network partitions）的情况下可能不能保持可用性。

2.7.1 备份日志：Quorums，ISRs，状态机

Kafka的核心是备份日志文件，备份日志文件是分布式数据系统最基础的要素之一，实现方法也有很多种。其他系统也可以用Kafka的备份日志模块来实现状态机风格的分布式系统。

备份日志按照一系列有序的值（通常是编号为0、1、2、...）进行建模，有很多种方法可以实现这一点，但最简单和最快的方法是由leader节点选择需要提供的有序的值，只要leader节点还存货，所有的follower只需要拷贝数据并按照leader节点的顺序排序。

当然，如果leader永远都不会挂掉，那我们就不需要follower了。但是如果leader crash，我们就需要从follower中选举出一个新的leader。但是followers自身也有可能落后或者crash，所以我们必须确保leader的候选者们是一个数据同步最新的follower节点。

如果选择写入的时候需要保证一定数量的副本写入成功，读取时需要保证读取一定数量的副本，读取和写入之间有重叠，这样的读写机制成为Quorum。

这种权衡的一种常见方法时对提交决策和leader选举使用多数投票机制。Kafka没有采取这种方式，但是我们还是要研究以下这种投票机制，来理解其中蕴含的权衡。假设我们有2f+1个副本，如果在leader宣布消息提交之前必须有f+1个副本收到该消息，并且如果我们从这者少f+1个副本之中，有着最完整日志记录的follower里来选择一个新的leader，那么在故障次数少于f的情况下，选举出的leader保证具有所有提交的消息。这是因为在任意f+1个副本中，至少有一个副本一定包含了所有提交的消息。该副本的日志将是最完整的，因此将被选为新的leader。这个算法都必须处理许多其他细节（例如精确定义怎样使日志更加完整，确保在leader down掉期间，保证日志一致性或者副本服务器的副本集的改变），但是现在我们将忽略这些细节。

这种大多数投票方法有一个非常好的有点：延迟是取决于最快的服务器，也就是说，如果副本数是3，则备份完成的等待时间取决于最快的Follower。

这里有很多分布式算法，包含ZooKeeper的Zab、Raft、View stamped Replication。我们所知道的与Kafka实际执行情况最相似的学术刊物是来自微软的PacificA。

大多数投票的缺点是，多数的节点挂掉让你不能选择leader。要荣誉单点故障需要三份数据，并且要荣誉两个故障需要五份数据。根据我们的经验，在一个系统中，仅仅靠荣誉来避免单点故障是不够的，但是每写5此，对磁盘空间需求是5倍，吞吐量下降到1/5，这对于处理海量数据问题是不切实际的。这可能是为什么quorum算法更常用于共享集群配置（如ZooKeeper），而不适用于原始数据存储的原因，例如HDFS中namenode的高可用是建立在基于投票的元数据，这种代价高昂的存储方式不适用数据本身。

Kafka采取了一种稍微不同的方法来选择它的投票集，Kafka不是用大多数投票选择leader。Kafka动态维护了一个同步状态的备份的集合（a set of in-sync replicas），简称ISR，在这个集合中的节点都是和leader保持高度一致的，只有这个集合的成员才有资格被选举为leader，一条消息必须被这个集合所有节点读取并追加到日志中，这条消息才能视为提交。这个ISR集合发生变化会在ZooKeeper持久化，正因为如此，这个集合中的任何一个节点都有资格被选为leader。这对于Kafka使用模型中，有很多分区并确保主从关系是很重要的。因为ISR模型和f+1副本，一个Kafka topic冗余f个节点故障而不会丢失任何已经提交的消息。

我们认为对于希望处理的大多数场景这种策略是合理的。在实际中，为了冗余f节点故障，大多数投票和ISR都会在提交消息前确认相同数量的备份被收到（例如在一次故障恢复之后，大多数的quorum需要三个备份节点和一次确认，ISR只需要两个备份节点和一次确认），多数投票方法的一个优点是提交时能避免最慢的服务器。但是，我们认为通过允许客户端选择是否阻塞消息提交来改善，和所需的备份数较低而产生额外的吞吐量和磁盘空间是值得的。

另一个重要的设计区别是，Kafka不要求崩溃的节点恢复所有的数据，在这种空间中的复制算法京城依赖于存在”稳定存储“，在没有违反潜在的一致性的情况下，出现任何故障再恢复情况下都不会丢失。这个假设有两个主要的问题。首先，我们在持久性数据系统的实际操作中观察到的最常见的问题是磁盘错误，并且他们通常不能保证数据的完整性。其次，即使磁盘错误不是问题，我们也不希望在每次写入时都要求使用fsync来保证一致性，因为这会使性能降低两到三个数量级。我们的协议能确保备份节点重新加入ISR之前，即使它挂时没有新的数据，它也必须完整再一次同步数据。

2.7.2 Unclean leader选举：如果节点全挂了？

请注意，Kafka对于数据不会丢失的保证，是基于至少一个节点在保持同步状态，一旦分区上的所有备份节点都挂了，就无法保证了。

但是，实际在运行的系统需要去考虑假设一旦所有的备份都挂了，怎么去保证数据不会丢失，这里有两种实现的方法：

等待一个ISR的副本重新恢复正常服务，并选择这个副本作为Leader（它有极大可能拥有全部数据）；
选择第一个重新恢复正常服务的副本（不一定是ISR中的）作为leader；

这是可用性和一致性之间的简单拖鞋，如果只等待ISR的备份节点，那么只要ISR备份节点都挂了，我们的服务将一直会不可用，如果他们的数据损坏了或者丢失了，那就回事长久的宕机。另一方面，如果不是ISR中的节点恢复服务并且我们允许它成为leader，那么它的数据就是可信的来源，即使它不能保证记录了每一个以经提交的消息。Kafka默认选择第二种策略，当所有的ISR副本都挂掉时，会选择一个可能不同步的备份作为leader，可以配置属性unclean.leader.election.enable禁用此策略，那么就会使用第一种策略即停机时间由于不同步。

这种困境不只有Kafka遇到，它存在于任何quorum-based规则中。例如，大多是投票算法中，如果大多数服务器永久性的挂了，那么要么选择丢失100%的数据，要么违背数据的一致性选择一个存活的服务器作为数据可信的来源。

2.7.3 可用性和持久性保证

向Kafka写数据时，Producers设置ack是否提交完成：

0：不等待broker返回确认消息；
1：leader保存成功返回
-1（all）：所有备份都保存成功返回。

请注意，设置ack=all并不能保证所有的副本都写入了消息。默认情况下，当ack=all时，只要ISR副本同步完成，就会返回消息已经写入。例如，一个topic仅仅设置了两个副本，那么只有一个ISR副本，那么当设置ack=all时返回写入成功，剩下的那个副本可能数据没有写入。尽管这确保了分区的最大可用性，但是对于偏好数据持久性而不是可用性的一些用户，可能不想用这种策略，因此，我们提供了两个topic配置，可用于配置消息数据持久性：

禁用unclean leader选举机制，如果所有的备份节点都挂了，分区数据就会不可用，知道最近的leader恢复正常，这种策略优先于数据丢失的风险，参看上一节的unclean leader选举机制；
指定最小的ISR集合大小，只有当ISR的大小大于最小值，分区才能接受写入操作，以防止仅写入单个备份的消息丢失造成消息不可用的情况，这个设置只有在生产者使用ack=all的情况下才会生效，这至少保证消息被ISR副本写入。此设置是一致性和可用性之间的折中，对于设置更大的最小ISR大小保证了更好的一致性，因为它保证将消息被写入了更多的备份，减少消息丢失的可能性。但是，这回降低可用性，因为如果ISR副本的数量低于最小阈值，那么分区将无法写入。

2.7.4 备份管理

以上关于备份日志的讨论只涉及单个日志文件，即一个topic分区，事实上，一个Kafka集群管理者成百上千个这样的partitions。我们尝试以轮询调度的方式将集群内的Partition负载均衡，避免大量topic拥有的分区集中在少数几个节点上。同样，我们也试图平衡leadership，以至于每个节点都是部分partition的leader节点。

优化主从关系的选举过程也是重要的，这是数据不可用的关键窗口。原始的实现是当有节点挂了后，进行主从关系选举时，会对挂掉节点的所有partition的领导权重新选举。相反，我们会选择一个broker作为controller节点。controller节点负责检测brokers级别故障，并负责在broker腹胀的情况下更改这个故障broker中的partition的leadership。这种方式可以批量的通知主从关系的变化，使得对于拥有大量partition的broker，选举过程的代价更低并且速度更快。如果controller节点挂了，其他存活的broker都可能成为新的controller节点。

2.8 日志压缩

日志压缩可确保kafka始终至少为单个topic partition的数据日志中的每个message key保留最新的已知值。这样的设计解决了应用程序崩溃、系统故障后恢复或者应用在运行维护过程中重启后重新加载缓存的场景，接下来让我们深入讨论这些在使用过程中的更多细节，阐述在这个过程中它是如何进行日志压缩的。

迄今为止，我们只介绍了简单的日志保留方法（当旧的数据保留时间超过指定时间、日志大小达到规定大小后就丢弃）。这样的策略非常适用于处理哪些暂存的数据，例如记录每条消息之间互相独立的日志。然而在实际使用过程中还有一种非常重要的场景——根据key进行数据变更（例如更改数据库表内容），使用以上方式显然不行。

让我们来讨论一个关于处理这样流式数据的具体例子。假设我们有一个topic，里面的内容包含用户的email地址；每次用户更新他们的email地址时，我们发送一条消息到这个topic，这里使用用户id作为消息的key值。现在，我们在一段时间内为id为123的用户发送一些消息，每个消息对应email地址的改变（其他ID消息省略）：

123 => bill@microsoft.com
.
.
.
123 => bill@gatesfoundation.org
.
.
.
123 => bill@gamil.com

日志压缩为我们提供了更精细的保留机制，所以我们至少保留每个key的最后一次更新（例如：bill@gmail.com)。这样我们保证日志包含每一个key的最终值而不是最近变更的完整快照数据，这意味着下游的消费者可以获得最终的状态而无需拿到所有的变化消息信息。

让我们先看几个有用的使用场景，然后再看看如何使用：

数据库更改订阅。通常需要在多个数据系统设置拥有一个数据集，这些系统中通常有一个是某种类型的数据库（无论是RDBMS或者辛流形的key-value数据库）。例如，我们有一个数据库、缓存、搜索引擎集群或者Hadoop集群，每次变更数据库，也同时需要变更缓存、搜索引擎以及hadoop集群。在只需处理最新日志实时更新的情况下，只需要最近的数据记录。但是，如果系统能够重新加载缓存或者恢复搜索失败的节点，可能需要一个完整的数据集。
事件源。这是一种应用程序设计风格，它将查询处理与应用程序设计相结合，并使用变更的日志作为应用程序的主要存储。
日志高可用。执行本地计算的进程可以通过注销对其本地状态所作的更改来实现容错，以便另一个进程可以重新加载这些更改并在出现故障时继续进行。一个具体的例子就是在流查询系统中进行计数，聚合和其他类似group by的操作。实时流处理框架Samza，使用这个特性正是出于这一原因。

在这些场景下，主要需要处理变化的实时feed，但是偶尔当机器崩溃或需要重新加载、重新处理数据时，需要处理所有数据，日志压缩允许在同一topic下同时使用这两个用例。

想法很简单，我们有无线的日志，以上每种情况记录变更日志，我们从一开始就捕获每一次变更。使用这个完整的日志，可以通过回放日志来恢复到任何一个时间点的状态。然而这种假设的情况下，完整的日志是不实际的，对于哪些每一行记录都会变更多次的系统，即使数据集很小，日志也会无限的增长下去。丢弃旧日志的简单操作可以限制空间的增长，但是无法重建状态，因为旧的日志被丢弃，可能一部分记录的状态会无法重建（这些记录所有的状态变更都在旧日志中）。

日志压缩机制是更细粒度的，每个记录都保留的机制，而不是基于时间的粗粒度。这个理念是选择性的删除那些有更新的变更记录的日志，这样最终日志至少包含每个key的最后一个状态。

这个策略可以为每个Topic设置，这样一个集群中，可以一部分Topic通过时间和大小保留日志，另外一些可以通过压缩策略保留。

这个功能的令该来自于LinkedIn的最古老且成功的基础设置，一个成为Databus的数据库变更日志缓存系统。不像大多数的日志存储系统，Kafka是专门为订阅和快速线性的读和写的组织数据。和Databus不同，Kafka作为真实的存储，压缩日志是非常有用的，这非常有利于上有数据不能重发的情况。

2.8.1 日志压缩基础

Log head中包含传统的Kafka日志，它包含了联系的offset和所有的消息。日志压缩增加了处理tail Log的选项，tail中的消息保存了初次写入时的offset。即使该offset的消息被压缩，所有offset仍然在日志中是有效的，在这个场景中，无法区分下一个出现的更高offset的位置。

压缩也允许删除，通过消息的Key和空负载（null payload）来标识该消息可从日志中删除，这个删除标记将会引起所有之前拥有相同key的消息被移除（包括拥有key相同的新消息）。但是删除标记比较特殊，它将在一定周期后被从日志中删除来释放空间，这个时间点被成为delete retention point。

压缩操作通过在后台周期性的拷贝日志段来完成，清楚操作不会阻塞读取，并且可以被配置不超过一定IO吞吐来避免影响Producer和Consumer。

2.8.2 What guarantees does log conpaction provide?

日志压缩的保障措施如下：

任何滞留在日志head中的所有消费者能看到写入的所有消息；这些消息都是有序的offset，topic使用min.compaction.lag.ms来保障消息写入之前必须经过的最小时间长度，才能被压缩，这限制了一条消息在log head中的最短存在时间；
始终保持消息的有序性，压缩永远不会重新排序消息，只是删除了一些；
消息的offset不会变更，这是消息在日志中的永久标志；
任何从头开始处理日志的Consumer至少会拿到每个key的最终状态，另外，只要Consumer在小于topic的delete.rentention.ms设置（默认24小时）的时间段内到达log head，将会看到所有删除记录的所有删除标记，换句话说，因为移除删除标记和读取是同时发生的，Consumer可能会因为落后超过delete.rentention.ms而导致错误删除标记。

2.8.3 日志压缩的细节

日志压缩由log cleaner执行，后台线程池重新拷贝日志段，移除哪些key存在于log head中的记录，每个压缩线程如下工作：

选择log head与log tail比率最高的日志；
在log head中为每个key的最后offset创建一个简单概要；
它从日志的开始到结束，删除那些在日志中最新出现的Key的旧值，新的、干净的日志将会立即被移交到日志中，所以只需要一个额外的日志段空间（不是日志的完整副本）；
日志head的概要本质上是一个空间密集型的哈希表，每个条目使用24个字节，所以如果由8G的整理缓冲区，则能迭代处理大约366G的日志头部（假设消息大小为1k）。

2.8.4 配置Log Cleaner

Log Cleaner默认启用，这回启动清理的线程池，如果要开始特定Topic的清理功能，可以开启特定的属性：

// 可以通过创建Topic时配置或者之后使用Topic命令实现
log.cleanup.policy=compact

Log Cleaner可以配置保留最小的不压缩的head log。可以通过配置压缩的延迟时间：

log.cleaner.min.compaction.lag.ms

这可以保证消息在配置的时长内不被压缩，如果没有设置，除了最后一个日志外，所有的日志都会被压缩。活动的segment是不会被压缩的，即使它保存的消息滞留时长超过了配置的最小压缩时长。

2.9 Quotas配额

Kafka集群可以对客户端请求进行配额，控制集群资源的是u用。Kafka Broker可以对客户端做两种资源类型的配额限制，同一个group的client共享配额：

定义字节流的阈值来限定网络带宽的配额（从0.9版本开始）；
request请求率的配额，网络和I/O线程cpu利用率的百分比（从0.11版本开始）。

2.9.1 为什么要对资源进行配额？

producers和consumers可能会产生或者消费大量的数据或者产生大量的请求，导致对broker资源的垄断，引起网络的饱和，对其他clients和brokers本身造成DOS攻击。资源的配额保护可以有效防止这些问题，在大型多租户集群中，因为一小部分表现不佳的客户端降低了良好的用户体验，这种情况下需要资源的配额保护。实际情况中，当把Kafka当作一种服务提供的时候，可以根据客户端和服务端的契约对API调用做限制。

2.9.2 Client groups

Kafka client是一个用户的概念，是在一个安全的集群中经过身份验证的用户。在一个支持非授权客户端的集群中，用户是一组非授权的users，broker使用一个可配置的PrincipalBuilder类来配置group规则。Client-id是客户端的逻辑分组，客户端应用使用一个有意义的名称进行标识。(user, client-id)元组定义了一个安全的客户端逻辑分组，使用相同的user和client-id标识。

资源配额可以针对(user, client-id)、users或者client-id groups三种规则进行配置。对于一个请求连接，连接会匹配最细化的配额规则的限制。同一个group的所有连接共享这个group的资源配额。举个例子，如果(user="test-user", client-id="test-client")客户端producer有10MB/sec的生产资源配置，这10MB/sec的资源在所有"test-user"用户，client-id是"test-client"的producer实例中是共享的。

2.9.3 Quota Configuration（资源配额的配置）

资源配额的配置可以根据(user, client-id)、user和client-id三种规则进行定义，在配额级别需要更高（或者更低）的配额的时候，是可以覆盖默认的配额配置。这种机制和每个topic可以自定义日志配置属性类似。覆盖User和(user, client-id)规则的配额配置会写道zookeeper的/config/users路径下，client-id配额的配置会写到/config/clients路径下。这些配置的覆盖会被所有的brokers实时的监听到并生效。所以这使得我们修改配额配置不需要重启整个集群，每个group的默认配额可以使用相同的机制进行动态更新。

配额配置的优先级顺序是：

/config/users//clients/
/config/users//clients/
/config/users/
/config/users//clients/
/config/users//clients/
/config/users/
/config/clients/
/config/clients/

Broker的配置属性（quota.producer.default、quota.consumer.default）也可以用来设置client-id groups默认的网络带宽配置。这些配置属性在未来的release版本会被deprecated。client-id的默认配额也是用zookeeper配置，和其他配额配置的覆盖和默认方式是相似的。

2.9.4 Network Bandwidth Quotas（网络带宽配额配置）

网络带宽配额使用字节速率阈值来定义每个group的客户端的共享配额。默认情况下，每个不同的客户端group时集群配置的固定配额，单位是bytes/sec。这个配额会议broker为基础进行定义。在clients被限制之前，每个group的clients可以发布和拉取单个broker的最大速率，单位是bytes/sec。

2.9.5 Requesr Rate Quotas 请求速率配额

请求速率的配额定义了一个客户端可以使用broker request handler I/O线程和网络线程在一个配额窗口时间内使用的百分比。n%的配置代表一个线程的n%的使用率，所以这种配额是建立在总容量((num.io.threads + num.network.threads) * 100)%之上的。每个group的client的资源在被限制之前可以使用单位配额时间窗口内I/O线程和网络线程利用率的n%。由于分配给I/O和网络线程的数量是基于broker的核数，所以请求量的配额代表每个group的client使用cpu的百分比。

2.9.6 Enforcement（限制）

默认情况下，集群给每个不同的客户端group配置固定的配额，这个配额是以broker为基础定义的。每个client在受到限制之前可以利用每个broker配置的配额资源。我们觉得给每个broker配置资源配额比为每个客户端配置一个固定的集群带宽资源要好，为每个客户端配置一个固定的集群带宽资源需要一个机制来共享client在brokers上的配额使用情况，这可能比配额本身实现更难。

broker在检测到有配额资源使用违反规则会怎么办？在我们计划中，broker不会返回error，而是会尝试减速client超出的配额设置。broker会计算出将客户端限制到配额之下的延迟时间，并且延迟response相应。这种方法对于客户端来说也是透明的（客户端指标除外），这也使得client不需要执行任何特殊的backoff和retry行为。而且不友好的客户端行为（没有backoff的重试）会加剧正在解决的资源配额问题。

网络字节速率和线程利用率可以用多个小窗口来衡量（例如1秒30个窗口），以便快速的检测和修正配额规则的违反行为。实际情况中，较大的测量窗口（例如，30秒10个窗口）会导致大量的突发流量，随后长时间的延迟，会使得用户体验不是很好。

3 实现思路

3.1 网络层

网络层相当于一个NIO服务，在此不再详细描述。sendfile（零拷贝）的实现是通过MessageSet接口的writeTo方法完成的，这样的胡照顾铁一般信仰file-backed集使用更高效的transferTo实现，而不再使用进程内的写缓存。线程模型是一个单独的接受线程和N个处理线程，每个线程处理固定数量的连接，这种设计方式再其他地方经过大量的测试，发现它是实现简单而且快速的，协议保持简单以允许未来实现其他语言的客户端。

3.2 消息

消息包含一个可变长度的header，一个可变长度不透明的字节数组key，一个可变长度不透明的字节数组value，消息中header的格式会在下一节描述。保持消息中的key和value不透明（二进制格式）是正确的决定：目前构建序列化库取得很大的进展，而且任何单一的序列化方式都不能满足所有的用途。毋庸置疑，使用Kafka的特定应用程序可能会要求特定的序列化类型作为自己使用的一部分。RecordBatch接口就是一种简单的消息迭代器，它可以使用特定的方法批量读写消息到NIO的channel中。

3.3 消息格式

消息通常按照批量的方式写入。record batch是批量消息的技术术语，它包含一条或者多条records。不良情况下，record batch只包含一条record。Record batches和records都有他们自己的headers。在Kafka 0.11.0及后续版本中（消息格式的版本为v2或者magix=2）解释了每种消息格式。

3.3.1 Record Batch

以下为Record Batch在硬盘上的格式。

baseOffset: int64
batchLength: int32
partitionLeaderEpoch: int32
magic: int8　(current magic value is 2)
crc: int32
attributes: int16
	bit 0~2:
		0: no compression
		1: gzip
		2: snappy
		3: lz4
    bit 3: timestampType
    bit 4: isTransactional (0 means not transactional)
    bit 5: isControlBatch (0 means not a control batch)
    bit 6~15: unused
lastOffsetDelta: int32
firstTimestamp: int64
maxTimestamp: int64
producerId: int64
producerEpoch: int16
baseSequence: int32
records:[Record]

请注意，启用压缩时，压缩的记录数据将直接按照记录数进行序列化。

CRC（一种数据校验码）会覆盖从属性到批处理结束的数据（即CRC后的所有字节数据）。CRC位于magic之后，这意味着，再决定如何解释批次的长度和magic类型之前，客户端需要解析magic类型。CRC计算不包括分区leaderepoch字段时为了避免broker收到两个批次的数据时，需要重新分配计算CRC。CRC-32C（Castagnoli）多项式用于计算。

压缩：不同于旧的消息格式，magic v2及以上版本在清理日志时保留原始日志中首次及最后一次offset/sequence。这是为了能够在日志重新加载时恢复生产者的状态。例如，如果我们不保留最后一次序列号，当分区leader失败以后，生产者会报OutOfSequence的错误，bicultural保留基础序列号来做重复检查（broker通过检查生产设该批次请求中第一次及最后一次序列号是否与上一次的序列号相匹配来判断是否重复）。因此，当批次中所有的记录被清理但批次数据依然保留是为了保存生者这最后一次的序列号，日志中可能有空的数据，不解的是在压缩中时间戳可能不会被保留，所以如果批次中的第一条记录被压缩，时间戳也会改变。

3.3.1.1 批次控制

批次控制包含成为控制记录的单条记录，控制记录不应该传送给应用程序，相反，他们是消费者用来过滤中断的事务消息。

控制记录的key符合以下模式：

version: int16 (current version is 0)
type: int16 (0 indicates an abort marker, 1 indicates a commit)

批次记录值得模式依赖于类型，对客户端来说它是透明的。

3.3.2 Record（记录）

Record level headers ware introduced in Kafka 0.11.0. The on-disk format of a record with Headers is delineated below.

length: varint
attributes: int8
	bit 0~7: unused
timestampDelta: varint
offsetDelta: varint
keyLength: varint
key: byte[]
vallueLen: varint
value: byte[]
Headers => [Header]

3.3.2.1 Record Header

headerKeyLength: varint
headerKey: String
headerValueLength: varint
Value: byte[]

3.4 日志

命名为my_topic的主题日志有两个分区，包含两个目录（命名为my_topic_0和my_topic_1），目录中分布着包含该topic消息的日志文件，日志文件的格式是“log entries”的序列，每个日志对象是由4位的数字N存储日志长度，后跟N字节的消息，每个消息使用64位的整数作为offset唯一标记，offset即为发送到该topic partition中所有流数据的起始位置。每个消息的磁盘格式如下，每个日志文件使用它包含的第一个日志的offset来命名，所以创建的第一个文件是0000000000,kafka，并且每个附件文件会有大概S字节前一个文件的整数名称，其中S是配置给出的最大文件大小。

记录的精确二进制格式是版本化的，并且按照标准接口进行维护，所以批量的记录可以在producer、broker和客户端之间传输，而不需要在使用时进行重新复制或转化。

消息的偏移量用作消息id是不常见的，我们最开始的想法是使用Produer自增的GUID，并维护从GUID到每个broker的offset的映射，这样的话每个消费者需要位每个服务端维护一个ID，提供全球唯一的GUID没有意义。而且，维护一个从随机ID到偏移量映射的复杂度需要一个重度的索引结构，它需要与磁盘进行同步，本质上需要一个完整的持久随机访问数据结构，因此为了简化查找结构，我们决定针对每个分区使用一个原子计数器，它可以利用分区id和节点id唯一标识一条消息。虽然这使得查找结构足够简单，但每个消费者的多个查询请求依然是相似的。一旦我们决定使用及数据，直接跳转到对应的偏移量显得更加自然，毕竟对于每个分区来说他们都是一个单调递增的整数，由于消费者API隐藏了偏移量，所以这个决定最终是一个实现细节，我们采用了更高效的方法。

3.4.1 Writes

日志允许序列化的追加到最后一个文件中，当文件大小达到配置的大小（默认1G）时，会生成一个新的文件，日志中有两个配置参数：M是在OS强制写文件到磁盘之前的消息条数，S是强制写盘的描述，这提供了一个在系统崩溃时最多丢失M条或者S秒消息的保证。

3.4.2 Reads

通过提供消息的64位逻辑偏移量和S位的max chunk size完成读请求，将会返回一个包含S位的消息缓存迭代器，S必须大于任何的单条的数据，但是在异常大消息的情况下，读取操作可以重试多次，每次会加倍缓冲的大小，直到消息被读取成功。可以指定最大消息和缓存大小使服务器拒绝接收超过其大小的消息，并为客户端设置消息的最大限度，它需要尝试读取多次获得完整的消息，读取缓冲区可能部分消息已结束，这很容易通过大小分界来检测。

按照偏移量读取的实际操作需要在数据存储目录中找到第一个日志分片的位置，在全局的偏移量中计算指定文件的偏移量，然后读取文件偏移量，搜索使用二分查找法查找在内存中保存的每个文件的偏移量来完成。

日志提供了将消息写入到当前的能力，以允许客户端从当前开始订阅，在消费者未能在其SLA指定的天数内消费其数据的情况下，这也是有用的，在这种情况下，客户端会尝试消费不存在的偏移量的数据，这会抛出OutOfRangeException异常，并且也会重置offset或者失败。

3.4.3 Deletes

在一个时间点下只有一个log segment的数据能被删除。日志管理器允许使用可插拔的删除策略来选择哪些文件符合删除条件，当前的删除策略会删除N天之前改动的日志，尽管保留最后的N GB数据可能有用。为了避免锁定读，同时允许删除修改Segment列表，我们使用copy-on-write形式的segment列表实现，在删除的同时它提供了一致的试图允许在多个segment列表视图上执行二进制的搜索。

3.4.4 Guarantees

日志提供了配置项M，它控制了在强制刷盘之前的最大消息数。启动时，日志恢复线程会运行，对最新的日志片段进行迭代，验证每条消息是否合法。如果消息对象的总数和偏移量小于文件的长度并且消息数据包的CRC32校验值与存储在消息中的CRC校验值相匹配的话，说明这个消息对象是合法的，如果检测到损坏，日志会在最后一个合法offset处截断。

请注意，有两种损坏必须处理：由于崩溃导致的未写入的数据块的丢失和将无意义已损坏的数据块添加到文件。原因是：通常系统不能保证文件索引节点和实际数据块之间的写入顺序，除此之外，如果在块数据被写入之前，文件索引已更新为新的大小，若此时系统崩溃，文件不到得到有意义的数据，则会导致数据丢失。

3.5 分布式

3.5.1 Consumer Offset Tracking （消费者offset跟踪）

高级别的consumer跟踪每个分区以消费的offset，并定期提交，以便在重启的情况下可以从这些offset中恢复。Kafka提供了一个选项在指定的broker中来存储所有给定的consumer组的offset，称为offset manager。例如，该consumer组的所有consumer实例向offset manager (broker) 发送提交和获取offset请求。高级别的consumer将会自动处理这些过程，如果使用低级别的consumer，将需要手动管理offset。目前在低级别的java consumer中不支持，只能在Zookeeper中提交或获取offset。如果使用简单的Scala consumer，将可拿到offset manager，并显式的提交或获取offset。对于包含offset manager的consumer可以通过发送GroupCoordinatorRequest到任意kafka broker，并接受GroupCoordinatorResponse相应，consumer可以继续向offset manager broker提交或获取offset。如果offset manager位置变动，consumer需要重新发现offset manager，如果向手动修改offset，可以参考OffsetCommitRequest和OffsetFetchRequest的源码是如何实现的。

当offset manager接收到一个OffsetCommitRequest，它将追加请求到一个特定的压缩名为_consumer_offsets的kafka topic中，当offset topic的所有副本接收offset之后，offset manager将发送一个提交offset成功的相应给consumer。万一offset无法在规定的时间内复制，offset将提交失败，consumer在回退之后可重试该提交（高级别consumer自动进行）。broker会定期压缩offset topic，因为只需要保存每个分区最近的offset。offset manager会缓存offset在内存表中，以便offset快速获取。

当offset manager接收一个offset的获取请求，将从offset缓存中返回最新的offset。如果offset manager刚启动或新的consumer组刚成为offset manager (成为offset topic分区的leader)，则需要加载offset topic的分区到缓存中，在这种情况下，offset将获取失败，并报出OffsetsLoadInProgress异常，consumer回滚后，重试OffsetFetchRequest（高级别consumer自动进行这些操作）。

3.5.2 从ZooKeeper迁移offset到kafka

Kafka consumers在早先的版本中offset默认存储在ZooKeeper中，可以通过下面的步骤迁移这些consumer到kafka：

在consumer配置中设置offsets.storage=kafka和dual.commit.enabled=true；
consumer做滚动消费，验证consumer是健康正常的；
在consumer配置中设置dual.commit.enabled=false；
consumer做滚动消费，验证consumer时健康正常的。

回滚（就是从kafka回到Zookeeper）也可以使用上面的步骤，通过设置offsets.storage=zookeeper。

3.5.3 ZooKeeper目录

下面给出了ZooKeeper的结构和算法，用于协调consumer和broker.

3.5.4 Notation

当一个path中的元素表示为[XYZ]，这意味着xyz的值是不固定的，实际上每个xyz的值可能是ZooKeeper的znode，例如/topic/[topic]是一个目录，/topic包含一个子目录（每个topic名称）。数字的范围如[0...5]来表示子目录0、1、2、3、4.箭头->用于标识znode的内容，例如/hello -> world标识znode /hello包含值world。

3.5.5 Broker节点注册

/brokers/ids/[0...N] --> {"jmx_port":...,"timestamp":...,"endpoints":[...],"host":...,"version":...,"port":...} (ephemeral node)

这是当前所有broker的节点列表，其中每个提供了一个唯一的逻辑broker的id标识它的consumer（必须作为配置的一部分）。在启动时，broker节点通过在/brokers/ids/下逻辑broker id创建一个znode来注册它自己。逻辑broker id的目的是当broker移动到不同的物理机器时，而不会影响消费者。尝试注册一个已存在的broker id时将返回错误（因为2个server配置了相同的broker id）。

由于broker在Zookeeper中用的是临时znode来注册，因此这个注册是动态的，如果broker关闭或宕机，节点将小时（通知consumer不再可用）。

3.5.6 Broker Topic注册

/brokers/topics/[topic]/partitions/[0...N]/state --> {"controller_epoch":...,"leader":...,"version":...,"leader_epoch":...,"isr":[...]} (ephemeral node)

每个broker在它自己的topic下注册、维护和存储该topic分区的数据。

3.5.7 Consumers and Consumer Groups

topic的consumer也在zookeeper中注册自己，以便互相协调和平衡数据的消耗。consumer也可以通过设置offsets.storage=zookeeper将他们的偏移量存储在zookeeper中。但是，这个偏移存储机制将在未来的版本中被弃用。因此，建议将数据迁移到kafka中。

多个consumer可组成一组，共同消费一个topic，在同一组中的每个consumer共享一个group_id。例如，如果一个consumer是foobar，在三台机器上运行，可能分配这个consumer的ID是"foobar"。这个组id是在consumer的配置文件中配置的。

每个分区正好被一个consumer组的consumer所消费，一组中的consumer尽可能公平地分配分区。

3.5.8 Consumer Id注册

除了由所有consumer共享的group_id，每个consumer都有一个临时且唯一的consumer_id（主机名的形式：uuid）用于识别。consumer的id在一下目录中注册。

/consumer/[group_id]/ids/[consumer_id] --> {"version":...,"subscription":{...:...},"pattern":...,"timestamp":...} (ephemeral node)

组中的每个consumer用consumer_id注册znode。znode的值包含一个map，这个id只是用来识别在组里目前活跃的consumer，这是个临时节点，如果consumer在处理中挂掉，它就会消失。

3.5.9 Consumer Offsets

Consumers track the maximum offset they have consumed in each partition. This value is stored in a ZooKeeper directory if offsets.storage=zookeeper。

/consumer/[group_id]/offsets/[topic]/[partition_id] --> offset_counter_value (persistent node)

3.5.10 Partition Owner registry

Each broker partition is consumed by a single consumer within a given consumer group. The consumer must establish its ownership of a given partition before any consumption can begin. To eatablish its ownership, a consumer writes ites own id in an ephemeral node under the particular broker partition it is claiming.

/consumer/[group_id]/owners/[topic]/[partition_id] --> consumer_node_id (ephemeral node)

3.5.11 Cluster Id

The cluster id is a unique and immutable identifier assigned to a Kafka cluster. The cluster id can have a maximum of 22 characters and the allowed characters are defined by the regular expression [a-zA-Z0-9_-]+, which corresponds to the characters used by the URL-safe Base64 variant with no padding. Conceptually, it is auto-generated when a cluster is started for the first time.

Implementation-wise, it is generated when a broker with version 0.10.1 or later is successfully started for the first time. The broker tries to get the cluster id from the /cluster/id znode during startup. If the znode does not exist, the broker generates a new cluster id an creates the znode with this cluster id.

3.5.12 Broker node registration

The broker nodes are basically independent, so they only publish information about what they have. When a broker joins, it registers itself under the broker node registry directory and writes information about its host name and port. The broker also register the list of exostomg topics and their logical partitions in the broker topic registry. New topics are registered dynamically when they are created on the broker.

3.5.13 Consumer registration algorithm

When a consumer starts, it does the following:

Register itself in the consumer id registry under its group.
Register a watch on changes (new consumers joining or any existing consumers leaving) under the consumer id registry. (Each change triggers rebalancing among all consumers within the group to which the changed consumer belongs.)
Register a watch on changes (new brokers joining or any existing brokers leaving) under the broker id registry. (Each change triggers rebalancing among all consumers in all consumer groups.)
If the consumer creates a message stream using a topic filter, it also registers a watch on changes (new topics being added) under the broker topic registry. (Each change will trigger re-evaluation of the avilable topics to determine which topics are allowed by the topic filter. A new allowed topic will trigger rebalancing among all consumers within the consumer group.)
Force itself to rebalance within in its consumer group.

3.5.14 Consumer rebalancing algorithm

The consumer rebalancing algorithms allows all the consumers in a group to come into consensus on which consumer is consuming which partitions. Consumer rebalancing is triggered on each addition or removal of both broker nodes and other consumers within the same group. For a given topic and given consumer group, broker partitions are divided evenly among consumers with the group. A partition is always consumed by a single consumer. This design simplifies the implementation. Had we allowed a partition to be concurrently consumed by multiple consumers, there would be contention on the partition and some kind of locking would be required. If there are more consumers than partitions, some consumers won't get any data at all. During rebalancing, we try to assign partitions to consumers in such a way that reduces the number of broker nodes each consumer has to connect to.

When rebalancing is triggered at one consumer, rebalancing should be triggered in other consumers within the same group about the same time.