1.入门

1.1简介

ApacheKafka®是一个分布式流媒体平台。这到底是什么意思？

我们认为流媒体平台具有三个关键功能：

1. 它可以让你发布和订阅记录流。在这方面，它类似于消​​息队列或企业消息传递系统。
2. 它允许您以容错方式存储记录流。
3. 它可以让您在发生记录时处理记录流。

什么是卡夫卡好？

它被用于两大类的应用程序：

1. 构建可在系统或应用程序之间可靠获取数据的实时流数据管道
2. 构建实时流应用程序，可以转换或响应数据流

要了解卡夫卡如何做这些事情，让我们深入探索卡夫卡的能力。

首先几个概念：

• Kafka作为一个或多个服务器上的集群运行。
• Kafka集群以称为主题的类别存储记录流。
• 每个记录由一个键，一个值和一个时间戳组成。

卡夫卡有四个核心API：

• 该制片API允许应用程序发布的记录流至一个或多个卡夫卡的话题。
• 该消费者API允许应用程序订阅一个或多个主题，并处理所产生的对他们记录的数据流。
• 所述流API允许应用程序充当流处理器，从一个或多个主题消耗的输入流，并产生一个输出流至一个或多个输出的主题，有效地变换所述输入流，以输出流。
• 该连接器API允许构建和运行卡夫卡主题连接到现有的应用程序或数据系统中重用生产者或消费者。例如，连接到关系数据库的连接器可能会捕获对表的每个更改。
• 

 

在Kafka中，客户端和服务器之间的通信是通过一个简单的，高性能的，与语言无关的TCP协议完成的。这个协议是版本化的，并保持与旧版本的向后兼容性。我们为Kafka提供了一个Java客户端，但客户端可以使用多种语言。

主题和日志

让我们首先深入核心抽象Kafka提供了一个记录流 - 主题。

主题是记录发布到的类别或饲料名称。卡夫卡的话题总是多用户的; 也就是说，一个主题可以有零个，一个或多个订阅写入数据的消费者。

对于每个主题，Kafka集群维护一个分区日志，如下所示：

 

每个分区是一个有序的，不可变的记录序列，不断追加到结构化的提交日志中。分区中的记录每个分配一个连续的id号，称为偏移量，用于唯一标识分区内的每条记录。

Kafka集群使用可配置的保留期限来保留所有已发布的记录（无论是否已被使用）。例如，如果保留策略设置为两天，则在记录发布后的两天内，保留策略可供使用，之后将被丢弃以腾出空间。卡夫卡的性能在数据大小方面是有效的，所以长时间存储数据不成问题。

 

实际上，以消费者为单位保留的唯一元数据是消费者在日志中的偏移或位置。这个偏移量是由消费者控制的：消费者通常会在读取记录时线性地推进其偏移量，但事实上，由于消费者的位置是由消费者控制的，所以它可以以任何喜欢的顺序消费记录。例如，消费者可以重置为较旧的偏移量以重新处理来自过去的数据，或者跳至最近的记录并从“now”开始消费。

这些功能的组合意味着卡夫卡的消费者非常便宜 - 他们可以来来去去，对集群或其他消费者没有太大的影响。例如，您可以使用我们的命令行工具来“尾巴”任何主题的内容，而不会改变现有的使用者所使用的内容。

日志中的分区有几个用途。首先，它们允许日志的大小超出适合单个服务器的大小。每个单独的分区必须适合托管它的服务器，但是一个主题可能有许多分区，因此它可以处理任意数量的数据。其次，它们作为并行的单位 - 更多的是在一点上。

分配

日志的分区分布在Kafka集群中的服务器上，每个服务器处理数据并请求共享分区。每个分区都通过可配置数量的服务器进行复制，以实现容错。

每个分区有一个服务器充当“领导者”，零个或多个服务器充当“追随者”。领导处理分区的所有读取和写入请求，而追随者被动地复制领导。如果领导失败，其中一个追随者将自动成为新领导。每个服务器充当其中一些分区的领导者和其他人的追随者，因此负载在集群内平衡良好。

生产者

生产者发布数据到他们选择的主题。生产者负责选择哪个记录分配给主题内的哪个分区。这可以以循环的方式完成，只是为了平衡负载，或者可以根据某些语义分区功能（例如基于记录中的某个键）来完成。更多关于使用分区在第二！

消费者

消费者用消费者组名称标记自己，并且发布到主题的每个记录被传递到每个订阅消费者组中的一个消费者实例。消费者实例可以在不同的进程中或在不同的机器上。

如果所有消费者实例具有相同的消费者组，则记录将有效地在消费者实例上负载平衡。

如果所有消费者实例具有不同的消费者组，则每个记录将被广播给所有消费者进程。

 

两个服务器Kafka集群托管四个分区（P0-P3）与两个消费者组。消费者组A有两个消费者实例，而组B有四个消费者实例。

然而，更普遍的是，我们发现话题中有少量消费群体，每个“逻辑用户”都有一个消费群体。每个组由许多消费者实例组成，具有可扩展性和容错性。这不过是发布 - 订阅语义，订阅者是一群消费者而不是一个进程。

在Kafka中实现消费的方式是将日志中的分区划分为消费者实例，以便每个实例在任何时间点都是“公平分享”分区的唯一消费者。这个维护组中成员资格的过程是由Kafka协议动态地处理的。如果新实例加入组，他们将接管来自组中其他成员的一些分区; 如果一个实例死亡，其分区将分配给其余的实例。

卡夫卡只提供一个分区内的记录总数，而不是主题中的不同分区之间。每个分区排序与按键分区数据的能力相结合，足以满足大多数应用程序的需求。但是，如果您需要全部订单而不是记录，则可以通过仅具有一个分区的主题来实现，但这意味着每个消费者组只有一个消费者进程。

担保

在一个高层次的卡夫卡提供以下保证：

• 由制作者发送到特定主题分区的消息将按照它们发送的顺序附加。也就是说，如果记录M1由同一个生产者作为记录M2被发送，并且M1被首先发送，则M1将具有比M2更低的偏移并且出现在日志中较早的地方。
• 消费者实例按照存储在日志中的顺序查看记录。
• 对于具有复制因子N的主题，我们将容忍多达N-1个服务器故障，而不会丢失任何提交给日志的记录。

有关这些保证的更多细节在文档的设计部分给出。

卡夫卡作为消息系统

卡夫卡的流概念如何与传统的企业消息传递系统相比较？

消息传统上有两种模式：排队和发布 - 订阅。在队列中，消费者池可以从服务器读取并且每个记录都转到其中的一个; 在发布 - 订阅记录被广播给所有消费者。这两种模式都有其优点和缺点。排队的优势在于，它允许您将数据处理划分为多个消费者实例，这样可以扩展您的处理。不幸的是，队列不是多用户的，一旦一个进程读取了数据，发布 - 订阅允许您将数据广播到多个进程，但无法进行扩展处理，因为每条消息都发送给每个订阅者。

卡夫卡的消费群体概念概括了这两个概念。与队列一样，消费者组允许您将一系列流程（消费者组的成员）的处理分开。与发布 - 订阅一样，Kafka允许您向多个消费者群体广播消息。

Kafka模型的优点是每个主题都具有这些属性 - 它可以扩展处理，也可以是多用户 - 不需要选择其中一个。

Kafka也比传统的消息系统有更强的订单保证。

传统队列在服务器上按顺序保留记录，并且如果多个使用者从队列中消耗，则服务器按照存储的顺序提交记录。然而，虽然服务器按顺序提交记录，但是记录是异步传送给消费者的，所以它们可能会在不同的消费者中出现。这实际上意味着记录的排序在并行消耗的情况下丢失。消息传递系统通常具有“排他消费者”的概念，只允许一个进程从队列中消费，但这当然意味着在处理过程中没有并行性。

卡夫卡做得更好。通过在主题内部具有并行的概念 - 分区，Kafka能够提供订单保证和对消费者流程池的负载平衡。这是通过将主题中的分区分配给使用者组中的使用者来实现的，以便每个分区仅由组中的一个使用者使用。通过这样做，我们确保消费者是该分区的唯一读者，并按顺序使用这些数据。由于有很多分区，这仍然可以平衡许多消费者实例的负载。但请注意，消费群组中的消费者实例不能多于分区。

卡夫卡作为存储系统

任何允许将消息发布出去的消息队列都可以充当存储系统。Kafka的不同之处在于它是一个非常好的存储系统。

