在 RHEL 中使用 AMQ Streams

AclAuthorizer 使用 ACL 规则管理对 Kafka 代理的访问。

ACL 规则以以下格式定义：

例如，可以设置规则以便用户：

主机是 John 连接的机器的 IP 地址。

在大多数情况下，用户是一个制作者或消费者应用程序：

如果 ACL 规则不存在

如果给定资源没有 ACL 规则，则拒绝所有操作。通过将 Kafka 配置文件 /opt/kafka/config/server.properties 中的属性 allow.everyone.if.no.acl.found 设置为 true，可更改此行为。

主体 表示用户的身份。ID 的格式取决于客户端用来连接到 Kafka 的身份验证机制：

DistinguishedName 是与客户端证书的可分辨名称。

Kerberos 用户名 是 Kerberos 主体的主要部分，在使用 Kerberos 连接时默认使用它。您可以使用 sasl.kerberos.principal.to.local.rules 属性配置从 Kerberos 主体构建 Kafka 主体的方式。

要使用授权，您需要启用身份验证并供您的客户端使用。否则，所有连接都有主体 User:ANONYMOUS。

有关身份验证方法的更多信息，请参阅 加密和身份验证。

无论 ACL 规则如何，超级用户都可执行所有操作。

超级用户在 Kafka 配置文件中使用属性 super.users 定义。

例如：

super.users=User:admin,User:operator

启用授权后，它将应用到所有监听器和所有连接。这包括用于在代理之间复制数据的代理间连接。因此，如果启用授权，请确保您对代理使用身份验证进行间隔，并授予代理使用的用户足够的权限。例如，如果代理之间的身份验证使用 kafka-broker 用户，那么超级用户配置必须包含用户名 super.users=User:kafka-broker。

您可以将 Kafka ACL 应用到这些资源类型：

AclAuthorizer 授权对资源的操作.

下表中带有 X 的字段标记每个资源支持的操作。

ACL 规则通过 bin/kafka-acls.sh 工具来管理，该工具作为 Kafka 分发软件包的一部分提供。

使用 kafka-acls.sh 参数选项添加、列出和删除 ACL 规则并执行其他功能。

参数需要双连规则，如 --add。

TLS 客户端身份验证只能用于已使用 TLS 加密的连接。要使用 TLS 客户端身份验证，可以向代理提供带有公钥的信任存储。这些密钥可用于验证连接到代理的客户端。信任存储应以 Java 密钥存储(JKS)格式提供，并且应包含证书颁发机构的公钥。所有由信任存储中包含的其中一个认证机构签名的公钥和私钥的客户端都将被验证。truststore 的位置使用字段 ssl.truststore.location 进行设置。如果信任存储受密码保护，应在 ssl.truststore.password 属性中设置密码。例如：

ssl.truststore.location=/path/to/keystore/server-1.jks
ssl.truststore.password=123456

配置信任存储后，必须使用 ssl.client.auth 属性启用 TLS 客户端身份验证。此属性可以设置为以下三个不同值之一：

当客户端使用 TLS 客户端身份验证进行身份验证时，经过身份验证的主体名称是与经过身份验证的客户端证书的区分名称。例如：具有可分辨名称 CN=someuser 的证书用户将使用以下主体 CN=someuser,OU=Unknown,O=Unknown,L=Unknown,ST=Unknown,C=Unknown 进行身份验证。如果不使用 TLS 客户端身份验证，并且禁用 SASL，则主体名称为 ANONYMOUS。

SASL 身份验证使用 Java 身份验证和授权服务(JAAS)进行配置。JAAS 也用于验证 Kafka 和 ZooKeeper 之间的连接。JAAS 使用自己的配置文件。此文件的建议位置是 /opt/kafka/config/jaas.conf。该文件必须可由 kafka 用户读取。运行 Kafka 时，使用 Java 系统属性 java.security.auth.login.config 指定此文件的位置。启动代理节点时，必须将此属性传递给 Kafka：

KAFKA_OPTS="-Djava.security.auth.login.config=/path/to/my/jaas.config"; bin/kafka-server-start.sh

SASL 身份验证既通过普通的未加密连接，也受 TLS 连接支持。可以为每个侦听器单独启用 SASL。要启用它，listener.security.protocol.map 中的安全协议必须是 SASL_PLAINTEXT 或 SASL_SSL。

Kafka 中的 SASL 身份验证支持多种不同的机制：

SASL 机制通过 JAAS 配置文件进行配置。Kafka 使用名为 KafkaServer 的 JAAS 上下文。在 JAAS 中配置它们后，必须在 Kafka 配置中启用 SASL 机制。这使用 sasl.enabled.mechanisms 属性完成。此属性包含一个以逗号分隔的启用机制列表：

sasl.enabled.mechanisms=PLAIN,SCRAM-SHA-256,SCRAM-SHA-512

如果用于broker 间通信的监听程序使用 SASL，则必须使用 sasl.mechanism.inter.broker.protocol 属性指定它应使用的 SASL 机制。例如：

sasl.mechanism.inter.broker.protocol=PLAIN

必须在 KafkaServer JAAS 上下文中使用 username 和 password 字段指定用于内部通信的用户名和密码。

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    user_admin="123456"
    user_user1="123456"
    user_user2="123456";
};

包含用户数据库的 JAAS 配置文件应该在所有 Kafka 代理上保持同步。

当 SASL PLAIN 也用于代理间验证时，username 和 password 属性应包含在 JAAS 上下文中：

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="123456"
    user_admin="123456"
    user_user1="123456"
    user_user2="123456";
};

KafkaServer {
    org.apache.kafka.common.security.scram.ScramLoginModule required;
};

在 Kafka 配置文件中启用 SASL 身份验证时，可以列出这两个 SCRAM 机制。但是，它们中只能选择其中之一作为代理间通信。例如：

sasl.enabled.mechanisms=SCRAM-SHA-256,SCRAM-SHA-512
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512

SCRAM 机制的用户凭据存储在 ZooKeeper 中。kafka-configs.sh 工具可用于管理它们。例如，运行以下命令来添加用户 user1，密码为 123456:

bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config 'SCRAM-SHA-256=[password=123456],SCRAM-SHA-512=[password=123456]' --entity-type users --entity-name user1

要删除用户凭证，请使用：

bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --delete-config 'SCRAM-SHA-512' --entity-type users --entity-name user1

