Kafka 0.8协议

• 介绍


• 概述


• 预备知识


• 网络


• 分区和引导


• 分区策略


• 批量处理


• 版本控制和兼容性


• 协议


• Protocol
  Primitive Types


• Notes
  on reading the request format grammars


• Common
  Request and Response Structure


• Requests


• Responses


• Message
  sets


• Compression


• The
  APIs


• Metadata
  API


• Metadata
  Request


• Metadata
  Response


• Produce
  API


• Produce
  Request


• Produce
  Response


• Fetch
  API


• Fetch
  Request


• Fetch
  Response


• Offset
  API


• Offset
  Request


• Offset
  Response


• Offset
  Commit/Fetch API


• Consumer
  Metadata Request


• Consumer
  Metadata Response


• Offset
  Commit Request


• Offset
  Commit Response


• Offset
  Fetch Request


• Offset
  Fetch Response


• Constants
  


• Api
  Keys


• Error
  Codes


• Some
  Common Philosophical Questions

介绍

该文档覆盖Kafka
0.8及后续版本所实现的协议，旨在提供一份可读性强的指导手册，包括可用请求、二进制格式和利用协议实现客户端等内容。该文档假设你理解此处描述的基本设计原则和术语。

0.7及之前版本的协议与当前版本类似，但我们选择一次性打破兼容性，来去除一些令人讨厌的东西以及产生新东西。

部分术语

metadata：元数据

broker：[暂无好用翻译]

message：消息

offset：位移

topic：主题

group：群组

partition：分区

概述

Kafka协议相当简单，只有六个客户端请求API：

1. Metadata – 描述当前可用brokers，以及它们的host和port，并提供哪个broker有哪些partitions的信息。


2. Send – 发送消息至broker。


3. Fetch – 从broker获取消息，包括获取数据、获取集群的metadata(元数据)、获取offset(位移)信息。


4. Offsets – 获取指定topic中partition的可用位移信息。


5. Offset Commit – 提交一个消费者group的位移集合。


6. Offset Fetch – 获取一个消费者group的位移集合。

每项内容会在之后详细描述。

预备知识

网络

Kafka使用基于TCP的二进制协议。协议定义了所有请求/响应消息对的API。所有消息有大小限定，并由以下主要类型组成。

客户端初始化一个socket连接，之后写入一系列请求消息，并读取相应的响应消息。在连接和断开连接时不需要执行握手。TCP乐于让你为许多请求保持长连接，并分摊TCP握手的开销。即便不考虑这些，TCP的开销也相当小。

由于数据分区的原因，客户端需要保持对多个brokers的连接，与包含数据的服务端保持通信。尽管如此，一个客户端实例一般没有必要保持对单个broker的多连接(如连接池)。

服务端保证在单个TCP连接上，请求会按照发送时的顺序来执行，返回时的顺序也一样。broker的请求处理模块对于一个连接只允许一个请求在处理中，以此来保证处理的顺序性。注意，客户端可以(理想上应该)使用非阻塞IO来实现请求管道及获取高吞吐量。例如，客户端可以在等待之前请求的响应结果时继续发送请求，因为未处理请求会缓存在底层操作系统的socket缓冲区中。除非有特别说明，否则所有的请求由客户端初始化，并从服务器获取响应结果消息。

服务器端可配置请求大小的最大限制，超过限制则导致socket重连。

分区和引导

Kafka是一种分区系统，因此并非所有的服务器都包含完整的数据集。topics被分割到预定义数目P的分区中，每个分区复制为N份拷贝。topic分区以0,
1, …提交日志顺序[？翻译有些不通顺]。

分区特性在所有系统上都有数据如何分配分区的问题。Kafka客户端直接控制分区分配，brokers没有何种消息该发布到哪个分区的特定语义。客户端发布消息时直接提交消息到指定分区，获取消息时直接从指定分区获取。若两个客户端想使用相同的分区模式，则它们必须有同样的方法来计算key到partition的映射。