写入Kafka的数据写入磁盘并进行复制以实现容错。Kafka允许生产者等待确认，以便在完全复制之前写入不被认为是完整的，并且即使写入的服务器失败也能保证持续。

Kafka的磁盘结构使用了很好的规模 - 无论您在服务器上有50 KB还是50 TB的持久性数据，Kafka都会执行相同的操作。

由于认真考虑存储并允许客户端控制其读取位置，所以可以将Kafka视为专用于高性能，低延迟提交日志存储，复制和传播的专用分布式文件系统。

有关Kafka提交日志存储和复制设计的详细信息，请阅读此页。

卡夫卡流处理

只读取，写入和存储数据流是不够的，目的是启用流的实时处理。

在Kafka中，流处理器是指从输入主题获取连续数据流，对该输入执行一些处理，并产生连续数据流以输出主题的任何东西。

例如，零售应用程序可能会接受输入的销售和货运流，并输出一系列重新排序和对这些数据进行计算的价格调整。

直接使用生产者和消费者API可以做简单的处理。但是对于更复杂的转换，Kafka提供了一个完全集成的Streams API。这允许构建应用程序进行非平凡的处理，从而计算聚合关闭流或将流连接在一起。

这个工具有助于解决这类应用程序面临的难题：处理乱序数据，重新处理代码更改的输入，执行有状态的计算等等。

流API基于Kafka提供的核心原语构建：它使用生产者和消费者API进行输入，使用Kafka进行有状态存储，并在流处理器实例之间使用相同的组机制来实现容错。

放在一起

消息传递，存储和流处理的这种组合看起来很不寻常，但对于Kafka作为一个流媒体平台来说，这是非常重要的。

像HDFS这样的分布式文件系统允许存储用于批处理的静态文件。有效地，这样的系统允许存储和处理过去的历史数据。

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

Kafka结合了这两种功能，而且这两种组合对于Kafka用作流媒体应用平台和流式数据流水线都是至关重要的。

通过将存储和低延迟订阅相结合，流式应用程序可以同样的方式处理过去和未来的数据。这是一个单一的应用程序可以处理历史，存储的数据，而不是结束，当它达到最后一个记录，它可以继续处理未来的数据到达。这是流处理的概括概念，包括批处理以及消息驱动的应用程序。

同样，对于流式传输数据流水线，订阅实时事件的组合使得可以将Kafka用于非常低延迟的流水线; 但是可靠地存储数据的能力可以将其用于必须保证数据交付的关键数据，或者与只能定期加载数据的离线系统集成，或者可能长时间停机进行维护。流处理设施可以在数据到达时进行转换。

有关Kafka提供的担保，API和功能的更多信息，请参阅文档的其余部分。

1.2用例

下面是一些ApacheKafka®流行用例的描述。有关这些领域的概述，请参阅此博客文章。

消息

卡夫卡可以很好地替代传统的信息经纪人。消息代理被用于各种原因（将数据处理与数据生成器分离，缓冲未处理的消息等）。与大多数消息传递系统相比，Kafka具有更好的吞吐量，内置的分区，复制和容错功能，使其成为大型消息处理应用程序的理想解决方案。

根据我们的经验，消息传递的使用往往是相对较低的吞吐量，但可能需要较低的端到端延迟，并且通常取决于Kafka提供的强大的持久性保证。

在这个领域，Kafka与传统的消息系统（如ActiveMQ或 RabbitMQ）相当。

网站活动跟踪

Kafka的原始用例是能够将用户活动跟踪管道重建为一组实时发布 - 订阅订阅源。这意味着网站活动（用户可能采用的网页浏览量，搜索或其他操作）将发布到每个活动类型一个主题的中心主题。这些订阅源可用于订阅各种用例，包括实时处理，实时监控，加载到Hadoop或离线数据仓库系统以进行离线处理和报告。

活动跟踪通常是非常高的量，因为为每个用户页面视图生成许多活动消息。

度量

卡夫卡通常用于运行监控数据。这涉及从分布式应用程序汇总统计数据以生成操作数据的集中式源。

日志聚合

许多人使用Kafka作为日志聚合解决方案的替代品。日志聚合通常从服务器收集物理日志文件，并将其置于*位置（可能是文件服务器或HDFS）进行处理。Kafka提取文件的细节，并将日志或事件数据作为消息流进行更清晰的抽象。这样可以实现更低延迟的处理，并且更容易支持多个数据源和分布式数据消耗。与Scribe或Flume等以日志为中心的系统相比，Kafka提供同样出色的性能，由复制产生的更强大的持久性保证以及更低的端到端延迟。

流处理

Kafka的许多用户在处理管道中处理数据，这些数据由多个阶段组成，其中原始输入数据从Kafka主题中消耗，然后聚合，丰富或以其他方式转化为新的主题，以供进一步消费或后续处理。例如，用于推荐新闻文章的处理流水线可以从RSS提要抓取文章内容并将其发布到“文章”主题; 进一步的处理可以对这个内容进行标准化或者重复删除，并且将已清理的文章内容发布到新的主题; 信读亦读信息范范信息亦读信息范范信息中信息 这种处理流水线基于各个主题创建实时数据流的图表。从0.10.0.0开始，这是一个轻量但功能强大的流处理库，称为Kafka Streams 在Apache Kafka中可用于执行如上所述的数据处理。除了Kafka Streams之外，替代性的开源流处理工具还包括Apache Storm和 Apache Samza。

事件采购

事件源是应用程序设计的一种风格，其中状态更改以时间排序的记录序列进行记录。Kafka对非常大的存储日志数据的支持使得它成为以这种风格构建的应用程序的优秀后端。

提交日志

Kafka可以作为分布式系统的一种外部提交日志。日志有助于复制节点之间的数据，并作为失败节点恢复数据的重新同步机制。Kafka中的日志压缩功能有助于支持这种用法。在这个用法中，Kafka与Apache BookKeeper项目类似。

1.3快速入门

本教程假定您开始新鲜，没有现有的Kafka或ZooKeeper数据。由于基于Unix和Windows平台的Kafka控制台脚本不同，因此在Windows平台上使用bin\windows\而不是bin/，并将脚本扩展名更改为.bat。

第1步：下载代码

下载 1.0.0版本并解压缩。

1 2	> tar -xzf kafka_2.11-1.0.0.tgz > cd kafka_2.11-1.0.0

第2步：启动服务器

Kafka使用ZooKeeper，因此如果您还没有ZooKeeper服务器，则需要先启动ZooKeeper服务器。您可以使用与kafka一起打包的便捷脚本来获取快速而简单的单节点ZooKeeper实例。

1 2 3

> bin/zookeeper-server-start.sh config/zookeeper.properties [2013-04-22 15:01:37,495] INFO Reading configuration from: config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig) ...

现在启动Kafka服务器：

1 2 3 4

> bin/kafka-server-start.sh config/server.properties [2013-04-22 15:01:47,028] INFO Verifying properties (kafka.utils.VerifiableProperties) [2013-04-22 15:01:47,051] INFO Property socket.send.buffer.bytes is overridden to 1048576 (kafka.utils.VerifiableProperties) ...

第3步：创建一个主题

我们用一个分区和一个副本创建一个名为“test”的主题：

1	> bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

我们现在可以看到这个话题，如果我们运行列表主题命令：

1 2	> bin/kafka-topics.sh --list --zookeeper localhost:2181 test

或者，您也可以将代理配置为在发布不存在的主题时自动创建主题，而不是手动创建主题。

第4步：发送一些消息

Kafka带有一个命令行客户端，它将从文件或标准输入中获取输入，并将其作为消息发送到Kafka集群。默认情况下，每行将作为单独的消息发送。

运行生产者，然后在控制台输入一些消息发送到服务器。

1 2 3

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test This is a message This is another message

第5步：启动一个用户

卡夫卡也有一个命令行消费者，将消息转储到标准输出。

1 2 3

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning This is a message This is another message

如果您将上述每个命令都运行在不同的终端中，则现在应该可以将消息键入生产者终端，并将其显示在消费者终端中。

所有的命令行工具都有其他选项。不带任何参数运行该命令将显示使用信息更详细地记录它们。

第6步：设置多代理群集

到目前为止，我们一直在反对一个经纪人，但这并不好玩。对于卡夫卡来说，一个经纪人只是一个规模一个的集群，所以除了开始一些经纪人实例之外没有太大的变化。但是为了得到它的感觉，让我们把我们的集群扩展到三个节点（仍然都在我们的本地机器上）。