KafkaServer {
    com.sun.security.auth.module.Krb5LoginModule required
    useKeyTab=true
    storeKey=true
    keyTab="/etc/security/keytabs/kafka_server.keytab"
    principal="kafka/kafka1.hostname.com@EXAMPLE.COM";
};

Kerberos 主体中的域名必须始终以大写形式。

除了 JAAS 配置外，Kerberos 服务名称还需要在 Kafka 配置的 sasl.kerberos.service.name 属性中指定：

sasl.enabled.mechanisms=GSSAPI
sasl.mechanism.inter.broker.protocol=GSSAPI
sasl.kerberos.service.name=kafka

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    user_admin="123456"
    user_user1="123456"
    user_user2="123456";

    com.sun.security.auth.module.Krb5LoginModule required
    useKeyTab=true
    storeKey=true
    keyTab="/etc/security/keytabs/kafka_server.keytab"
    principal="kafka/kafka1.hostname.com@EXAMPLE.COM";

    org.apache.kafka.common.security.scram.ScramLoginModule required;
};

启用多个机制后，客户端将能够选择要使用的机制。

您可以使用 Java 身份验证和授权服务(JAAS)属性或环境变量来配置 OAuth 2.0 设置。

AMQ Streams OAuth 2.0 库使用开头的属性：

要将 Kafka 代理配置为验证在会话启动过程中收到的令牌，推荐的方法是在授权服务器中创建 OAuth 2.0 客户端 定义（配置为 机密 ）并启用以下客户端凭证：

要在 Kafka 集群中使用 OAuth 2.0 身份验证，您可以在 Kafka server.properties 文件中为 Kafka 集群启用 OAuth 身份验证监听程序配置。最低配置是必需的。您还可以配置 TLS 侦听器，其中 TLS 用于代理间通信。

您可以使用以下方法之一配置授权服务器用于令牌验证的代理：

您可以配置 OAUTHBEARER 或 PLAIN 身份验证，或两者。

以下示例显示了应用 全局 监听器配置的最小配置，这意味着代理间通信与应用客户端通过相同的侦听器。

这个示例还显示了特定监听器的 OAuth 2.0 配置，您可以在其中指定 listener.name.LISTENER-NAME.sasl.enabled.mechanisms 而不是 sasl.enabled.mechanisms。LISTENER-NAME 是监听器不区分大小写的名称。在这里，我们为监听程序命名 CLIENT，因此属性名称为 listener.name.client.sasl.enabled.mechanisms。

这个示例使用 OAUTHBEARER 身份验证。

sasl.enabled.mechanisms=OAUTHBEARER 1
listeners=CLIENT://0.0.0.0:9092 2
listener.security.protocol.map=CLIENT:SASL_PLAINTEXT 3
listener.name.client.sasl.enabled.mechanisms=OAUTHBEARER 4
sasl.mechanism.inter.broker.protocol=OAUTHBEARER 5
inter.broker.listener.name=CLIENT 6
listener.name.client.oauthbearer.sasl.server.callback.handler.class=io.strimzi.kafka.oauth.server.JaasServerOauthValidatorCallbackHandler 7
listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \ 8
  oauth.valid.issuer.uri="https://AUTH-SERVER-ADDRESS" \ 9
  oauth.jwks.endpoint.uri="https://AUTH-SERVER-ADDRESS/jwks" \ 10
  oauth.username.claim="preferred_username"  \ 11
  oauth.client.id="kafka-broker" \ 12
  oauth.client.secret="kafka-secret" \ 13
  oauth.token.endpoint.uri="https://AUTH-SERVER-ADDRESS/token" ; 14
listener.name.client.oauthbearer.sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler 15
listener.name.client.oauthbearer.connections.max.reauth.ms=3600000 16

以下示例显示了 TLS 侦听器的最低配置，其中 TLS 用于代理间通信。

listeners=REPLICATION://kafka:9091,CLIENT://kafka:9092 1
listener.security.protocol.map=REPLICATION:SSL,CLIENT:SASL_PLAINTEXT 2
listener.name.client.sasl.enabled.mechanisms=OAUTHBEARER
inter.broker.listener.name=REPLICATION
listener.name.replication.ssl.keystore.password=KEYSTORE-PASSWORD 3
listener.name.replication.ssl.truststore.password=TRUSTSTORE-PASSWORD
listener.name.replication.ssl.keystore.type=JKS
listener.name.replication.ssl.truststore.type=JKS
listener.name.replication.ssl.endpoint.identification.algorithm=HTTPS 4
listener.name.replication.ssl.secure.random.implementation=SHA1PRNG 5
listener.name.replication.ssl.keystore.location=PATH-TO-KEYSTORE 6
listener.name.replication.ssl.truststore.location=PATH-TO-TRUSTSTORE 7
listener.name.replication.ssl.client.auth=required 8
listener.name.client.oauthbearer.sasl.server.callback.handler.class=io.strimzi.kafka.oauth.server.JaasServerOauthValidatorCallbackHandler
listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  oauth.valid.issuer.uri="https://AUTH-SERVER-ADDRESS" \
  oauth.jwks.endpoint.uri="https://AUTH-SERVER-ADDRESS/jwks" \
  oauth.username.claim="preferred_username" ; 9

以下示例显示了使用 PLAIN 身份验证机制通过 SASL 进行凭证交换的 OAuth 2.0 身份验证的最小配置。使用快速本地令牌验证。

listeners=CLIENT://0.0.0.0:9092 1
listener.security.protocol.map=CLIENT:SASL_PLAINTEXT 2
listener.name.client.sasl.enabled.mechanisms=OAUTHBEARER,PLAIN 3
sasl.mechanism.inter.broker.protocol=OAUTHBEARER 4
inter.broker.listener.name=CLIENT 5
listener.name.client.oauthbearer.sasl.server.callback.handler.class=io.strimzi.kafka.oauth.server.JaasServerOauthValidatorCallbackHandler 6
listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \ 7
  oauth.valid.issuer.uri="http://AUTH_SERVER/auth/realms/REALM" \ 8
  oauth.jwks.endpoint.uri="https://AUTH_SERVER/auth/realms/REALM/protocol/openid-connect/certs" \ 9
  oauth.username.claim="preferred_username"  \ 10
  oauth.client.id="kafka-broker" \ 11
  oauth.client.secret="kafka-secret" \ 12
  oauth.token.endpoint.uri="https://AUTH-SERVER-ADDRESS/token" ; 13