这些发布和获取数据的请求必须发送到当前作为指定分区leader的broker上。该条件只broker执行，所以对特定分区的请求若发送到错误的broker上，会导致返回NotLeaderForPartition错误码(后详述)。

客户端如何找出哪些topics存在、包含哪些分区、这些分区在哪些brokers上，从而发送请求到正确的主机上。这个信息是动态的，所以不能以静态映射文件形式配置在客户端。Kafka的所有brokers可以响应metadata请求，以及描述集群的当前状态。

换句话说，客户端需要找到其中一个broker，该broker会告知客户端其他存在的brokers以及它们持有的分区。第一个broker可能无效，因此客户端的实现需要包括两或三个引导启动的URL。用户可以选择在客户端使用负载均衡器或只是静态配置两或三个kafka的hosts。

客户端不需要保持轮询来检测集群是否有变更，它可以在初始化的时候获取并缓存metadata，直到它收到标示metadata过期的错误。这个错误有两种形式：

1)     
socket错误，表示客户端不能与特定的broker通信。

2)      该broker不再持有所请求的数据

1. 循环访问kafka”引导”地址，直到找到可以连接的一个。获取集群metadata。


2. 处理获取和生产请求，发送到指定topic/partitions相应的broker


3. 若产生相应的错误，刷新metadata并重试。

分区策略

上面提到消息分配分区由生产者客户端来控制。这意味着，如何将该功能展示给终端用户是一个问题。

Kafka分区实际上有两个目的：

1. 在brokers之间均衡数据和请求负载It balances data and request load over brokers


2. 在分区中允许本地状态和维持顺序，从而在消费者进程间分摊处理。称之为语义分区。

对于给定的应用场景，你可能只关注其中一个或者全部。

为了实现简单的负载均衡，一个简单的客户端实现方法是，对于所有brokers的请求采用Round-Robin方式。在生产者producers远多于brokers的环境中，可以采用单个客户端随机选择一个partition发布消息的方式作为另一个选择。后一个策略的TCP连接更少。

语义分区的意思是使用消息中的某种key来分配消息到不同分区。比如，如果你在处理点击消息流，你可能想通过用户ID来分区，从而使一个特定用户的所有数据进入一个消费者。为了满足这个需求，客户端可以采用与消息相关的某个key，并应用哈希算法来选择分区。

批量处理

我们的API鼓励对批量小数据放在一起进行批处理，以提升效率。我们发现批处理可以获取很大的性能提升。发送和获取消息的API都鼓励操作批量消息而非单条消息。智能的客户端可以利用这一点，并支持“异步”模式，将单个发送的消息分批处理，以更大的数据块发送数据。更进一步的说，允许跨多个topics和partitions的批处理，因此生产消息请求可能包括多个分区的数据，获取消息请求可能一次性从多个分区拉取数据。

客户端实现也可以选择一次发送一条消息，从而忽略这个特性。

版本管理和兼容性

协议设计需要能够在向后兼容模式上进行增量式进化[？翻译不顺]。版本控制建立在每个API的基础上，每个版本包含一个请求和响应对。每个请求包含一个API
key来标示调用的API，以及一个版本号来标示请求的格式及期望的响应格式。

这样做的目的是，客户端可以实现协议的特定版本，并在请求中标示该版本。我们的目标是，在不允许停机以及客户端和服务器不能一次性全部变更的情况下，允许API的进化。

服务端拒绝不支持的版本请求，并且响应给客户端，告知基于该请求版本的期望协议格式。期望的升级方式是，新的特性先在服务端铺开（旧客户端不会使用它们），当新客户端布署时，这些新特性逐渐得到使用。

当前所有版本的基数值为0，我们引入这些API时，会指出每个版本各自的格式。

协议

Protocol Primitive Types

The protocol is built out of the following primitive types.

Fixed Width
Primitives

int8, int16, int32, int64 - Signed integers with
the given precision (in bits) stored in big endian order.

Variable Length
Primitives