首先，我们为每个代理创建一个配置文件（在Windows上使用copy命令代替）：

1 2	> cp config/server.properties config/server-1.properties > cp config/server.properties config/server-2.properties

现在编辑这些新文件并设置以下属性：

1 2 3 4 五 6 7 8 9

config/server-1.properties:     broker.id=1     listeners=PLAINTEXT://:9093     log.dir=/tmp/kafka-logs-1   config/server-2.properties:     broker.id=2     listeners=PLAINTEXT://:9094     log.dir=/tmp/kafka-logs-2

该broker.id属性是群集中每个节点的唯一且永久的名称。我们必须重写端口和日志目录，因为我们在同一台机器上运行这些端口和日志目录，我们希望让所有代理都试图在同一个端口注册或覆盖彼此的数据。

我们已经有Zookeeper和我们的单节点了，所以我们只需要启动两个新的节点：

1 2 3 4

> bin/kafka-server-start.sh config/server-1.properties & ... > bin/kafka-server-start.sh config/server-2.properties & ...

现在创建一个复制因子为三的新主题：

1	> bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic

1 2 3

> bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic Topic:my-replicated-topic   PartitionCount:1    ReplicationFactor:3 Configs:     Topic: my-replicated-topic  Partition: 0    Leader: 1   Replicas: 1,2,0 Isr: 1,2,0

好，但现在我们有一个集群，我们怎么知道哪个经纪人在做什么？要查看运行“描述主题”命令：

这里是对输出的解释。第一行给出了所有分区的摘要，每个附加行给出了关于一个分区的信息。由于我们只有一个分区，所以只有一行。

• “leader”是负责给定分区的所有读取和写入的节点。每个节点将成为分区随机选择部分的领导者。
• “副本”是复制此分区的日志的节点列表，无论它们是领导者还是即使他们当前还活着。
• “isr”是一组“同步”副本。这是复制品列表的子集，当前活着并被引导到领导者。

请注意，在我的示例中，节点1是该主题的唯一分区的领导者。

我们可以在我们创建的原始主题上运行相同的命令来查看它的位置：

1 2 3

> bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test Topic:test  PartitionCount:1    ReplicationFactor:1 Configs:     Topic: test Partition: 0    Leader: 0   Replicas: 0 Isr: 0

所以这里并不奇怪，原来的主题没有副本，而且在服务器0上，这是我们创建集群时唯一的服务器。

让我们发表一些信息给我们的新主题：

1 2 3 4 五

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic ... my test message 1 my test message 2 ^C

现在让我们消费这些消息：

1 2 3 4 五

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic ... my test message 1 my test message 2 ^C

现在我们来测试一下容错。经纪人1是作为领导者，所以让我们杀了它：

1 2 3

> ps aux | grep server-1.properties 7564 ttys002    0:15.91 /System/Library/Frameworks/JavaVM.framework/Versions/1.8/Home/bin/java... > kill -9 7564

在Windows上使用：

1 2 3 4

> wmic process where "caption = 'java.exe' and commandline like '%server-1.properties%'" get processid ProcessId 6016 > taskkill /pid 6016 /f

领导已经切换到其中一个从属节点，并且节点1不再处于同步副本集合中：

1 2 3

> bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic Topic:my-replicated-topic   PartitionCount:1    ReplicationFactor:3 Configs:     Topic: my-replicated-topic  Partition: 0    Leader: 2   Replicas: 1,2,0 Isr: 2,0

但是，即使原先写入的领导者失败，这些消息仍然可用于消费：

1 2 3 4 五

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic ... my test message 1 my test message 2 ^C

第7步：使用Kafka Connect来导入/导出数据

从控制台写入数据并将其写回控制台是一个方便的起点，但您可能想要使用其他来源的数据或将数据从Kafka导出到其他系统。对于许多系统，您可以使用Kafka Connect来导入或导出数据，而不必编写自定义集成代码。

Kafka Connect是Kafka包含的一个工具，可以将数据导入和导出到Kafka。它是一个可扩展的工具，运行 连接器，实现与外部系统交互的自定义​​逻辑。在这个快速入门中，我们将看到如何使用简单的连接器运行Kafka Connect，这些连接器将数据从文件导入到Kafka主题，并将数据从Kafka主题导出到文件。

首先，我们将通过创建一些种子数据开始测试：

1	> echo -e "foo\nbar" > test.txt

或在Windows上：

1 2	> echo foo> test.txt > echo bar>> test.txt

接下来，我们将启动两个以独立模式运行的连接器，这意味着它们将在单个本地专用进程中运行。我们提供三个配置文件作为参数。首先是Kafka Connect过程的配置，包含常见的配置，例如要连接的Kafka代理以及数据的序列化格式。其余的配置文件都指定一个要创建的连接器。这些文件包括唯一的连接器名称，要实例化的连接器类以及连接器所需的任何其他配置。

1	> bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties

Kafka附带的这些示例配置文件使用您之前启动的默认本地群集配置，并创建两个连接器：第一个是源连接器，用于读取输入文件中的行，并将每个连接生成为Kafka主题，第二个为连接器它从Kafka主题读取消息，并在输出文件中产生每行消息。

在启动过程中，您会看到一些日志消息，其中一些指示连接器正在实例化。Kafka Connect进程启动后，源连接器应该开始读取test.txt主题connect-test，并将其生成主题，并且接收器连接器应该开始读取主题中的消息connect-test 并将其写入文件test.sink.txt。我们可以通过检查输出文件的内容来验证通过整个管道传输的数据：

1 2 3

> more test.sink.txt foo bar

请注意，数据存储在Kafka主题中connect-test，因此我们也可以运行控制台使用者来查看主题中的数据（或使用自定义使用者代码来处理它）：

1 2 3 4

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-beginning {"schema":{"type":"string","optional":false},"payload":"foo"} {"schema":{"type":"string","optional":false},"payload":"bar"} ...

连接器继续处理数据，所以我们可以将数据添加到文件，并看到它在管道中移动：

1	> echo Another line>> test.txt

您应该看到该行出现在控制台使用者输出和接收器文件中。

第8步：使用Kafka流来处理数据

Kafka Streams是用于构建关键任务实时应用程序和微服务的客户端库，输入和/或输出数据存储在Kafka集群中。Kafka Streams结合了在客户端编写和部署标准Java和Scala应用程序的简单性以及Kafka服务器端集群技术的优势，使这些应用程序具有高度可伸缩性，弹性，容错性，分布式等特性。本快速入门示例将演示如何运行在此库中编码的流式应用程序。

1.4生态系统

在主发行版之外还有大量与Kafka集成的工具。该生态系统页面中列出的许多信息，包括流处理系统，Hadoop的集成，监控和部署工具。

 

 

1.5从以前的版本升级

从0.8.x，0.9.x，0.10.0.x，0.10.1.x，0.10.2.x或0.11.0.x升级到1.0.0

卡夫卡1.0.0引入了有线协议的变化。通过遵循以下建议的滚动升级计划，您可以保证在升级过程中不会出现停机。但是，请在升级之前查看1.0.0中的显着更改。

对于滚动升级：

1. 更新所有代理上的server.properties并添加以下属性。CURRENT_KAFKA_VERSION是指您要升级的版本。CURRENT_MESSAGE_FORMAT_VERSION指的是当前正在使用的消息格式版本。如果您以前没有重写消息格式，那么应该将CURRENT_MESSAGE_FORMAT_VERSION设置为匹配CURRENT_KAFKA_VERSION。
   • inter.broker.protocol.version = CURRENT_KAFKA_VERSION（例如0.8.2,0.9.0,0.10.0,0.10.1,0.10.2,0.11.0）。
2. 一次升级一个代理：关闭代理，更新代码并重新启动代理。
3. 整个群集升级后，通过编辑修改协议版本inter.broker.protocol.version并将其设置为1.0。
4. 重新启动代理，以使新的协议版本生效。

其他升级说明：

1. 如果你愿意接受停机时间，你可以简单地把所有的经纪人关闭，更新代码并重新开始。他们将默认启动新的协议。
2. 颠覆协议版本并重新启动可以在代理升级后的任何时候完成。它不一定要在之后立即。同样的消息格式版本。

1.0.0中的显着变化