listener.name.client.oauthbearer.sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler 14
listener.name.client.plain.sasl.server.callback.handler.class=io.strimzi.kafka.oauth.server.plain.JaasServerOauthOverPlainValidatorCallbackHandler 15
listener.name.client.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ 16
  oauth.valid.issuer.uri="https://AUTH-SERVER-ADDRESS" \ 17
  oauth.jwks.endpoint.uri="https://AUTH-SERVER-ADDRESS/jwks" \ 18
  oauth.username.claim="preferred_username"  \ 19
  oauth.token.endpoint.uri="http://AUTH_SERVER/auth/realms/REALM/protocol/openid-connect/token" ; 20
connections.max.reauth.ms=3600000 21

快速本地 JWT 令牌验证在本地检查 JWT 令牌签名。

本地检查可确保令牌：

在配置监听器时，您可以指定 一个有效的签发者 URI，以便拒绝授权服务器未发布的任何令牌。

在快速本地 JWT 令牌验证期间，不需要联系授权服务器。您可以通过指定 OAuth 2.0 授权服务器公开的 JWKs 端点 URI 来 激活快速本地 JWT 令牌验证。端点包含用于验证签名 JWT 令牌的公钥，这些令牌由 Kafka 客户端作为凭据发送。

对于 TLS 侦听器，您可以配置证书 信任存储 并指向信任存储文件。

listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  oauth.valid.issuer.uri="https://AUTH-SERVER-ADDRESS" \ 1
  oauth.jwks.endpoint.uri="https://AUTH-SERVER-ADDRESS/jwks" \ 2
  oauth.jwks.refresh.seconds="300" \ 3
  oauth.jwks.refresh.min.pause.seconds="1" \ 4
  oauth.jwks.expiry.seconds="360" \ 5
  oauth.username.claim="preferred_username" \ 6
  oauth.ssl.truststore.location="PATH-TO-TRUSTSTORE-P12-FILE" \ 7
  oauth.ssl.truststore.password="TRUSTSTORE-PASSWORD" \ 8
  oauth.ssl.truststore.type="PKCS12" ; 9

使用 OAuth 2.0 内省端点进行令牌验证，将收到的访问令牌视为不透明。Kafka 代理将访问令牌发送到内省端点，该端点使用验证所需的令牌信息进行响应。更重要的是，如果特定访问令牌有效，它将返回最新信息，以及关于令牌何时到期的信息。

要配置基于 OAuth 2.0 内省的验证，您需要指定一个 内省端点 URI，而不是为快速本地 JWT 令牌验证指定的 JWKs 端点 URI。根据授权服务器，您通常必须指定 客户端 ID 和客户端 机密，因为内省端点通常受到保护。

listener.name.client.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  oauth.introspection.endpoint.uri="https://AUTH-SERVER-ADDRESS/introspection" \ 1
  oauth.client.id="kafka-broker" \ 2
  oauth.client.secret="kafka-broker-secret" \ 3
  oauth.ssl.truststore.location="PATH-TO-TRUSTSTORE-P12-FILE" \ 4
  oauth.ssl.truststore.password="TRUSTSTORE-PASSWORD" \ 5
  oauth.ssl.truststore.type="PKCS12" \ 6
  oauth.username.claim="preferred_username" ; 7

您可以在 Kafka 会话验证过程中看到用于不同配置的 Kafka 客户端和代理的通信流。

这个步骤描述了如何将 Red Hat Single Sign-On 部署为授权服务器，并将其配置为与 AMQ Streams 集成。

授权服务器为身份验证和授权以及用户、客户端和权限的管理提供中央点。红帽单点登录有一个域概念，其中 域 代表一组单独的用户、客户端、权限和其他配置。您可以使用默认的 master 域，或者创建一个新域。每个域公开自己的 OAuth 2.0 端点，这意味着应用客户端和应用服务器都需要使用相同的域。

要将 OAuth 2.0 与 AMQ Streams 搭配使用，您需要部署授权服务器才能创建和管理身份验证域。

有关安装和管理说明，请参阅：

对于 Red Hat Single Sign-On 部署：

这个步骤描述了如何配置 Kafka 代理，以便代理监听程序被启用为使用授权服务器使用 OAuth 2.0 身份验证。

我们建议通过配置 TLS 侦听器在加密接口上使用 OAuth 2.0。不建议纯监听器。

使用支持所选授权服务器的属性以及您要实施的授权类型来配置 Kafka 代理。

有关监听程序配置中使用的属性的描述，请参阅：

这个步骤描述了如何配置 Kafka 制作者和使用者 API，以使用 OAuth 2.0 与 Kafka 代理交互。

将客户端回调插件添加到 pom.xml 文件，并配置系统属性。

Red Hat Single Sign-On 授权程序( KeycloakRBACAuthorizer)由 AMQ Streams 提供。为了能够将 Red Hat Single Sign-On REST 端点用于 Red Hat Single Sign-On 提供的授权服务，您可以在 Kafka 代理上配置自定义授权程序。

授权者根据需要从授权服务器获取授予权限的列表，并在 Kafka Broker 上强制本地实施授权，从而对每个客户端请求做出快速的授权决策。

基本配置将包括以下属性来识别代理并提供安全访问：

您可以在 配置 Kafka 中找到更多有关基本配置选项的详细信息。

典型的代理配置还包括与主题、线程和日志相关的属性设置。

# ...
num.partitions=1
default.replication.factor=3
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
num.network.threads=3
num.io.threads=8
num.recovery.threads.per.data.dir=1
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
group.initial.rebalance.delay.ms=0
zookeeper.connection.timeout.ms=6000
# ...

基本主题属性为主题设置默认分区和复制因子，它们将应用到没有明确设置这些属性的主题，包括在自动创建主题时。

# ...
num.partitions=1
auto.create.topics.enable=false
default.replication.factor=3
min.insync.replicas=2
replica.fetch.max.bytes=1048576
# ...

auto.create.topics.enable 属性默认启用，以便生产者和消费者需要时自动创建不存在的主题。如果您使用自动创建主题，可以使用 num.partitions 为主题设置默认分区数。但通常禁用此属性，以便通过显式创建主题来提供对主题的更多控制

对于高可用性环境，建议将主题的复制因素增加到至少 3，并将所需同步副本的最小数量设置为比复制因素少 1 个。

对于 数据持久性，您应该在主题配置中设置 min.insync.replicas，并在制作者配置中使用 acks=all 设置消息发送确认。