bytes, string - These types consist of a signed integer giving a
length N followed by N bytes of content. A length of -1 indicates null. string
uses an int16 for its size, and bytes uses an int32.

Arrays

This is a notation for handling repeated structures. These will
always be encoded as an int32 size containing the length N followed by N
repetitions of the structure which can itself be made up of other primitive
types. In the BNF grammars below we will show an array of a structure foo as
[foo].

Optional
entries

Certain fields may be present under certain conditions (say if
there is no error). These are represented as |foo|.

Notes on reading the request format
grammars

The BNFs
below give an exact context free grammar for the request and response binary
format. For each API I will give the request and response together followed by
all the sub-definitions. The BNF is intentionally not compact in order to give
human-readable name (for example I define a production for ErrorCode even though
it is just an int16 in order to give it a symbolic name). As always in a BNF a
sequence of productions indicates concatenation, so the MetadataRequest given
below would be a sequence of bytes containing first a VersionId, then a
ClientId, and then an array of TopicNames (each of which has its own
definition). Productions are always given in camel case and primitive types in
lower case. When there are multiple possible productions these
are separated with ‘|‘ and may be enclosed in parenthesis for
grouping. The top-level definition is always given first and subsequent
sub-parts are indented.

Common Request and Response Structure

All requests and responses originate from the following grammar
which will be incrementally describe through the rest of this document:

Requests

Requests all have the following format:

The various request and response messages will be described
below.

Responses

The response will always match the paired request (e.g. we will
send a MetadataResponse in return to a MetadataRequest).

Message sets

One structure common to both the produce and fetch requests is
the message set format. A message in kafka is a key-value pair with a small
amount of associated metadata. A message set is just a sequence of messages with
offset and size information. This format happens to be used both for the on-disk
storage on the broker and the on-the-wire format.

A message set is also the unit of compression in Kafka, and we
allow messages to recursively contain compressed message sets to allow batch
compression.

N.B., MessageSets are not preceded by an int32 like other array
elements in the protocol.

Message format

Compression

Kafka supports compressing messages for additional efficiency,
however this is more complex than just compressing a raw message. Because
individual messages may not have sufficient redundancy to enable good
compression ratios, compressed messages must be sent in special batches
(although you may use a batch of one if you truly wish to compress a message on
its own). The messages to be sent are wrapped (uncompressed) in a MessageSet
structure, which is then compressed and stored in the Value field of a single
"Message" with the appropriate compression codec set. The receiving system
parses the actual MessageSet from the decompressed value.

Kafka currently supports two compression codecs with the
following codec numbers:

The APIs

This section gives details on each of the individual APIs, their
usage, their binary format, and the meaning of their fields.

Metadata API

This API answers the following questions:

• What topics exist?


• How many partitions does each topic have?


• Which broker is currently the leader for each partition?


• What is the host and port for each of these brokers

This is the only request that can be addressed to any broker in
the cluster.

Since there may be many topics the client can give an optional
list of topic names in order to only return metadata for a subset of topics.

The metdata returned is at the partition level, but grouped
together by topic for convenience and to avoid redundancy. For each partition
the metadata contains the information for the leader as well as for all the
replicas and the list of replicas that are currently in-sync.

Metadata Request

Metadata Response

The response contains metadata for each partition, with
partitions grouped together by topic. This metadata refers to brokers by their
broker id. The brokers each have a host and port.

Produce API

The produce API is used to send message sets to the server. For
efficiency it allows sending message sets intended for many topic partitions in
a single request.

The produce API uses the generic message set format, but since
no offset has been assigned to the messages at the time of the send the producer
is free to fill in that field in any way it likes.

Produce Request

Produce Response

Fetch API

The fetch API is used to fetch a chunk of one or more logs for
some topic-partitions. Logically one specifies the topics, partitions, and
starting offset at which to begin the fetch and gets back a chunk of
messages.

Fetch requests follow a long poll model so they can be made to
block for a period of time if sufficient data is not immediately available.