• 主题删除现在默认启用，因为功能现在是稳定的。希望保留以前行为的用户应将代理配置设置delete.topic.enable为false。请记住，主题删除删除数据，操作是不可逆的（即没有“取消删除”操作）
• 对于支持时间戳搜索的主题，如果没有为分区找到偏移量，则该分区现在包含在搜索结果中，并具有空偏移值。以前，分区不包含在地图中。此更改是为了使搜索行为与不支持时间戳搜索的主题的情况一致。
• 如果inter.broker.protocol.version是1.0或更高版本，即使存在脱机日志目录，代理现在也将保持联机，以便在实时日志目录上提供副本。由于硬件故障导致的IOException，日志目录可能会变为脱机状态。用户需要监视每个代理度量标准offlineLogDirectoryCount来检查是否存在脱机日志目录。
• 增加了KafkaStorageException，这是一个可以回溯的异常。如果客户端的FetchRequest或ProducerRequest的版本不支持KafkaStorageException，则KafkaStorageException将在响应中转换为NotLeaderForPartitionException。
• -XX：在默认的JVM设置中+ DisableExplicitGC被-XX：+ ExplicitGCInvokesConcurrent替换。这有助于避免在某些情况下通过直接缓冲区分配本机内存期间的内存不足异常。
• 重写的handleError方法实现已经从以下过时类中除去kafka.api：包FetchRequest，GroupCoordinatorRequest，OffsetCommitRequest，OffsetFetchRequest，OffsetRequest，ProducerRequest，和TopicMetadataRequest。这只是为了在代理上使用，但它不再被使用，实现也没有被维护。为了二进制兼容性保留了一个存根实现。
• Java客户端和工具现在接受任何字符串作为客户端ID。
• 弃用的工具kafka-consumer-offset-checker.sh已被删除。使用kafka-consumer-groups.sh得到消费群的详细信息。
• SimpleAclAuthorizer现在默认将访问拒绝记录到授权人日志中。
• 现在，身份验证失败报告给客户端，作为其中的一个子类AuthenticationException。如果客户端连接验证失败，则不会执行重试。
• 自定义SaslServer实现可能会抛出SaslAuthenticationException提供错误消息返回给客户端，指出身份验证失败的原因。执行者应注意不要在异常消息中包含任何不应泄露给未经身份验证的客户端的安全关键信息。
• app-info向JMX注册以提供版本和提交ID 的mbean将被弃用，并由提供这些属性的度量替换。
• 卡夫卡指标现在可能包含非数字值。org.apache.kafka.common.Metric#value()已被弃用，并将0.0在这种情况下返回，以最大限度地减少读取每个客户端度量值（通过MetricsReporter实现或调用metrics()方法）的用户的概率。 org.apache.kafka.common.Metric#metricValue()可以用来检索数字和非数字度量值。
• 现在，每个Kafka速率指标都有相应的累计计数度量标准，并带有后缀-total 以简化下游处理。例如，records-consumed-rate有一个相应的度量标准records-consumed-total。
• 只有系统属性kafka_mx4jenable设置为Mx4j才能启用true。由于逻辑反转错误，它以前是默认启用的，如果kafka_mx4jenable被设置为禁用则禁用true。
• org.apache.kafka.common.security.auth客户端jar 包中的包已经公开并添加到javadoc中。以前在这个软件包中的内部类已经移到其他地方了。
• 在使用授权人并且用户对主题没有必需的权限时，代理将无论主体是否存在，都会将TOPIC_AUTHORIZATION_FAILED错误返回给请求。如果用户具有所需权限并且该主题不存在，则将返回UNKNOWN_TOPIC_OR_PARTITION错误代码。
• config / consumer.properties文件更新为使用新的使用者配置属性。

新协议版本

• KIP-112：LeaderAndIsrRequest v1引入了一个分区级别的is_new字段。
• KIP-112：UpdateMetadataRequest v4引入了分区级offline_replicas字段。
• KIP-112：MetadataResponse v5引入了分区级offline_replicas字段。
• KIP-112：ProduceResponse v4引入了KafkaStorageException的错误代码。
• KIP-112：FetchResponse v6引入了KafkaStorageException的错误代码。
• KIP-152：已添加SaslAuthenticate请求以启用身份验证失败的报告。如果SaslHandshake请求版本大于0，将使用此请求。

升级1.0.0 Kafka Streams应用程序

• 将Streams应用程序从0.11.0升级到1.0.0不需要代理升级。Kafka Streams 1.0.0应用程序可以连接到0.11.0，0.10.2和0.10.1代理（尽管如此，不能连接到0.10.0代理）。
• 如果您正在监控流指标，则需要对报告和监控代码中的指标名称进行一些更改，因为指标传感器层次结构已更改。
• 有一些公共的API，包括ProcessorContext#schedule()，Processor#punctuate()和KStreamBuilder，TopologyBuilder正在被新的API弃用。我们建议进行相应的代码更改，因为在升级时新的API看起来非常相似，所以这应该是非常小的。
• 有关更多详细信息，请参阅1.0.0中的Streams API更改。

从0.8.x，0.9.x，0.10.0.x，0.10.1.x或0.10.2.x升级到0.11.0.0

卡夫卡0.11.0.0引入了一个新的消息格式版本以及有线协议的变化。通过遵循以下建议的滚动升级计划，您可以保证在升级过程中不会出现停机。不过，请在升级之前查看0.11.0.0中的显着更改。

从版本0.10.2开始，Java客户端（生产者和消费者）已经获得了与旧代理进行通信的能力。版本0.11.0客户可以与版本0.10.0或更新的代理进行通信。但是，如果您的经纪人年龄大于0.10.0，则必须先升级Kafka集群中的所有经纪人，然后再升级您的客户。版本0.11.0经纪人支持0.8.x和更新的客户端。

对于滚动升级：

1. 更新所有代理上的server.properties并添加以下属性。CURRENT_KAFKA_VERSION是指您要升级的版本。CURRENT_MESSAGE_FORMAT_VERSION指的是当前正在使用的消息格式版本。如果您以前没有重写消息格式，那么应该将CURRENT_MESSAGE_FORMAT_VERSION设置为匹配CURRENT_KAFKA_VERSION。
   • inter.broker.protocol.version = CURRENT_KAFKA_VERSION（例如0.8.2,0.9.0,0.10.0,0.10.1或0.10.2）。
   • log.message.format.version = CURRENT_MESSAGE_FORMAT_VERSION（请参阅升级后的潜在性能影响，了解有关此配置的详细信息。）
2. 一次升级一个代理：关闭代理，更新代码并重新启动代理。
3. 整个群集升级后，通过编辑修改协议版本inter.broker.protocol.version并将其设置为0.11.0，但不要更改log.message.format.version。
4. 重新启动代理，以使新的协议版本生效。
5. 一旦所有（或大部分）使用者升级到0.11.0或更高版本，则将每个代理上的log.message.format.version更改为0.11.0，然后逐个重新启动它们。请注意，较早的Scala消费者不支持新的消息格式，所以为了避免下转换的性能成本（或者只利用一次语义），必须使用新的Java消费者。

其他升级说明：

1. 如果你愿意接受停机时间，你可以简单地把所有的经纪人关闭，更新代码并重新开始。他们将默认启动新的协议。
2. 颠覆协议版本并重新启动可以在代理升级后的任何时候完成。它不一定要在之后立即。同样的消息格式版本。
3. bin/kafka-topics.sh在更新全局设置之前，还可以使用主题管理工具（）在个别主题上启用0.11.0消息格式log.message.format.version。
4. 如果要从0.10.0之前的版本升级，则在切换到0.11.0之前，不必先将消息格式更新为0.10.0。

升级0.10.2 Kafka Streams应用程序

• 将Streams应用程序从0.10.2升级到0.11.0不需要代理升级。Kafka Streams 0.11.0应用程序可以连接到0.11.0，0.10.2和0.10.1代理（尽管如此，不能连接到0.10.0代理）。
• 如果您指定自定义key.serde，value.serde并且timestamp.extractor在配置中，建议使用替换的配置参数，因为这些配置已被弃用。
• 有关更多详细信息，请参阅0.11.0中的Streams API更改。

在0.11.0.0中有显着的变化