使用 replica.fetch.max.bytes 设置复制领导分区的每个后续程序获取的信息的最大大小（以字节为单位）。根据平均消息大小和吞吐量更改此值。在考虑读/写缓冲所需的总内存分配时，可用内存还必须能够适应所有跟随者乘以时的最大复制消息大小。大小必须大于 message.max.bytes，以便复制所有信息。

delete.topic.enable 属性默认启用，以允许删除主题。在生产环境中，您应该禁用此属性以避免意外删除主题，从而导致数据丢失。但是，您可以临时启用它并删除主题，然后再次禁用它。

# ...
auto.create.topics.enable=false
delete.topic.enable=true
# ...

如果您使用事务来 启用从生产者到分区的原子写入，事务状态会存储在内部 __transaction_state 主题中。默认情况下，代理被配置为有 3 个复制因素，这个主题最少需要 2 个 in-sync 副本，这意味着 Kafka 集群中至少需要三个代理。

# ...
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2
# ...

同样，存储消费者状态的内部 __consumer_offsets 主题对分区数量和复制因素具有默认设置。

# ...
offsets.topic.num.partitions=50
offsets.topic.replication.factor=3
# ...

不要在生产环境中减少这些设置。您可以 在生产环境中 增加设置。作为例外，您可能要减少单broker 测试 环境中的设置。

网络线程处理对 Kafka 集群的请求，如从客户端应用程序生成和获取请求。将生成请求放入请求队列中。将响应放入响应队列中。

网络线程数量应当反映复制因素以及与 Kafka 集群交互的客户端生产者和消费者的活动级别。如果您要有大量请求，您可以增加线程数量，使用时间线程闲置来确定何时添加更多线程。

要减少拥塞并规范请求流量，您可以在阻止网络线程前限制请求队列中允许的请求数。

I/O 线程从请求队列获取请求来处理它们。添加更多线程可提高吞吐量，但 CPU 内核数和磁盘带宽会施加一个实际的上限。至少，I/O 线程数量应当等于存储卷的数量。

# ...
num.network.threads=3 1
queued.max.requests=500 2
num.io.threads=8 3
num.recovery.threads.per.data.dir=1 4
# ...

所有代理的线程池的配置更新可能会在集群级别动态进行。这些更新限制为当前大小的一半和当前大小的两倍。

如果线程因为磁盘数量慢或受限，您可以尝试增加网络请求的缓冲大小以提高吞吐量：

# ...
replica.socket.receive.buffer.bytes=65536
# ...

另外，增加 bytes Kafka 可接收的最大字节数：

# ...
socket.request.max.bytes=104857600
# ...

Kafka 批量数据，以便在从 Kafka 到客户端的高延迟连接中实现合理的吞吐量，如数据中心之间的连接。但是，如果存在高延迟问题，您可以增加用于发送和接收消息的缓冲区的大小。

# ...
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
# ...

您可以使用 带宽延迟（以字节/秒为单位）估算缓冲区的最大带宽，以估算保持最大 吞吐量所需的缓冲区的大小（以字节/为单位）。

Kafka 使用日志来存储消息数据。日志是与各种索引关联的一系列片段。新消息写入 活动 段，随后从不修改。服务片段从消费者获取请求时读取。活动片段定期被 滚动 为只读，并创建一个新的活跃网段来替换它。次只有一个部分处于活动状态。旧片段会保留，直到它们符合删除条件。

代理级别的配置设置日志片段的最大大小，以及推出活跃片段前以毫秒为单位的最大时间：

# ...
log.segment.bytes=1073741824
log.roll.ms=604800000
# ...

您可以使用 segment.bytes 和 segment.ms 在主题级别覆盖这些设置。您是否需要降低还是提升这些值取决于片段删除策略。较大的大小意味着活跃片段包含更多信息，且不频繁地推出。段也不再频繁地符合删除条件。

您可以设置基于时间或基于大小的日志保留和清理策略，以便保持日志可被管理。根据您的要求，您可以使用日志保留配置来删除旧片段。如果使用日志保留策略，在达到保留限制时会删除非活跃日志片段。删除旧片段会绑定日志所需的存储空间，因此您不会超过磁盘容量。

对于基于时间的日志保留，您需要根据小时、分钟和毫秒设置保留期限。保留周期基于附加至网段的时间信息。

毫秒配置具有超过分钟的优先级，其优先级高于小时。默认情况下，分钟和毫秒配置为空，但这三个选项提供了对您要保留的数据的大量控制。应优先使用毫秒配置，因为它是唯一可动态更新的三个属性之一。

# ...
log.retention.ms=1680000
# ...

如果 log.retention.ms 设置为 -1，则不会对日志保留应用任何时间限制，因此所有日志都会保留。磁盘使用量应始终受到监控，但通常不建议 -1 设置，因为它可能会导致完整磁盘出现问题，而这可能难以修复。

对于基于大小的日志保留，您需要以字节为单位设置最大日志大小（日志中所有片段）：

# ...
log.retention.bytes=1073741824
# ...

换句话说，当日志达到稳定状态后，通常大约有 log.retention.bytes/log.segment.bytes 片段。达到最大日志大小时，会删除旧的网段。

使用最大日志大小时潜在的问题在于它没有考虑消息附加到片段中的时间。您可以将基于时间和大小的日志保留用于清理策略，以获得所需的平衡。达到阈值时首先触发清理。

如果要在从系统中删除片段文件前添加时间延迟，您可以为代理级别的所有主题使用 log.segment.delete.delay.ms 或 file.delete.delay.ms 为主题配置中的特定主题添加延迟。

# ...
log.segment.delete.delay.ms=60000
# ...

删除较旧日志数据的方法由 日志清理 配置决定。

默认情况下，为代理启用日志清理功能：

# ...
log.cleaner.enable=true
# ...

您可以在主题或代理级别设置清理策略。代理级配置是尚未设置策略的主题的默认设置。

您可以将策略设置为删除日志、紧凑日志，或同时执行这两者：

# ...
log.cleanup.policy=compact,delete
# ...

delete 策略与使用数据保留策略管理日志对应。当不需要永久保留数据时，它非常适合。compact 策略保证为每个消息键保留最新的消息。日志紧凑适合更改消息值，您想要保留最新的更新。

如果将清理策略设置为删除日志，则会根据日志保留限制删除旧的片段。否则，如果没有启用日志清理功能，且没有日志保留限制，日志将继续增加。