As an optimization the server is allowed to return a partial
message at the end of the message set. Clients should handle this case.

One thing to note is that the fetch API requires specifying the
partition to consume from. The question is how should a consumer know what
partitions to consume from? In particular how can you balance the partitions
over a set of consumers acting as a group so that each consumer gets a subset of
partitions. We have done this assignment dynamically using zookeeper for the
scala and java client. The downside of this approach is that it requires a
fairly fat client and a zookeeper connection. We haven‘t yet created a Kafka API
to allow this functionality to be moved to the server side and accessed more
conveniently. A simple consumer client can be implemented by simply requiring
that the partitions be specified in config, though this will not allow dynamic
reassignment of partitions should that consumer fail. We hope to address this
gap in the next major release.

Fetch Request

Fetch Response

Offset API

This API describes the valid offset range available for a set of
topic-partitions. As with the produce and fetch APIs requests must be directed
to the broker that is currently the leader for the partitions in question. This
can be determined using the metadata API.

The response contains the starting offset of each segment for
the requested partition as well as the "log end offset" i.e. the offset of the
next message that would be appended to the given partition.

We agree that this API is slightly funky.

Offset Request

Offset Response

Offset Commit/Fetch API

These APIs allow for centralized management of offsets. Read
more Offset
Management. As per comments on KAFKA-993
these API calls are not fully functional in releases until Kafka 0.8.1.1. It
will be available in the 0.8.2 release.

Consumer Metadata Request

The offsets for a given consumer group are maintained by a
specific broker called the offset coordinator. i.e., a consumer needs to issue
its offset commit and fetch requests to this specific broker. It can discover
the current offset coordinator by issuing a consumer metadata request.

Consumer Metadata Response

On a successful (ErrorCode == 0) response, the coordinator
fields provide the id/host/port details of the offset coordinator.

Offset Commit Request

(If the time stamp field is set to -1, then the broker sets the
time stamp to the receive time before committing the offset.)

Offset Commit Response

Offset Fetch Request

Offset Fetch Response

Note that if there is no offset associated with a
topic-partition under that consumer group the broker does not set an error code
(since it is not really an error), but returns empty metadata and sets the
offset field to -1.

Constants

Api Keys

The following are the numeric codes that the ApiKey in the
request can take for each of the above request types.

Error Codes

We use numeric codes to indicate what problem occurred on the
server. These can be translated by the client into exceptions or whatever the
appropriate error handling mechanism in the client language. Here is a table of
the error codes currently in use:

Some Common Philosophical Questions

Some people have asked why we don‘t use HTTP. There are a number
of reasons, the best is that client implementors can make use of some of the
more advanced TCP features--the ability to multiplex requests, the ability to
simultaneously poll many connections, etc. We have also found HTTP libraries in
many languages to be surprisingly shabby.

Others have asked if maybe we shouldn‘t support many different
protocols. Prior experience with this was that it makes it very hard to add and
test new features if they have to be ported across many protocol
implementations. Our feeling is that most users don‘t really see multiple
protocols as a feature, they just want a good reliable client in the language of
their choice.

Another question is why we don‘t adopt XMPP, STOMP, AMQP or an
existing protocol. The answer to this varies by protocol, but in general the
problem is that the protocol does determine large parts of the implementation
and we couldn‘t do what we are doing if we didn‘t have control over the
protocol. Our belief is that it is possible to do better than existing messaging
systems have in providing a truly distributed messaging system, and to do this
we need to build something that works differently.

A final question is why we don‘t use a system like Protocol
Buffers or Thrift to define our request messages. These packages excel at
helping you to managing lots and lots of serialized messages. However we have
only a few messages. Support across languages is somewhat spotty (depending on
the package). Finally the mapping between binary log format and wire protocol is
something we manage somewhat carefully and this would not be possible with these
systems. Finally we prefer the style of versioning APIs explicitly and checking
this to inferring new values as nulls as it allows more nuanced control
of compatibility.

Kafka 0.8协议,布布扣,bubuko.com