• 不洁净的*选举现在默认禁用。新的默认值有利于耐用性而不是可用性。希望保留以前行为的用户应将代理配置设置unclean.leader.election.enable为true。
• 生产者配置block.on.buffer.full，metadata.fetch.timeout.ms并timeout.ms已被删除。他们最初在Kafka 0.9.0.0中被弃用。
• 该offsets.topic.replication.factor券商的配置现在在强制汽车主题产生。内部自动主题创建将失败并带有GROUP_COORDINATOR_NOT_AVAILABLE错误，直到群集大小满足此复制因子要求。
• 当用快速压缩数据时，制造商和代理商将使用压缩方案的默认块大小（2 x 32 KB）而不是1 KB来提高压缩率。信息信息范范范范挥辛区中信息亦范范信亦信息信息信息信息信息 信息内暂名说预全
• 同样，使用gzip压缩数据时，生产者和代理将使用8 KB而不是1 KB作为缓冲区大小。gzip的默认值过低（512字节）。
• 代理配置max.message.bytes现在适用于一批消息的总大小。之前的设置应用于批量压缩消息，或单独应用于非压缩消息。批量消息可能只包含单个消息，因此在大多数情况下，单个消息的大小限制只能通过批量格式的开销来减少。但是，消息格式转换有一些细微的含义（详见下文）。还要注意的是，虽然代理以前会确保每个提取请求中返回至少一条消息（不管总提取大小和分区级别的提取大小如何），但同样的行为现在适用于一个消息批处理。
• GC日志旋转默认启用，详情请参阅KAFKA-3754。
• RecordMetadata，MetricName和Cluster类的弃用构造函数已被删除。
• 通过提供用户头读写访问的新Headers接口增加了用户头支持。
• ProducerRecord和ConsumerRecord通过Headers headers()方法调用公开新的Headers API 。
• ExtendedSerializer和ExtendedDeserializer接口被引入来支持头文件的序列化和反序列化。如果配置的串行器和解串器不是上述类，那么标题将被忽略。
• group.initial.rebalance.delay.ms引入了一个新的配置。该配置指定以毫秒为单位的时间GroupCoordinator将延迟最初的消费者重新平衡。group.initial.rebalance.delay.ms新会员加入团队的价值将进一步推迟重新平衡，最高可达max.poll.interval.ms。这个的默认值是3秒。在开发和测试过程中，为了不延迟测试执行时间，可能需要将其设置为0。
• org.apache.kafka.common.Cluster#partitionsForTopic，partitionsForNode和availablePartitionsForTopic方法会返回一个空列表，而不是null在的情况下（这被认为是不好的做法）的元数据所要求的主题不存在。
• 流API的配置参数timestamp.extractor，key.serde以及value.serde被弃用，替换default.timestamp.extractor，default.key.serde和default.value.serde分别。
• 对于Java消费者commitAsyncAPI 中的偏移提交失败，当实例RetriableCommitFailedException传递给提交回调时，我们不再公开底层原因。有关 更多详细信息，请参阅 KAFKA-5052。

新协议版本

• KIP-107：FetchRequest v5引入了一个分区级别的log_start_offset字段。
• KIP-107：FetchResponse v5引入了一个分区级别的log_start_offset字段。
• KIP-82：ProduceRequest v3 header在消息协议中引入了一个包含key字段和value字段的数组。
• KIP-82：FetchResponse v5 header在消息协议中引入了一个包含key字段和value字段的数组。

关于一次语义的注记

卡夫卡0.11.0支持生产者的幂等和事务性能力。幂等式传递确保消息在单个生产者的生命周期内仅传递给特定的主题分区一次。事务交付允许生产者发送数据到多个分区，使得所有的消息都被成功地传递，或者它们都不是。在一起，这些功能使卡夫卡“恰好一次语义”。有关这些功能的更多详细信息，请参阅用户指南，但下面我们添加一些关于在升级群集中启用它们的特定注意事项。请注意，启用EoS不是必需的，如果未使用，则不会影响经纪人的行为。

1. 只有新的Java生产者和消费者支持一次语义。
2. 这些功能主要取决于0.11.0消息格式。试图以较旧的格式使用它们将导致不受支持的版本错误。
3. 事务状态存储在一个新的内部主题中__transaction_state。直到首次尝试使用事务性请求API时才创建此主题。类似于消费者偏移主题，有几个设置来控制主题的配置。例如，transaction.state.log.min.isr控制此主题的最小ISR。请参阅用户指南中的配置部分以获取完整的选项列表。
4. 对于安全集群，事务性API需要新的ACL，可以使用新的ACL打开bin/kafka-acls.sh。工具。
5. Kafka的EoS引入了新的请求API，并修改了几个现有的API。有关 完整的详细信息，请参阅 KIP-98

关于0.11.0中新消息格式的说明

为了支持生产者更好的交付语义（见KIP-98）和改进的复制容错能力（见KIP-101），0.11.0消息格式包括几个主要的增强。虽然新格式包含更多信息以使这些改进成为可能，但是我们已经使批处理格式更有效率。只要每批消息的数量大于2，就可以降低整体开销。然而，对于较小的批次，可能会有一个小的性能影响。请参阅这里了解我们对新消息格式的初始性能分析结果。您还可以在KIP-98提案中找到关于消息格式的更多细节 。

新消息格式的显着差异之一是，即使未压缩的消息一起存储为一个批次。这对代理配置有一些影响max.message.bytes，这会限制单个批次的大小。首先，如果一个较老的客户端使用旧格式产生消息到一个主题分区，并且这些消息比单独的小 max.message.bytes，那么代理可能会在上转换过程中合并为一个批次后拒绝它们。通常，这可能发生在单个消息的聚合大小大于max.message.bytes。老年消费者阅读从新格式下转换的消息也有类似的效果：如果提取大小没有被设置为至少与 max.message.bytes即使单个未压缩的消息小于配置的获取大小，消费者也可能无法取得进展。此行为不影响Java客户端的0.10.1.0及更高版本，因为它使用更新的获取协议，该协议确保即使超过获取大小也能返回至少一条消息。为了解决这些问题，你应该确保1）生产者的批量大小没有被设置得大于max.message.bytes，并且2）消费者的获取大小被设置为至少与max.message.bytes。

大多数关于升级到0.10.0消息格式对性能影响的讨论 仍然与0.11.0升级有关。这主要影响不使用TLS保护的群集，因为在这种情况下，“零复制”传输已经不可行。为了避免下变换成本，您应该确保客户应用程序升级到最新的0.11.0客户端。重要的是，由于旧消费者已经在0.11.0.0中被弃用，所以它不支持新的消息格式。您必须升级才能使用新消费者使用新的消息格式，而不需要下转换成本。请注意，0.11.0消费者支持与经纪商0.10.0向下兼容，因此可以先在经纪商之前升级客户。

从0.8.x，0.9.x，0.10.0.x或0.10.1.x升级到0.10.2.0

0.10.2.0有线协议变化。通过遵循以下建议的滚动升级计划，您可以保证在升级过程中不会出现停机。但是，请在升级之前查看0.10.2.0中的显着更改。

从版本0.10.2开始，Java客户端（生产者和消费者）已经获得了与旧代理进行通信的能力。版本0.10.2客户可以与版本0.10.0或更新的代理进行通话。但是，如果您的经纪人年龄大于0.10.0，则必须先升级Kafka集群中的所有经纪人，然后再升级您的客户。0.10.2版经纪人支持0.8.x和更新的客户端。

对于滚动升级：

1. 更新所有代理上的server.properties文件并添加以下属性：
   • inter.broker.protocol.version = CURRENT_KAFKA_VERSION（例如0.8.2,0.9.0,0.10.0或0.10.1）。
   • log.message.format.version = CURRENT_KAFKA_VERSION（请参阅升级后潜在的性能影响，了解有关此配置的详细信息。）
2. 一次升级一个代理：关闭代理，更新代码并重新启动代理。
3. 一旦整个群集升级，通过编辑inter.broker.protocol.version并将其设置为0.10.2来提升协议版本。
4. 如果以前的消息格式是0.10.0，则将log.message.format.version更改为0.10.2（因为消息格式对于0.10.0，0.10.1和0.10.2而言是相同的，所以这是无操作的）。如果以前的消息格式版本低于0.10.0，则不要更改log.message.format.version - 只有在所有使用者已升级到0.10.0.0或更高版本后，才应更改此参数。
5. 重新启动代理，以使新的协议版本生效。
6. 如果log.message.format.version此时仍低于0.10.0，请等到所有使用者升级到0.10.0或更高版本，然后在每个代理上将log.message.format.version更改为0.10.2，重新启动他们一个。

注意：如果您愿意接受停机时间，您可以简单地关闭所有经纪人，更新代码并启动所有代理。他们将默认启动新的协议。