如果为日志紧凑设置了清理策略，日志 的头 将作为一个标准 Kafka 日志运行，按顺序为新消息写入操作。在紧凑日志的 尾部 （日志清理操作），如果日志中稍后出现具有相同密钥的另一条记录，则将删除记录。带有 null 值的消息也会被删除。如果不使用键，就无法使用紧凑，因为需要键来识别相关消息。虽然 Kafka 保证每个密钥的最新消息将被保留，但它不能保证整个紧凑的日志不会包含重复项。

图 6.1. 在压缩前以偏移位置显示键值写入的日志

使用密钥来识别信息，Kafka 紧凑会保留特定消息键的最新消息（带有最高偏移值），最终丢弃之前具有相同密钥的消息。换句话说，其最新状态的消息始终可用，当日志安全运行时，该特定消息的任何过期记录最终都会被删除。您可以将消息恢复到以前的状态。

即使记录被删除，记录也会保留原始偏移。因此，尾部可能有非连续偏移。当消耗在尾部中不再可用的偏移时，会找到具有下一个高偏移量的记录。

图 6.2. 紧凑后日志

如果您只选择紧凑的策略，您的日志仍然可以变得任意大。在这种情况下，您可以将策略设置为压缩 和删除 日志。如果您选择压缩和删除，则首先压缩日志数据，从而删除日志头带有键的记录。之后，在删除日志保留阈值之前发布的数据。

图 6.3. 日志保留点和压缩点

您需要设置以毫秒为单位检查清理的频率：

# ...
log.retention.check.interval.ms=300000
# ...

调整与日志保留设置相关的日志保留检查间隔。较小的保留大小可能需要更频繁地检查。

清理的频率应该足以管理磁盘空间，但经常影响某个主题的性能。

您还可以以毫秒为单位设置一个时间，以便在没有要清理的日志时将清理干净到备用中：

# ...
log.cleaner.backoff.ms=15000
# ...

如果您选择删除旧的日志数据，您可以在清除数据前设置一个毫秒来保留删除的数据：

# ...
log.cleaner.delete.retention.ms=86400000
# ...

删除的数据保留期限给予了时间，可以注意到数据在不可删除之前已丢失。

若要删除与特定密钥相关的所有消息，制作者可以发送 tomb stone 消息。am bstone 具有 null 值，充当标记来告知消费者删除其值。紧凑后，只保留了 tombstone，它必须持续足够长的时间，使消费者能够知道该消息被删除。删除旧消息后，没有值，也会从分区中删除 tombstone 密钥。

还有许多其他与日志清理相关的配置设置，但特别重要的是内存分配。

deduplication 属性指定在所有日志清理线程中清理的总内存。您可以设置通过缓冲区负载因数使用的内存百分比的上限。

# ...
log.cleaner.dedupe.buffer.size=134217728
log.cleaner.io.buffer.load.factor=0.9
# ...

每个日志条目都使用正好 24 字节，因此您可以计算一次缓冲区可以处理的日志条目数量，并相应地调整设置。

如果您希望减少日志清理时间，请考虑增加日志清理线程数量：

# ...
log.cleaner.threads=8
# ...

如果您遇到 100% 磁盘带宽使用情况的问题，您可以限制日志清理 I/O，根据执行操作的磁盘功能，读写操作的总和小于指定的双值：

# ...
log.cleaner.io.max.bytes.per.second=1.7976931348623157E308
# ...

消息的默认批处理大小为 1MB，这是大多数用例中最大吞吐量的最佳选择。如果有足够的磁盘容量，Kafka 能够以更低的吞吐量来容纳较大的批处理。

处理较大的消息的大小有四种方式：

建议使用基于参考的消息和消息压缩选项，并覆盖大多数情况。对于这些选项，必须小心谨慎以避免引入性能问题。

压缩还会增加制作者的额外处理开销和消费者的解压缩开销，但在批量中包含更多数据，因此在消息数据压缩不错时，通常对吞吐量很有用。

将制作者侧压缩与批处理大小的微调相结合，促进最佳吞吐量。使用指标有助于测量所需的平均批处理大小。

图 6.4. 基于参考的消息传递流

由于消息传递需要更多行程，端到端延迟会增加。这种方法的另一个显著缺点是，在清理 Kafka 消息时，没有自动清理外部系统中的数据。混合方法是仅向数据存储直接发送大型消息，并直接处理标准大小的消息。

如果消息太大，产生客户端应用必须按顺序进行序列化，然后对数据进行块。然后，制作者使用 Kafka ByteArraySerializer 或者类似地再次对每个块进行序列化，然后再发送它。使用者跟踪消息和缓冲块，直到消息完整消息为止。消耗的客户端应用程序接收区块，这些区块在反序列化之前组装。根据每组区块消息的第一块或最后一个块的偏移，将完整消息传送到正在使用的应用的其余部分。针对偏移元数据检查成功发送完整消息，以避免在重新平衡期间重复。

图 6.5. 内联消息传递流

由于需要缓冲，内联消息传递在消费者一侧具有性能开销，特别是在并行处理一系列大型消息时。大型消息的块可能会相互交集，因此，如果缓冲区中另一条大消息的块不完整，则始终无法在消息的所有块被使用时提交。因此，缓冲区通常由持久消息区块或实施提交逻辑来支持。

代理将拒绝任何大于 message.max.bytes 设置的限制信息。生产者(max.request.size)和使用者(message.max.bytes)的缓冲大小必须能够容纳较大的信息。

日志刷新属性控制缓存的消息数据的定期写入磁盘。调度程序以毫秒为单位指定日志缓存中的检查频率：

# ...
log.flush.scheduler.interval.ms=2000
# ...

您可以根据信息保存内存的最大时间以及日志中的信息在写入磁盘前的最大信息数来控制刷新的频率：

# ...
log.flush.interval.ms=50000
log.flush.interval.messages=100000
# ...

清空之间的等待包括检查的时间和执行清除前的指定间隔。增加清空的频率可能会影响吞吐量。

通常，建议不要设置明确的清空阈值，并让操作系统使用其默认设置执行后台刷新。分区复制比写入单个磁盘的数据持久性要高，因为故障代理可以从其同步副本中恢复。

如果您使用应用程序清空管理，如果使用更快的磁盘，则可能最好设置较低的清空阈值。