注意：在升级代理后，可以随时进行升级协议版本并重新启动。它不一定要在之后立即。

升级0.10.1 Kafka Streams应用程序

• 将Streams应用程序从0.10.1升级到0.10.2不需要代理升级。Kafka Streams 0.10.2应用程序可以连接到0.10.2和0.10.1代理（尽管如此，不能连接到0.10.0代理）。
• 你需要重新编译你的代码。只需交换Kafka Streams库jar文件将不起作用，并会破坏您的应用程序。
• 如果您使用自定义（即，用户实现的）时间戳提取器，则需要更新此代码，因为TimestampExtractor界面已更改。
• 如果您注册自定义指标，则需要更新此代码，因为StreamsMetric界面已更改。
• 请参阅0.10.2中的Streams API更改以获取更多详细信息。

0.10.2.1中的显着变化

• StreamsConfig类的两个配置的默认值已更改，以提高Kafka Streams应用程序的弹性。内部Kafka Streams生产者retries默认值从0更改为10.内部Kafka Streams消费者max.poll.interval.ms 默认值从300000更改为Integer.MAX_VALUE。

0.10.2.0中的显着变化

• Java客户端（生产者和消费者）已经获得了与旧经纪人沟通的能力。版本0.10.2客户可以与版本0.10.0或更新的代理进行通话。请注意，某些功能在使用旧代理时不可用或受限。
• InterruptException如果调用线程中断，则Java消费者的几个方法现在可能会抛出。请参阅KafkaConsumerJavadoc以获得对此更改的更深入的解释。
• Java消费者现在关闭优雅。默认情况下，消费者等待最多30秒以完成待处理的请求。已经添加了一个新的超时关闭API KafkaConsumer来控制最大等待时间。
• 用逗号分隔的多个正则表达式可以通过--whitelist选项与新的Java使用者一起传递给MirrorMaker。当使用旧的Scala消费者时，这使得行为与MirrorMaker一致。
• 将Streams应用程序从0.10.1升级到0.10.2不需要代理升级。Kafka Streams 0.10.2应用程序可以连接到0.10.2和0.10.1代理（尽管如此，不能连接到0.10.0代理）。
• Zookeeper依赖项已从Streams API中删除。Streams API现在使用Kafka协议来管理内部主题，而不是直接修改Zookeeper。这消除了直接访问Zookeeper的权限，不应该在Streams应用中设置“StreamsConfig.ZOOKEEPER_CONFIG”。如果Kafka集群受到保护，Streams应用程序必须具有所需的安全权限才能创建新主题。
• StreamsConfig类添加了几个新的字段，包括“security.protocol”，“connections.max.idle.ms”，“retry.backoff.ms”，“reconnect.backoff.ms”和“request.timeout.ms”。用户应该注意默认值，并根据需要进行设置。欲了解更多详情，请参阅3.5卡夫卡流配置。

新协议版本

• KIP-88：OffsetFetchRequest v2支持检索所有主题的偏移量，如果topics数组设置为null。
• KIP-88：OffsetFetchResponse v2引入了一个*error_code字段。
• KIP-103：UpdateMetadataRequest v3 listener_name为end_points数组的元素引入一个字段。
• KIP-108：CreateTopicsRequest v1引入了一个validate_only字段。
• KIP-108：CreateTopicsResponse v1 error_message为topic_errors数组的元素引入一个字段。

从0.8.x，0.9.x或0.10.0.X升级到0.10.1.0

0.10.1.0有线协议变化。通过遵循以下建议的滚动升级计划，您可以保证在升级过程中不会出现停机。但是，请注意在升级之前0.10.1.0中的潜在中断更改。 
注意：由于引入了新协议，所以在升级客户端之前升级您的Kafka群集非常重要（即，0.10.1.x客户端仅支持0.10.1.x或更高版本的代理，而0.10.1.x代理也支持较旧的客户端） 。

对于滚动升级：

1. 更新所有代理上的server.properties文件并添加以下属性：
   • inter.broker.protocol.version = CURRENT_KAFKA_VERSION（例如0.8.2.0，0.9.0.0或0.10.0.0）。
   • log.message.format.version = CURRENT_KAFKA_VERSION（请参阅升级后潜在的性能影响，了解有关此配置的详细信息。）
2. 一次升级一个代理：关闭代理，更新代码并重新启动代理。
3. 一旦整个群集升级，通过编辑inter.broker.protocol.version并将其设置为0.10.1.0来提升协议版本。
4. 如果以前的消息格式为0.10.0，则将log.message.format.version更改为0.10.1（由于0.10.0和0.10.1的消息格式相同，因此这是无操作）。如果以前的消息格式版本低于0.10.0，则不要更改log.message.format.version - 只有在所有使用者已升级到0.10.0.0或更高版本后，才应更改此参数。
5. 重新启动代理，以使新的协议版本生效。
6. 如果log.message.format.version此时仍低于0.10.0，请等到所有使用者升级到0.10.0或更高版本，然后在每个代理上将log.message.format.version更改为0.10.1，重新启动他们一个。

注意：如果您愿意接受停机时间，您可以简单地关闭所有经纪人，更新代码并启动所有代理。他们将默认启动新的协议。

注意：在升级代理后，可以随时进行升级协议版本并重新启动。它不一定要在之后立即。

0.10.1.0中潜在的突破变化

• 日志保留时间不再基于日志段的上次修改时间。相反，它将基于日志段中消息的最大时间戳。
• 日志滚动时间不再取决于日志段创建时间。相反，它现在基于消息中的时间戳。进一步来说。如果段中第一条消息的时间戳是T，则当新消息的时间戳大于或等于T + log.roll.ms
• 由于每个段添加了时间索引文件，0.10.0的开放文件处理程序将增加〜33％。
• 时间索引和偏移索引共享相同的索引大小配置。由于每次索引条目是偏移索引条目大小的1.5倍。用户可能需要增加log.index.size.max.bytes以避免潜在的频繁的日志滚动。
• 由于索引文件数量的增加，在一些具有大量日志段（例如大于15K）的代理上，代理启动期间的日志加载过程可能会更长。根据我们的实验，将num.recovery.threads.per.data.dir设置为1可能会减少日志加载时间。

升级0.10.0 Kafka Streams应用程序

• 将Streams应用程序从0.10.0升级到0.10.1确实需要代理升级，因为Kafka Streams 0.10.1应用程序只能连接到0.10.1代理。
• 有几个API的变化，这是不是向后兼容（更多的细节比较0.10.1流更改API）。因此，您需要更新并重新编译您的代码。只需交换Kafka Streams库jar文件将不起作用，并会破坏您的应用程序。

0.10.1.0中的显着变化

• 新的Java消费者不再处于测试阶段，我们推荐所有新的开发。旧的Scala消费者仍然支持，但是在下一个版本中它们将被弃用，并将在未来的主要版本中被删除。
• 将--new-consumer/ --new.consumer开关不再需要使用像MirrorMaker和控制台消费者与消费者的新工具; 只需要通过一个卡夫卡经纪人来连接，而不是ZooKeeper合奏。此外，与旧用户一起使用控制台消费者已被弃用，并将在未来的主要版本中删除。
• 卡夫卡群集现在可以由群集ID唯一标识。当代理升级到0.10.1.0时，它会自动生成。群集ID可通过kafka.server：type = KafkaServer，name = ClusterId指标获得，它是元数据响应的一部分。串行器，客户端拦截器和度量记录器可以通过实现ClusterResourceListener接口来接收集群ID。
• BrokerState“RunningAsController”（值4）已被删除。由于一个错误，一个经纪人在转换出来之前只会处于这个状态，因此移除的影响应该是最小的。检测指定代理是否为控制器的推荐方法是通过kafka.controller：type = KafkaController，name = ActiveControllerCount指标。
• 新的Java使用者现在允许用户通过分区上的时间戳搜索偏移量。
• 新的Java使用者现在支持来自后台线程的心跳。有一个新的配置 max.poll.interval.ms，它控制在用户主动离开组之前轮询调用之间的最大时间（默认为5分钟）。配置的值 request.timeout.ms必须总是大于max.poll.interval.ms因为这是JoinGroup请求在服务器重新平衡时可以阻止的最大时间，所以我们已经将其默认值更改为刚好在5分钟以上。最后，默认值session.timeout.ms已经调整到10秒，默认值max.poll.records已经改为500。
• 当使用授权人，并且用户没有描述对主题的授权时，代理将不再向请求返回TOPIC_AUTHORIZATION_FAILED错误，因为这会泄漏主题名称。相反，UNKNOWN_TOPIC_OR_PARTITION错误代码将被返回。这可能会导致使用生产者和使用者时出现意外超时或延迟，因为Kafka客户端通常会在未知主题错误时自动重试。如果您怀疑这可能会发生，您应该咨询客户端日志。
• 提取响应的默认大小限制（消费者为50 MB，复制为10 MB）。现有的每个分区限制也适用（1 MB用于消费者和复制）。请注意，这些限制都不是下一个解释的绝对最大值。
• 如果发现大于响应/分区大小限制的消息，则消费者和副本可以取得进展。更具体地说，如果提取的第一个非空分区中的第一条消息大于一个或两个限制，则该消息仍将被返回。
• 被添加的重载构造函数被添加到kafka.api.FetchRequest并kafka.javaapi.FetchRequest允许调用者指定分区的顺序（因为在v3中顺序是重要的）。先前存在的构造函数已被弃用，在发送请求之前将分区进行混洗，以避免饥饿问题。