分区可以在代理之间复制，以实现容错。对于给定的分区，一个代理被选为领导，并处理所有生成的请求（写入到日志）。在出现领导故障时，其他代理的分区跟踪者复制分区领导者的分区数据，以实现数据可靠性。

跟随者通常不为客户端服务，但 broker.rack 允许使用者在 Kafka 集群跨越多个数据中心时消耗来自最接近的副本的消息。跟随者只从分区领导机复制消息，并在领导机失败时允许恢复。恢复需要一个同步跟踪器。跟随者通过向领导发送获取请求来保持同步，后者按顺序向跟随者返回消息。如果跟踪者发现了领导上最近提交的邮件，则被视为同步。领导机通过查看跟随者请求的最后偏移来检查这一点。除非允许未清理的领导选举，否则无法同步跟进者作为当前的领导机 故障。

您可以调整跟随者被视为不同步前的滞后时间：

# ...
replica.lag.time.max.ms=30000
# ...

滞后时间为将消息复制到所有同步副本的时间施加了上限，以及生产者必须等待确认的时间。如果跟踪器无法发出获取请求并在指定滞后时间内跟上最新的消息，则会从同步副本中删除它。您可以缩短检测失败副本的滞后时间，但这样做可能会增加无用同步的后续者的数量。正确的滞后时间值取决于网络延迟和代理磁盘带宽。

当领导分区不再可用时，会选择同步内的副本之一作为新领导。分区副本列表中的第一个代理称为 首选 的领导。默认情况下，Kafka 会根据对领导发行的定期检查来为自动分区领导重新平衡启用 Kafka。也就是说，Kafka 检查首选的领导是否是 当前的 领导机。重新平衡可确保领先者在代理和代理之间均匀分布，不会过载。

您可以使用 AMQ Streams 的 Cruise Control 来找出副本分配，以便代理在整个集群中平均平衡负载。其计算考虑了领导者和追随者所经历的不同负载。失败的领导机会影响 Kafka 集群的平衡，因为剩余的代理会获得前导额外分区的额外工作。

要使 Cruise Control 发现的分配真正达到平衡，分区必须要由首选领导领导。Kafka 可以自动确保使用首选的领导机（在可能的情况下），根据需要更改当前的领导机。这样可确保集群保持 Cruise Control 找到的均衡状态。

您可以在触发重新平衡前，以秒为单位控制重新平衡检查的频率和代理允许的最大发生率。

#...
auto.leader.rebalance.enable=true
leader.imbalance.check.interval.seconds=300
leader.imbalance.per.broker.percentage=10
#...

代理的领导地位百分比是代理作为当前领导者的分区数量与首选领导者分区数量之间的比例。您可以将百分比设置为零，以确保始终选中首选领导者，假设他们同步。

如果重新平衡的检查需要更多控制，您可以禁用自动重新平衡。然后，您可以选择何时使用 kafka-leader-election.sh 命令行工具触发重新平衡。

同步副本的领导选举被视为干净，因为它保证不会丢失数据。这是默认情况下会发生的情况。但是，如果没有可以承担领导地位的同步副本，情况会怎样？或许 ISR（同步副本）仅在领导磁盘终止时包含领导机。如果没有设置最少的同步副本数，且其硬盘驱动器出现故障时没有跟随分区领导器同步，则数据已经丢失。不仅 不能这种情况，而且不能选择新的领导，因为不存在同步的追随者。

您可以配置 Kafka 如何处理领导故障：

# ...
unclean.leader.election.enable=false
# ...

Unclean leader 选举默认被禁用，这意味着不同步的副本不能成为领导。使用干净的领导选举机制时，如果没有其他代理在旧领导丢失时处于 ISR 中，Kafka 会等待该领导者重新在线，然后再写入或读取消息。unclean leader 选举机制意味着不同步的副本可能会成为领导者，但您会面临丢失消息的风险。您选择的选择取决于您的需求是否有利于可用性或持久性。

您可以覆盖主题级别上特定主题的默认配置。如果无法承担数据丢失的风险，请保留默认配置。

对于加入新消费者组的消费者，您可以添加延迟，以避免不必要的重新平衡到代理：

# ...
group.initial.rebalance.delay.ms=3000
# ...

延迟是协调会等待会员加入的时间。延迟时间越长，所有成员越有可能加入并避免重新平衡。但是，这一延迟也会阻止该组在句点结束前被消耗掉。

每个制作者都需要连接和序列化程序属性。通常而言，最好是添加客户端 ID 以进行跟踪，并对制作者使用压缩来减少请求中的批处理大小。

在基本制作者配置中：

# ...
bootstrap.servers=localhost:9092 1
key.serializer=org.apache.kafka.common.serialization.StringSerializer 2
value.serializer=org.apache.kafka.common.serialization.StringSerializer 3
client.id=my-client 4
compression.type=gzip 5
# ...

您可以使用消息发送确认，应用更大的数据持久性，以最大程度降低消息丢失的可能性。

# ...
acks=all 1
# ...

在将确认发送到生产者之前，需要将信息附加到日志的代理数量由主题的 min.insync.replicas 配置决定。典型的起点是将主题复制因数为 3，其他代理上有两个内联副本。在这种配置中，如果单个代理不可用，生产者可以继续不受影响。如果第二个代理不可用，生产者将不会收到确认并且无法生成更多消息。

# ...
min.insync.replicas=2 1
# ...

幂等制作者避免重复，因为消息只发送一次。为消息分配了 ID 和序列号，以确保传送顺序，即使出现故障也是如此。如果您使用 acks=all 以实现数据一致性，启用幂等性对有序交付有利。

# ...
enable.idempotence=true 1
max.in.flight.requests.per.connection=5 2
acks=all 3
retries=2147483647 4
# ...

如果您没有因为性能成本而使用 acks=all 和幂等性，请将提示内（未确认）请求数设置为 1 来保持排序。否则，只有在 Message- B 已写入代理后 Message- A 可能无法成功。

# ...
enable.idempotence=false 1
max.in.flight.requests.per.connection=1 2
retries=2147483647
# ...

仅对写入单个分区一次，Idempotence 非常有用。事务处理与幂等性一起使用时，允许在多个分区间只写入一次。

事务可确保使用相同事务 ID 的消息只生成一次，并且将 所有 消息都成功写入到对应的日志中，或者其中任何消息 都不是.

# ...
enable.idempotence=true
max.in.flight.requests.per.connection=5
acks=all
retries=2147483647
transactional.id=UNIQUE-ID 1
transaction.timeout.ms=900000 2
# ...

选择 transactional.id 非常重要，以便保持事务保证。每个事务 id 都应该用于一组唯一的主题分区。例如，这可以通过外部映射主题分区名称到事务 id 来实现，或者通过使用避免冲突的功能计算主题分区名称中的事务 ID。

通常，系统的要求是满足给定延迟内一定比例消息的特定吞吐量目标。例如，以每秒 500,000条消息为目标，95% 的消息会在 2 秒内得到确认。

生产者的消息传递语义（消息排序和持久性）很可能根据您的应用程序的要求进行定义。例如，您可以选择在不破坏某些重要属性的情况下使用 acks=0 或 acks=1 或者应用程序提供保证。

Broker 重新启动对高百分比统计数据有显著影响。例如，在很长一段时间内，99 百分点延迟由围绕代理重启的行为占据主导地位。在设计基准测试时，或比较基准测试与生产中显示的性能数字时，需要考虑这一点。

根据您的目标，Kafka 提供了多个配置参数和技术来调节吞吐量和延迟的性能。

send() 方法返回的时间点。send() 被阻塞的时间由以下方法决定：

批处理将保留在队列中，直到出现以下情况之一：

查看批处理和缓冲的配置，以减轻 send() 阻塞对延迟的影响。

# ...
linger.ms=100 1
batch.size=16384 2
buffer.memory=33554432 3
# ...

您还可以通过编写自定义分区程序来替换默认分区，将消息定向到指定分区。

# ...
delivery.timeout.ms=120000 1
partitioner.class=my-custom-partitioner 2

# ...

每个消费者都需要连接和反序列化器属性。通常，最好添加客户端 ID 进行跟踪。

在使用者配置中，无论后续配置如何：

# ...
bootstrap.servers=localhost:9092 1
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer  2
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer  3
client.id=my-client 4
group.id=my-group-id 5
# ...

消费者组共享一个由一个或多个生产者从给定主题生成的大型数据流。用户使用 group.id 属性分组，允许信息分散到成员中。组中的一个消费者被选为领导，并决定如何将该分区分配给组中的使用者。每个分区只能分配给一个消费者。

如果您还没有多少消费者作为分区，可以通过添加相同 group.id 的更多消费者实例来扩展数据消耗。将更多的消费者添加到组中，超过现有分区，但这意味着如果一个分区停止工作，就表示待机使用者处于待机状态。如果您能够以更少的消费者达到吞吐量目标，就可以节省资源。

同一消费者组中的消费者发送偏移提交和心跳到同一代理。因此组中的消费者数量越多，代理上的请求负载就越大。

# ...
group.id=my-group-id 1
# ...

Kafka 代理从客户接收请求，要求代理从主题、分区和偏移位置列表中发送消息。

用户会按照提交至代理的顺序以单个分区中观察消息，这意味着 Kafka 仅在 单一分区中为消息提供排序保证。相反，如果消费者使用来自多个分区的消息，则使用者观察到的不同分区中消息的顺序不一定反映它们的发送顺序。

如果您需要严格排序一个主题的消息，请为每个使用者使用一个分区。

控制客户端应用调用 KafkaConsumer.poll() 时返回的消息数量。

使用 fetch.max.wait.ms 和 fetch.min.bytes 属性增加消费者从 Kafka 代理获取的最小数据量。使用 fetch.max.wait.ms 配置基于时间的批处理，并使用 fetch.min.bytes 配置基于大小的批处理。

如果消费者或代理中的 CPU 使用率较高，则可能是因为消费者的请求太多。您可以调整 fetch.max.wait.ms 和 fetch.min.bytes 属性，以便在更大的批处理中发送的请求和信息较少。通过调整较高的吞吐量，可以降低延迟成本。如果产生的数据量较低，您也可以调整更高的值。

例如：如果您将 fetch.max.wait.ms 设置为 500ms，fetch.min.bytes 设置为 16384 字节，当 Kafka 收到来自消费者的获取请求时，它会在达到任一阈值的第一个阈值时响应。

相反，您可以降低 fetch.max.wait.ms 和 fetch.min.bytes 属性调整，以改进端到端延迟。

# ...
fetch.max.wait.ms=500 1
fetch.min.bytes=16384 2
# ...

fetch.max.bytes 属性设置一次从代理获取的数据量上限，以字节为单位。

max.partition.fetch.bytes 为每个分区返回的数据量以字节为单位设置的最大限制，必须始终大于代理或 max.message.bytes 主题配置中设置的字节数。

客户端可消耗的最大内存量计算如下：

NUMBER-OF-BROKERS * fetch.max.bytes and NUMBER-OF-PARTITIONS * max.partition.fetch.bytes

如果内存用量可以容纳它，您可以增加这两个属性的值。通过在每个请求中允许更多数据，可以提高延迟，因为获取请求的数量较少。

# ...
fetch.max.bytes=52428800 1
max.partition.fetch.bytes=1048576 2
# ...

Kafka 自动提交机制 允许使用者自动提交消息偏移。如果启用，消费者将以 5000ms 间隔提交从轮询代理收到的偏移。

自动提交机制很方便，但会带来数据丢失和复制风险。如果使用者已获取并转换了大量消息，但执行自动提交时，系统会对消费者缓冲区中已处理的消息崩溃，该数据将会丢失。如果系统在处理消息后崩溃，但在执行自动使用前，数据会在重新平衡后在另一个消费者实例上重复。

仅当在下一次轮询到代理或消费者关闭之前处理所有消息时，自动使用才可以避免数据丢失。

要最小化数据丢失或重复的可能性，您可以将 enable.auto.commit 设置为 false，并开发您的客户端应用程序，使其对提交的偏移拥有更多控制权。或者您可以使用 auto.commit.interval.ms 减少提交之间的间隔。

# ...
enable.auto.commit=false 1
# ...

通过将 enable.auto.commit 设置为 false，您可以在执行了 所有 处理并且消息已被使用后提交偏移。例如，您可以将应用程序设置为调用 Kafka commitSync 和 commitAsync 提交 API。

commitSync API 在从轮询返回的消息批处理中提交偏移。完成批处理中的所有消息后，您将调用 API。如果您使用 commitSync API，在提交批处理中的最后一个偏移前，应用不会轮询新信息。如果这会对吞吐量造成负面影响，您可以降低提交的频率，也可以使用 commitAsync API。commitAsync API 不等待代理响应提交请求，但会在重新平衡时造成更多的重复。种常见的做法是将应用中的两个提交 API 与刚刚在关闭使用者或重新平衡前使用的 commitSync API 合并，以确保最终提交成功。