新协议版本

• ListOffsetRequest v1支持基于时间戳的精确偏移搜索。
• MetadataResponse v2引入了一个新的字段：“cluster_id”。
• FetchRequest v3支持限制响应大小（除现有的每个分区限制之外），如果需要进行更改，则返回大于限制的消息，并且请求中的分区顺序现在很重要。
• JoinGroup v1引入了一个新的字段：“rebalance_timeout”。

从0.8.x或0.9.x升级到0.10.0.0

0.10.0.0有潜在的重大变化（请在升级之前查看）以及升级后可能的 性能影响。通过遵循以下建议的滚动升级计划，可确保在升级过程中和升级后不会出现停机和性能影响。 
注意：由于引入了新的协议，在升级客户端之前升级您的Kafka集群非常重要。

对于版本为0.9.0.0的客户的说明：由于在0.9.0.0中引入了一个错误，依赖于ZooKeeper的客户端（旧Scala高级Consumer和MirrorMaker，如果与旧客户一起使用）将不能与0.10.0.x代理。因此，在将代理升级到0.10.0.x 之前，应将0.9.0.0客户升级到0.9.0.1。对于0.8.X或0.9.0.1客户端，这一步不是必需的。

对于滚动升级：

1. 更新所有代理上的server.properties文件并添加以下属性：
   • inter.broker.protocol.version = CURRENT_KAFKA_VERSION（例如0.8.2或0.9.0.0）。
   • log.message.format.version = CURRENT_KAFKA_VERSION（请参阅升级后潜在的性能影响，了解有关此配置的详细信息。）
2. 升级经纪人。这可以通过简单地将其关闭，更新代码并重新启动它来完成。
3. 一旦整个群集升级，通过编辑inter.broker.protocol.version并将其设置为0.10.0.0来冲击协议版本。注意：您不应该触摸log.message.format.version - 只有当所有使用者升级到0.10.0.0后，该参数才会更改
4. 重新启动代理，以使新的协议版本生效。
5. 一旦所有使用者升级到0.10.0，在每个代理上将log.message.format.version更改为0.10.0，然后逐个重新启动。

注意：如果您愿意接受停机时间，您可以简单地关闭所有经纪人，更新代码并启动所有代理。他们将默认启动新的协议。

注意：在升级代理后，可以随时进行升级协议版本并重新启动。它不一定要在之后立即。

在升级到0.10.0.0后，可能会对性能造成影响

0.10.0中的消息格式包含一个新的时间戳字段，并使用压缩消息的相对偏移量。磁盘上的消息格式可以通过server.properties文件中的log.message.format.version来配置。默认的磁盘上消息格式是0.10.0。如果消费者客户端使用的是0.10.0.0之前的版本，则只能理解0.10.0之前的消息格式。在这种情况下，代理可以将消息从0.10.0格式转换为较早的格式，然后将响应发送给旧版本的使用者。但是，在这种情况下，经纪人不能使用零复制转移。卡夫卡社区对性能影响的报告显示，在升级之后，CPU使用率从20％上升到100％，迫使所有客户端立即升级，以使性能恢复正常。为避免消费者升级到0.10.0.0之前的这种消息转换，可以将代理升级到0.10.0.0时将log.message.format.version设置为0.8.2或0.9.0。这样，经纪人仍然可以使用零拷贝转移将数据发送给旧消费者。消费者升级之后，可以将代理上的消息格式更改为0.10.0，并享受包含新时间戳和改进压缩的新消息格式。支持该转换以确保兼容性，对于支持尚未更新到较新客户端的少数应用程序可能会有用，但即使在过度配置的群集上也不支持所有使用者流量。因此，在经纪人升级后尽可能避免信息转换是非常重要的，但大多数客户还没有。将代理升级到0.10.0.0时，message.format.version为0.8.2或0.9.0。这样，经纪人仍然可以使用零拷贝转移将数据发送给旧消费者。消费者升级之后，可以将代理上的消息格式更改为0.10.0，并享受包含新时间戳和改进压缩的新消息格式。支持该转换以确保兼容性，对于支持尚未更新到较新客户端的少数应用程序可能会有用，但即使在过度配置的群集上也不支持所有使用者流量。因此，在经纪人升级后尽可能避免信息转换是非常重要的，但大多数客户还没有。将代理升级到0.10.0.0时，message.format.version为0.8.2或0.9.0。这样，经纪人仍然可以使用零拷贝转移将数据发送给旧消费者。消费者升级之后，可以将代理上的消息格式更改为0.10.0，并享受包含新时间戳和改进压缩的新消息格式。支持该转换以确保兼容性，对于支持尚未更新到较新客户端的少数应用程序可能会有用，但即使在过度配置的群集上也不支持所有使用者流量。因此，在经纪人升级后尽可能避免信息转换是非常重要的，但大多数客户还没有。经纪人仍然可以使用零拷贝转移将数据发送给旧消费者。消费者升级之后，可以将代理上的消息格式更改为0.10.0，并享受包含新时间戳和改进压缩的新消息格式。支持该转换以确保兼容性，对于支持尚未更新到较新客户端的少数应用程序可能会有用，但即使在过度配置的群集上也不支持所有使用者流量。因此，在经纪人升级后尽可能避免信息转换是非常重要的，但大多数客户还没有。经纪人仍然可以使用零拷贝转移将数据发送给旧消费者。消费者升级之后，可以将代理上的消息格式更改为0.10.0，并享受包含新时间戳和改进压缩的新消息格式。支持该转换以确保兼容性，对于支持尚未更新到较新客户端的少数应用程序可能会有用，但即使在过度配置的群集上也不支持所有使用者流量。因此，在经纪人升级后尽可能避免信息转换是非常重要的，但大多数客户还没有。支持该转换以确保兼容性，对于支持尚未更新到较新客户端的少数应用程序可能会有用，但即使在过度配置的群集上也不支持所有使用者流量。因此，在经纪人升级后尽可能避免信息转换是非常重要的，但大多数客户还没有。支持该转换以确保兼容性，对于支持尚未更新到较新客户端的少数应用程序可能会有用，但即使在过度配置的群集上也不支持所有使用者流量。因此，在经纪人升级后尽可能避免信息转换是非常重要的，但大多数客户还没有。

对于升级到0.10.0.0的客户端，不会影响性能。

注意：通过设置消息格式版本，可以证明所有现有消息都在该消息格式版本之上或之下。否则0.10.0.0之前的消费者可能会中断。特别是，在消息格式设置为0.10.0之后，不应将其更改回以前的格式，因为它可能会破坏0.10.0.0之前的版本中的使用者。

注意：由于在每条消息中引入了额外的时间戳，发送小消息的生产者可能由于增加的开销而看到消息吞吐量下降。同样，复制现在每个消息传输额外的8个字节。如果您运行的是接近集群的网络容量，则可能会淹没网卡，并看到由于过载而导致的故障和性能问题。

注意：如果您在生产者上启用了压缩，在某些情况下，您可能会注意到生产者吞吐量降低和/或代理上的压缩率降低。在接收压缩消息时，0.10.0的代理避免重新压缩消息，这通常减少了延迟并提高了吞吐量。然而，在某些情况下，这可能会减少生产者的配料尺寸，这可能导致更差的吞吐量。如果发生这种情况，用户可以调整生产者的linger.ms和batch.size以获得更好的吞吐量。另外，用于压缩snappy消息的生产者缓冲区比代理使用的缓冲区小，这可能会对磁盘上​​的消息的压缩率产生负面影响。我们打算在未来的Kafka版本中进行配置。