# ...
enable.auto.commit=false
isolation.level=read_committed 1
# ...

使用 session.timeout.ms 和 heartbeat.interval.ms 属性配置检查并恢复消费者组中的消费者故障的时间。

session.timeout.ms 属性指定在被视为不活跃并在组中活跃消费者之间触发 重新平衡 前，消费者组中的最大时间（毫秒）可以无法与代理联系。当组重新平衡时，这些分区将重新分配给组的成员。

heartbeat.interval.ms 属性指定心跳与消费者组协调器之间的 心跳 之间的间隔，以指示消费者活跃并连接。heartbeat 间隔必须小于会话超时间隔，通常为第三个。

如果您将 session.timeout.ms 属性设置为较低，则之前检测到失败的使用者，重新平衡可以更快地进行。但是，请不要设置超时时间，以便代理无法及时收到心跳，并触发不必要的重新平衡。

减少心跳间隔降低了意外重新平衡的可能性，但更频繁的心跳会增加对代理资源的开销。

使用 auto.offset.reset 属性控制当没有提交偏移时消费者的行为方式，或者提交偏移无效或删除。

假设您第一次部署使用者应用，并且它从现有主题读取消息。因为这是第一次使用 group.id，所以 __consumer_offsets 主题不包含这个应用程序的任何偏移信息。新应用可以从日志开始时开始处理所有现有消息，或者仅处理新消息。默认重置值为 latest，它从分区末尾开始，因此表示丢失了一些信息。要避免数据丢失，但增加处理量，将 auto.offset.reset 设置为 earliest 以从分区开头开始。

另请考虑使用 earliest 选项避免在为代理配置的偏移保留周期(offsets.retention.minutes)终止时丢失消息。如果消费者组或独立消费者不活跃，且在保留周期内未提交偏移，则之前提交的偏移会从 __consumer_offsets 中删除。

# ...
heartbeat.interval.ms=3000 1
session.timeout.ms=10000 2
auto.offset.reset=earliest 3
# ...

如果单个获取请求中返回的数据量较大，则使用者处理数据之前可能会发生超时。在这种情况下，您可以降低 max.partition.fetch.bytes 或增加 session.timeout.ms。

在组中活跃使用者之间重新平衡分区是以下时间：

显然，这个过程会增加服务的停机时间，特别是在客户组群集滚动重启期间重复发生时。

在这种情况下，您可以使用 静态成员资格 的概念来减少重新平衡的数量。重新平衡使用者组成员之间均匀分配主题分区。静态成员资格使用持久性，以便在会话超时后在重启期间识别使用者实例。

用户组协调可以使用使用 group.instance.id 属性指定的唯一 ID 来识别新的消费者实例。在重启期间，会为消费者分配一个新成员 ID，但作为静态成员，它将继续使用相同的实例 ID，并分配相同的主题分区。

如果消费者应用程序没有调用来至少轮询每个 max.poll.interval.ms 毫秒，则消费者会被视为失败，从而导致重新平衡。如果应用无法及时处理轮询返回的所有记录，您可以使用 max.poll.interval.ms 属性为消费者的新消息指定轮询间隔，以避免重新平衡。或者您可以使用 max.poll.records 属性设置从消费者缓冲区返回的记录数上限，允许您的应用程序处理较少的 max.poll.interval.ms 限制中的记录。

# ...
group.instance.id=_UNIQUE-ID_ 1
max.poll.interval.ms=300000 2
max.poll.records=500 3
# ...

提高某个主题吞吐量的主要方法是增加该主题的分区数量。这是因为分区允许在集群中的代理之间共享该主题的负载。当代理都受到某些资源（通常是 I/O）的限制时，使用更多分区将不会产生吞吐量的增长。相反，您必须在集群中添加代理。

当您在集群中添加额外的代理时，AMQ Streams 不会自动为其分配任何分区。您必须决定哪些分区需要从现有代理迁移到新代理。

在所有代理之间重新分布分区后，每个代理的资源使用率应较低。

在从集群中删除代理前，您必须确定它没有被分配给任何分区。您应该决定剩余的代理将负责代理的每个要停用的分区。代理没有分配分区后，就可以停止它。

重新分配 JSON 文件 有一个特定的结构：

{
  "version": 1,
  "partitions": [
    <PartitionObjects>
  ]
}

其中 <PartitionObjects> 是一个用逗号分开的对象列表，例如：

{
  "topic": <TopicName>,
  "partition": <Partition>,
  "replicas": [ <AssignedBrokerIds> ],
  "log_dirs": [<LogDirs>]
}

"log_dirs" 属性是可选的，用于将分区移到特定的日志目录中。

以下是重新分配 JSON 文件的示例，该文件分配主题 topic-a，分区 4 分配给代理 2、4 和 7，以及主题 topic-b 分区 2 分配给代理 1、5 和 7 ：

{
  "version": 1,
  "partitions": [
    {
      "topic": "topic-a",
      "partition": 4,
      "replicas": [2,4,7]
    },
    {
      "topic": "topic-b",
      "partition": 2,
      "replicas": [1,5,7]
    }
  ]
}

JSON 中没有包括的分区不会更改。

将给定主题集的所有分区分配给给定代理集合的最简单方法是使用 kafka-reassign-partitions.sh --generate 命令生成重新分配 JSON 文件。

bin/kafka-reassign-partitions.sh --zookeeper <ZooKeeper> --topics-to-move-json-file <TopicsFile> --broker-list <BrokerList> --generate

<TopicsFile> 是一个 JSON 文件，列出了要移动的主题。它具有以下结构：

{
  "version": 1,
  "topics": [
    <TopicObjects>
  ]
}

其中 <TopicObjects> 是一个以逗号分隔的对象列表，例如：

{
  "topic": <TopicName>
}

例如，将 topic-a 和 topic-b 的所有分区移动到代理 4 和 7

bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --topics-to-move-json-file topics-to-be-moved.json --broker-list 4,7 --generate

其中 topics-to-be-moved.json 包含内容：

{
  "version": 1,
  "topics": [
    { "topic": "topic-a"},
    { "topic": "topic-b"}
  ]
}

如果要移动特定的分区，可以手动创建重新分配 JSON 文件。