0.10.0.0中潜在的重大更改

• 从Kafka 0.10.0.0开始，Kafka中的消息格式版本被表示为Kafka版本。例如，消息格式0.9.0指的是Kafka 0.9.0支持的最高消息版本。
• 消息格式0.10.0已经被引入，并且被默认使用。它包含消息中的时间戳字段，相对偏移量用于压缩消息。
• ProduceRequest / Response v2已被引入，默认情况下使用它来支持消息格式0.10.0
• FetchRequest / Response v2已被引入，默认情况下使用FetchRequest / Response v2来支持消息格式0.10.0
• MessageFormatter接口已从更改def writeTo(key: Array[Byte], value: Array[Byte], output: PrintStream)为 def writeTo(consumerRecord: ConsumerRecord[Array[Byte], Array[Byte]], output: PrintStream)
• MessageReader接口从def readMessage(): KeyedMessage[Array[Byte], Array[Byte]]改为 def readMessage(): ProducerRecord[Array[Byte], Array[Byte]]
• MessageFormatter的包已经被更改kafka.tools为kafka.common
• MessageReader的包已经被更改kafka.tools为kafka.common
• MirrorMakerMessageHandler不再公开该handle(record: MessageAndMetadata[Array[Byte], Array[Byte]])方法，因为它从来没有被调用过。
• 0.7 KafkaMigrationTool不再与Kafka打包在一起。如果您需要从0.7迁移到0.10.0，请先迁移到0.8，然后按照记录的升级过程从0.8升级到0.10.0。
• 新的消费者已经将其API标准化，以接受java.util.Collection作为方法参数的序列类型。现有的代码可能需要更新才能使用0.10.0客户端库。
• LZ4压缩的消息处理已更改为使用可互操作的帧规范（LZ4f v1.5.1）。为了保持与旧客户端的兼容性，此更改仅适用于消息格式0.10.0及更高版本。使用v0 / v1（消息格式0.9.0）生成/获取LZ4压缩消息的客户端应该继续使用0.9.0成帧实现。使用Produce / Fetch协议v2或更高版本的客户端应使用可互操作的LZ4f成帧。可在http://www.lz4.org/上找到可互操作的LZ4库列表。

0.10.0.0中的显着变化

• 从Kafka 0.10.0.0开始，一个名为Kafka Streams的新客户端库可用于对存储在Kafka主题中的数据进行流处理。由于上面提到的消息格式更改，此新客户端库仅适用于0.10.x版本和向上版本的代理。欲了解更多信息，请阅读Streams文档。
• 新用户的配置参数的默认值receive.buffer.bytes现在是64K。
• 新的使用者现在公开配置参数exclude.internal.topics以限制内部主题（例如消费者偏移主题）被意外地包含在正则表达式订阅中。默认情况下，它被启用。
• 旧的斯卡拉生产者已被弃用。用户应尽快将其代码迁移到包含在kafka-clients JAR中的Java生产者。
• 新的消费者API已被标记为稳定。

从0.8.0，0.8.1.X或0.8.2.X升级到0.9.0.0

0.9.0.0有潜在的重大变化（请在升级之前查看）以及与之前版本的代理间协议更改。这意味着升级的经纪人和客户可能与旧版本不兼容。在升级客户端之前，升级您的Kafka集群非常重要。如果您正在使用MirrorMaker，则应首先升级下游群集。

对于滚动升级：

1. 更新所有代理上的server.properties文件并添加以下属性：inter.broker.protocol.version = 0.8.2.X
2. 升级经纪人。这可以通过简单地将其关闭，更新代码并重新启动它来完成。
3. 一旦整个群集升级，通过编辑inter.broker.protocol.version并将其设置为0.9.0.0来打破协议版本。
4. 重新启动代理，以使新的协议版本生效

注意：如果您愿意接受停机时间，您可以简单地关闭所有经纪人，更新代码并启动所有代理。他们将默认启动新的协议。

注意：在升级代理后，可以随时进行升级协议版本并重新启动。它不一定要在之后立即。

0.9.0.0中潜在的重大更改

• Java 1.6不再支持。
• 斯卡拉2.9不再支持。
• 超过1000的经纪商ID现在默认保留为自动分配的经纪商ID。如果您的群集具有高于该阈值的现有代理ID，请确保相应地增加reserved.broker.max.id代理配置属性。
• 配置参数replica.lag.max.messages被删除。当决定哪些副本同步时，分区领导将不再考虑滞后消息的数量。
• 现在，配置参数replica.lag.time.max.ms不仅指自上次从副本获取请求之后经过的时间，还指自上次复制最后一次被抓取的时间。仍然从领导获取消息的副本，但没有赶上replica.lag.time.max.ms中的最新消息将被视为不同步。
• 压缩主题不再接受没有**的消息，并且如果尝试这样做，生产者抛出异常。在0.8.x中，没有键的消息会导致日志压缩线程随后发出抱怨并退出（并停止压缩所有压缩的主题）。
• MirrorMaker不再支持多个目标群集。因此它只接受一个--consumer.config参数。要镜像多个源群集，每个源群集至少需要一个MirrorMaker实例，每个群集都有自己的消费者配置。
• 在org.apache.kafka.clients.tools。*下打包的工具已被移至org.apache.kafka.tools。*。所有包含的脚本仍然照常运行，只有直接导入这些类的自定义代码才会受到影响。
• kafka-run-class.sh中已经更改了默认的Kafka JVM性能选项（KAFKA_JVM_PERFORMANCE_OPTS）。
• kafka-topics.sh脚本（kafka.admin.TopicCommand）现在以非零退出代码退出。
• kafka-topics.sh脚本（kafka.admin.TopicCommand）现在将在主题名称由于使用“。”而导致风险度量标准冲突时显示警告。或主题名称中的“_”，以及实际发生碰撞时的错误。
• kafka-console-producer.sh脚本（kafka.tools.ConsoleProducer）将使用Java生产者而不是旧的Scala生产者作为默认的，并且用户必须指定“old-producer”来使用旧的生产者。
• 默认情况下，所有命令行工具都将打印所有日志消息到stderr而不是stdout。

0.9.0.1中的显着变化

• 可以通过将broker.id.generation.enable设置为false来禁用新的代理ID生成功能。
• 配置参数log.cleaner.enable默认为true。这意味着具有cleanup.policy = compact的主题现在将被默认压缩，128 MB的堆将通过log.cleaner.dedupe.buffer.size分配给清除进程。您可能需要根据您使用的压缩主题来查看log.cleaner.dedupe.buffer.size和其他log.cleaner配置值。
• 新用户的配置参数fetch.min.bytes的默认值现在默认为1。

在0.9.0.0中弃用

• 已经不建议使用kafka-topics.sh脚本（kafka.admin.TopicCommand）更改主题配置。今后，请使用kafka-configs.sh脚本（kafka.admin.ConfigCommand）来实现此功能。
• kafka-consumer-offset-checker.sh（kafka.tools.ConsumerOffsetChecker）已被弃用。今后，请使用kafka-consumer-groups.sh（kafka.admin.ConsumerGroupCommand）来实现此功能。
• kafka.tools.ProducerPerformance类已被弃用。今后，请使用org.apache.kafka.tools.ProducerPerformance来实现此功能（kafka-producer-perf-test.sh也将更改为使用新类）。
• 生产者配置block.on.buffer.full已被弃用，并将在未来的版本中被删除。目前其默认值已被更改为false。KafkaProducer将不再抛出BufferExhaustedException，而是使用max.block.ms值来阻塞，之后将抛出一个TimeoutException。如果将block.on.buffer.full属性显式设置为true，则会将max.block.ms设置为Long.MAX_VALUE，并且不会将metadata.fetch.timeout.ms

从0.8.1升级到0.8.2

0.8.2与0.8.1完全兼容。升级可以通过简单地将其关闭，更新代码并重新启动来一次完成一个代理。

从0.8.0升级到0.8.1

0.8.1与0.8完全兼容。升级可以通过简单地将其关闭，更新代码并重新启动来一次完成一个代理。

从0.7升级

版本0.7与新版本不兼容。对API，ZooKeeper数据结构，协议和配置进行了重大更改，以便添加复制（0.7中缺少）。从0.7升级到更高版本需要专门的迁移工具。这种迁移可以在没有停机的情况下完成。