在 RHEL 中使用 AMQ Streams

流程

1. 在 FIPS 模式中安装 RHEL。

   如需更多信息，请参阅 RHEL 文档中的 安全强化的信息。

2. 继续安装 AMQ Streams。

3. 将 AMQ Streams 配置为使用 FIPS 兼容算法和协议。

   如果使用，请确保以下配置兼容：

   • JDK 框架必须支持 SSL 密码套件和 TLS 版本。
   • SCRAM-SHA-512 密码必须至少为 32 个字符。

1. 添加新的 kafka 用户和组：

   groupadd kafka
   useradd -g kafka kafka
   passwd kafka

2. 将 amq-streams-<version>-bin.zip 文件的内容提取到 /opt/kafka 目录中：

   unzip amq-streams-<version>-bin.zip -d /opt
   mv /opt/kafka*redhat* /opt/kafka

   此步骤需要管理员特权。

3. 将 /opt/kafka 目录的所有权更改为 kafka 用户：

   chown -R kafka:kafka /opt/kafka

4. 创建用于存储 ZooKeeper 数据的 /var/lib/zookeeper 目录，并将其所有权设置为 kafka 用户：

   mkdir /var/lib/zookeeper
   chown -R kafka:kafka /var/lib/zookeeper

5. 创建用于存储 Kafka 数据的目录 /var/lib/kafka，并将其所有权设置为 kafka 用户：

   mkdir /var/lib/kafka
   chown -R kafka:kafka /var/lib/kafka

   现在 ，您可以作为单节点集群运行 Kafka 的默认配置。

   您还可以使用安装在同一主机上运行其他 Kafka 组件，如 Kafka Connect。

   要运行其他组件，请使用组件配置中的 bootstrap.servers 属性指定要连接到 Kafka 代理的主机名和端口。

   指向同一主机上单个 Kafka 代理的 bootstrap 服务器配置示例

   bootstrap.servers=localhost:9092

   但是，我们建议在单独的主机上安装并运行 Kafka 组件。

6. （可选）在单独的主机上安装 Kafka 组件。

   1. 将安装文件提取到每个主机上的 /opt/kafka 目录中。
   2. 将 /opt/kafka 目录的所有权更改为 kafka 用户。

   3. 添加 bootstrap.servers 配置，将组件连接到运行 Kafka 代理的主机（或多节点环境中的主机）。

      指向不同主机上 Kafka 代理的 bootstrap 服务器配置示例

      bootstrap.servers=kafka0.<host_ip_address>:9092,kafka1.<host_ip_address>:9092,kafka2.<host_ip_address>:9092

      您可以将此配置用于 Kafka Connect、MirrorMaker 2 和 Kafka Bridge。

运行集群

1. 编辑 ZooKeeper 配置文件 /opt/kafka/config/zookeeper.properties。将 dataDir 选项设置为 /var/lib/zookeeper/:

   dataDir=/var/lib/zookeeper/

2. 编辑 Kafka 配置文件 /opt/kafka/config/server.properties。将 log.dirs 选项设置为 /var/lib/kafka/ ：

   log.dirs=/var/lib/kafka/

3. 切换到 kafka 用户：

   su - kafka

4. 启动 ZooKeeper：

   /opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties

5. 检查 ZooKeeper 是否正在运行：

   jcmd | grep zookeeper

   返回：

   number org.apache.zookeeper.server.quorum.QuorumPeerMain /opt/kafka/config/zookeeper.properties

6. 启动 Kafka：

   /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

7. 检查 Kafka 是否正在运行：

   jcmd | grep kafka

   返回：

   number kafka.Kafka /opt/kafka/config/server.properties

流程

1. 启动 Kafka 控制台制作者并将其配置为将信息发送到新主题：

   /opt/kafka/bin/kafka-console-producer.sh --broker-list <bootstrap_address> --topic <topic-name>

   例如：

   /opt/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-topic

2. 在控制台中输入多个信息。按 Enter 将每个消息发送到您的新主题：

   >message 1
   >message 2
   >message 3
   >message 4

   当 Kafka 自动创建新主题时，您可能会收到主题不存在的警告：

   WARN Error while fetching metadata with correlation id 39 :
   {4-3-16-topic1=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)

   在发送进一步消息后，该警告不应重新显示。

3. 在新的终端窗口中，启动 Kafka 控制台消费者，并将其配置为从您的新主题开始读取消息。

   /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server <bootstrap_address> --topic <topic-name> --from-beginning

   例如：

   /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning

   传入的消息显示在消费者控制台中。

4. 切换到制作者控制台并发送其他消息。检查它们是否在使用者控制台中显示。
5. 停止 Kafka 控制台制作者，然后按 Ctrl+C。

流程

1. 停止 Kafka 代理。

   su - kafka
   /opt/kafka/bin/kafka-server-stop.sh

2. 确认 Kafka 代理已停止。

   jcmd | grep kafka

3. 停止 ZooKeeper。

   su - kafka
   /opt/kafka/bin/zookeeper-server-stop.sh

运行集群

1. 在 /var/lib/zookeeper/ 中创建 myid 文件。为第一个 ZooKeeper 节点输入 ID 1，为第二个 ZooKeeper 节点输入 2，以此类推。

   su - kafka
   echo "<NodeID>" > /var/lib/zookeeper/myid

   例如：

   su - kafka
   echo "1" > /var/lib/zookeeper/myid

2. 编辑 ZooKeeper /opt/kafka/config/zookeeper.properties 配置文件：

   • 将选项 dataDir 设置为 /var/lib/zookeeper/。
   • 配置 initLimit 和 syncLimit 选项。
   • 配置 reconfigEnabled 和 standaloneEnabled 选项。

   • 添加所有 ZooKeeper 节点的列表。该列表还应包含当前的节点。

     具有五个成员的 ZooKeeper 集群节点配置示例

     tickTime=2000
     dataDir=/var/lib/zookeeper/
     initLimit=5
     syncLimit=2
     reconfigEnabled=true
     standaloneEnabled=false
     listener.security.protocol.map=PLAINTEXT:PLAINTEXT,REPLICATION:PLAINTEXT
     
     server.1=172.17.0.1:2888:3888:participant;172.17.0.1:2181
     server.2=172.17.0.2:2888:3888:participant;172.17.0.2:2181
     server.3=172.17.0.3:2888:3888:participant;172.17.0.3:2181
     server.4=172.17.0.4:2888:3888:participant;172.17.0.4:2181
     server.5=172.17.0.5:2888:3888:participant;172.17.0.5:2181

3. 使用默认配置文件启动 ZooKeeper。

   su - kafka
   /opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties

4. 验证 ZooKeeper 是否正在运行。

   jcmd | grep zookeeper

   返回：

   number org.apache.zookeeper.server.quorum.QuorumPeerMain /opt/kafka/config/zookeeper.properties

5. 在集群的所有节点上重复此步骤。

6. 使用 ncat 实用程序向每个节点发送 stat 命令，以验证所有节点是否都是集群的成员。

   使用 ncat stat 检查节点状态

   echo stat | ncat localhost 2181

   要使用四个字母单词命令，如 stat，您需要在 zookeeper.properties 中指定 4lw.commands.whitelist prerequisites。

   输出显示节点是 领导机 或后续 。

   ncat 命令的输出示例

   ZooKeeper version: 3.4.13-2d71af4dbe22557fda74f9a9b4309b15a7487f03, built on 06/29/2018 00:39 GMT
   Clients:
    /0:0:0:0:0:0:0:1:59726[0](queued=0,recved=1,sent=0)
   
   Latency min/avg/max: 0/0/0
   Received: 2
   Sent: 1
   Connections: 1
   Outstanding: 0
   Zxid: 0x200000000
   Mode: follower
   Node count: 4

1. 编辑 /opt/kafka/config/server.properties Kafka 配置文件，如下所示：

   • 在第一个代理中，将 broker.id 设置为 0，在第二个代理中，将其设置为 1，以此类推。
   • 在 zookeeper.connect 选项中配置连接到 ZooKeeper 的详情。
   • 配置 Kafka 侦听程序。

   • 设置提交日志应存储在 logs.dir 目录中的目录。

     在这里，我们看到 Kafka 代理的示例配置：

     broker.id=0
     zookeeper.connect=zoo1.my-domain.com:2181,zoo2.my-domain.com:2181,zoo3.my-domain.com:2181
     listeners=REPLICATION://:9091,PLAINTEXT://:9092
     listener.security.protocol.map=PLAINTEXT:PLAINTEXT,REPLICATION:PLAINTEXT
     inter.broker.listener.name=REPLICATION
     log.dirs=/var/lib/kafka

     在典型的安装中，每个 Kafka 代理在相同的硬件上运行，只有 broker.id 配置属性在每个代理配置间有所不同。

2. 使用默认配置文件启动 Kafka 代理。

   su - kafka
   /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

3. 验证 Kafka 代理是否正在运行。

   jcmd | grep Kafka

   返回：

   number kafka.Kafka /opt/kafka/config/server.properties

4. 使用 ncat 实用程序将 dump 命令发送到其中一个 ZooKeeper 节点，以验证所有节点是否为 Kafka 集群的成员。

   使用 ncat dump 检查 ZooKeeper 中注册的所有 Kafka 代理

   echo dump | ncat zoo1.my-domain.com 2181

   要使用包括四个字母的命令，如 dump，需要在 zookeeper.properties 中指定 4lw.commands.whitelist=*。

   输出必须包含您刚才配置和启动的所有 Kafka 代理。

   具有 3 个节点的 Kafka 集群的 ncat 命令的输出示例

   SessionTracker dump:
   org.apache.zookeeper.server.quorum.LearnerSessionTracker@28848ab9
   ephemeral nodes dump:
   Sessions with Ephemerals (3):
   0x20000015dd00000:
           /brokers/ids/1
   0x10000015dc70000:
           /controller
           /brokers/ids/0
   0x10000015dc70001:
           /brokers/ids/2

1. 停止 Kafka 代理：

   /opt/kafka/bin/kafka-server-stop.sh

2. 对在完成后需要重启的代理配置进行任何更改。

   如需更多信息，请参阅以下内容：

   • 配置 Kafka
   • 升级 Kafka 代理和 ZooKeeper

3. 重启 Kafka 代理：

   /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

4. 检查 Kafka 是否正在运行：

   jcmd | grep kafka

   返回：

   number kafka.Kafka /opt/kafka/config/server.properties

5. 使用 ncat 实用程序将 dump 命令发送到其中一个 ZooKeeper 节点，以验证所有节点是否为 Kafka 集群的成员。

   使用 ncat dump 检查 ZooKeeper 中注册的所有 Kafka 代理

   echo dump | ncat zoo1.my-domain.com 2181

   要使用包括四个字母的命令，如 dump，需要在 zookeeper.properties 中指定 4lw.commands.whitelist=*。

   输出必须包含您启动的 Kafka 代理。

   具有 3 个节点的 Kafka 集群的 ncat 命令的输出示例

   SessionTracker dump:
   org.apache.zookeeper.server.quorum.LearnerSessionTracker@28848ab9
   ephemeral nodes dump:
   Sessions with Ephemerals (3):
   0x20000015dd00000:
           /brokers/ids/1
   0x10000015dc70000:
           /controller
           /brokers/ids/0
   0x10000015dc70001:
           /brokers/ids/2

6. 等待直到代理有 0 个重复分区。您可以从命令行检查或使用指标。

   • 使用带有 --under-replicated-partitions 参数的 kafka-topics.sh 命令：

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server <bootstrap_address>  --describe --under-replicated-partitions

     例如：

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --under-replicated-partitions

     命令提供了集群中带有重复分区的主题列表。

     带有复制分区的主题

     Topic: topic3 Partition: 4 Leader: 2 Replicas: 2,3 Isr: 2
     Topic: topic3 Partition: 5 Leader: 3 Replicas: 1,2 Isr: 1
     Topic: topic1 Partition: 1 Leader: 3 Replicas: 1,3 Isr: 3
     # …

     如果 ISR （同步副本）计数小于副本数，则会列出 in-replicated 分区。如果没有返回列表，则没有复制的分区。

   • 使用 UnderReplicatedPartitions 指标：

     kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions

     指标提供副本未捕获的分区计数。您等到计数为零。

     提示

     当主题有一个或多个重复分区时，请使用 Kafka Exporter 创建警报。

流程

1. 使用 kafka-storage 工具为 Kafka 集群生成 ID：

   /opt/kafka/bin/kafka-storage.sh random-uuid

   该命令返回一个 ID。KRaft 模式需要一个集群 ID。

2. 为集群中的每个代理创建一个配置文件。

   您可以在 Kafka 提供的示例上构建该文件。

   1. 将角色指定为 broker、'controller 或 broker, controller

      例如，为双角色指定 process.roles=broker，为 controller 指定。

   2. 从 0 开始，为集群中的每个节点指定唯一的 node.id。

      例如，node.id=1。

   3. 指定一个 controller.quorum.voters 列表，格式为 <node_id>@<hostname:port>。

      例如，controller.quorum.voters=1@localhost:9093.

3. 为 Kafka 集群中的每个节点设置日志目录：

   /opt/kafka/bin/kafka-storage.sh format -t <uuid> -c /opt/kafka/config/kraft/server.properties

   返回：

   Formatting /tmp/kraft-combined-logs

   将 <uuid > 替换为您生成的集群 ID。对集群中的每个节点使用相同的 ID。

   使用您为代理创建的属性文件应用代理配置。

   server.properties 配置文件中指定的默认日志目录位置为 /tmp/kraft-combined-logs。您可以添加以逗号分隔的列表来设置多个日志目录。

4. 启动每个 Kafka 代理。

   /opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/kraft/server.properties

5. 检查 Kafka 是否正在运行：

   jcmd | grep kafka

   返回：

   number kafka.Kafka /opt/kafka/config/kraft/server.properties

流程

1. 将 Environment Variables Configuration Provider JAR 文件添加到 Kafka libs 目录。

2. 在 Kafka 组件的配置属性文件中初始化 Environment Variables Configuration Provider。例如，要初始化 Kafka 的供应商，请将配置添加到 server.properties 文件中。

   配置以启用 Environment Variables Configuration Provider

   config.providers=env
   config.providers.env.class=io.strimzi.kafka.EnvVarConfigProvider

3. 添加配置到属性文件，以从环境变量加载数据。

   配置以从环境变量加载数据

   option=${env:<MY_ENV_VAR_NAME>}

   使用大写或大写环境变量命名约定，如 MY_ENV_VAR_NAME。

4. 保存更改。

5. 重启 Kafka 组件。

   有关在多节点集群中重启代理的详情，请参考 第 4.3 节 “执行 Kafka 代理的安全滚动重启”。

创建主题

1. 使用 kafka-topics.sh 实用程序创建主题并指定以下内容：

   • 在 --bootstrap-server 选项中的 Kafka 代理的主机和端口。
   • 要在 --create 选项中创建的新主题。
   • --topic 选项中的主题名称。
   • --partitions 选项中的分区数量。

   • --replication-factor 选项中的主题复制因素。

     您还可以使用 --config 选项覆盖一些默认主题配置选项。这个选项可多次使用来覆盖不同的选项。

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_address> --create --topic <TopicName> --partitions <NumberOfPartitions> --replication-factor <ReplicationFactor> --config <Option1>=<Value1> --config <Option2>=<Value2>

     创建名为 mytopic的主题的命令示例

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic mytopic --partitions 50 --replication-factor 3 --config cleanup.policy=compact --config min.insync.replicas=2

2. 验证主题是否使用 kafka-topics.sh 是否存在。

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_address> --describe --topic <TopicName>

   描述名为 mytopic的主题的命令示例

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic mytopic

描述主题

1. 使用 kafka-topics.sh 工具描述主题并指定以下内容：

   • 在 --bootstrap-server 选项中的 Kafka 代理的主机和端口。
   • 使用 --describe 选项指定您要描述主题。
   • 主题名称必须在 --topic 选项中指定。

   • 当省略 --topic 选项时，它将描述所有可用的主题。

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_address> --describe --topic <TopicName>

     描述名为 mytopic的主题的命令示例

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic mytopic

     命令列出属于此主题的所有分区和副本。它还列出所有主题配置选项。

修改主题配置

1. 使用 kafka-configs.sh 工具获取当前的配置。

   • 在 --bootstrap-server 选项中指定 Kafka 代理的主机和端口。
   • 将 --entity-type 设置为 topic，将 --entity-name 设置为主题的名称。

   • 使用 --describe 选项获取当前的配置。

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_address> --entity-type topics --entity-name <TopicName> --describe

     获取名为 mytopic的主题配置的命令示例

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name mytopic --describe

2. 使用 kafka-configs.sh 工具更改配置。

   • 在 --bootstrap-server 选项中指定 Kafka 代理的主机和端口。
   • 将 --entity-type 设置为 topic，将 --entity-name 设置为主题的名称。
   • 使用 --alter 选项修改当前配置。

   • 在选项 --add-config 中指定您要添加或更改的选项。

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_address> --entity-type topics --entity-name <TopicName> --alter --add-config <Option>=<Value>

     更改名为 mytopic的主题配置的命令示例

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name mytopic --alter --add-config min.insync.replicas=1

3. 使用 kafka-configs.sh 工具删除现有配置选项。

   • 在 --bootstrap-server 选项中指定 Kafka 代理的主机和端口。
   • 将 --entity-type 设置为 topic，将 --entity-name 设置为主题的名称。
   • 使用 --delete-config 选项删除现有配置选项。

   • 在选项 --remove-config 中指定您要删除的选项。

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_address> --entity-type topics --entity-name <TopicName> --alter --delete-config <Option>

     更改名为 mytopic的主题配置的命令示例

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name mytopic --alter --delete-config min.insync.replicas

删除主题

1. 使用 kafka-topics.sh 工具删除主题。

   • 在 --bootstrap-server 选项中的 Kafka 代理的主机和端口。
   • 使用 --delete 选项指定应删除现有主题。

   • 主题名称必须在 --topic 选项中指定。

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_address> --delete --topic <TopicName>

     创建名为 mytopic的主题的命令示例

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic mytopic

2. 使用 kafka-topics.sh 验证主题是否已删除。

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_address> --list

   列出所有主题的命令示例

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list

流程

1. 在文本编辑器中打开示例属性文件，或创建一个新文件，并编辑该文件使其包含连接信息以及每个 Kafka 集群的复制流程。

   以下示例显示了连接两个集群的配置，即 cluster-1 和 cluster-2 双向。集群名称可通过集群属性 配置。

   MirrorMaker 2 配置示例

   clusters=cluster-1,cluster-2 1
   
   cluster-1.bootstrap.servers=<cluster_name>-kafka-bootstrap-<project_name_one>:443 2
   cluster-1.security.protocol=SSL 3
   cluster-1.ssl.truststore.password=<truststore_name>
   cluster-1.ssl.truststore.location=<path_to_truststore>/truststore.cluster-1.jks_
   cluster-1.ssl.keystore.password=<keystore_name>
   cluster-1.ssl.keystore.location=<path_to_keystore>/user.cluster-1.p12
   
   cluster-2.bootstrap.servers=<cluster_name>-kafka-bootstrap-<project_name_two>:443 4
   cluster-2.security.protocol=SSL 5
   cluster-2.ssl.truststore.password=<truststore_name>
   cluster-2.ssl.truststore.location=<path_to_truststore>/truststore.cluster-2.jks_
   cluster-2.ssl.keystore.password=<keystore_name>
   cluster-2.ssl.keystore.location=<path_to_keystore>/user.cluster-2.p12
   
   cluster-1->cluster-2.enabled=true 6
   cluster-2->cluster-1.enabled=true 7
   cluster-1->cluster-2.topics=.* 8
   cluster-2->cluster-1.topics=topic-1, topic-2 9
   cluster-1->cluster-2.groups=.* 10
   cluster-2->cluster-1.groups=group-1, group-2 11
   
   replication.policy.separator=- 12
   sync.topic.acls.enabled=false 13
   refresh.topics.interval.seconds=60 14
   refresh.groups.interval.seconds=60 15

   1

   每个 Kafka 集群都使用其别名来标识。

   2

   使用 bootstrap address 和端口 443 用于 cluster-1 的连接信息。两个集群都使用端口 443 连接到使用 OpenShift Routes 的 Kafka。

   3

   ssl. 属性定义 cluster-1 的 TLS 配置。

   4

   cluster-2 的连接信息。

   5

   ssl. 属性定义 cluster-2 的 TLS 配置。

   6

   从 cluster-1 启用复制流到 cluster-2。

   7

   从 cluster-2 启用复制流到 cluster-1。

   8

   将所有主题从 cluster-1 复制到 cluster-2。源连接器复制指定的主题。checkpoint 连接器跟踪指定主题的偏移。

   9

   将特定主题从 cluster-2 复制到 cluster-1。

   10

   将所有消费者组从 cluster-1 复制到 cluster-2。checkpoint 连接器复制指定的消费者组。

   11

   将特定消费者组从 cluster-2 复制到 cluster-1。

   12

   定义用于重命名远程主题的分隔符。

   13

   启用后，ACL 将应用到同步主题。默认值为 false。

   14

   检查之间的期间，检查要同步的新主题。

   15

   检查要同步的新消费者组之间的周期。

2. OPTION：如果需要，请添加一个策略来覆盖远程主题的自动重命名。该主题不会用源集群的名称来附加名称，而是保留其原始名称。

   此可选设置用于主动/被动备份和数据迁移。

   replication.policy.class=org.apache.kafka.connect.mirror.IdentityReplicationPolicy

3. OPTION：如果要同步消费者组偏移，请添加配置来启用和管理同步：

   refresh.groups.interval.seconds=60
   sync.group.offsets.enabled=true 1
   sync.group.offsets.interval.seconds=60 2
   emit.checkpoints.interval.seconds=60 3

   1

   用于同步消费者组偏移的可选设置，这对于在主动/被动配置中恢复非常有用。默认不启用同步。

   2

   如果启用了使用者组偏移的同步，您可以调整同步的频率。

   3

   调整检查偏移跟踪的频率。如果您更改了偏移同步的频率，您可能需要调整这些检查的频率。

4. 在目标集群中启动 ZooKeeper 和 Kafka：

   su - kafka
   /opt/kafka/bin/zookeeper-server-start.sh -daemon \
   /opt/kafka/config/zookeeper.properties

   /opt/kafka/bin/kafka-server-start.sh -daemon \
   /opt/kafka/config/server.properties

5. 使用您在属性文件中定义的集群连接配置和复制策略启动 MirrorMaker：

   /opt/kafka/bin/connect-mirror-maker.sh \
   /opt/kafka/config/connect-mirror-maker.properties

   MirrorMaker 在集群之间设置连接。

6. 对于每个目标集群，验证主题是否被复制：

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_address> --list

流程

1. 编辑 MirrorMaker consumer.properties 和 producer.properties 文件，以关闭 MirrorMaker 2 功能。

   例如：

   replication.policy.class=org.apache.kafka.mirror.LegacyReplicationPolicy 1
   
   refresh.topics.enabled=false 2
   refresh.groups.enabled=false
   emit.checkpoints.enabled=false
   emit.heartbeats.enabled=false
   sync.topic.configs.enabled=false
   sync.topic.acls.enabled=false

   1

   模拟之前的 MirrorMaker 版本。

   2

   MirrorMaker 2 禁用了功能，包括内部 检查点 和 心跳 主题

2. 保存更改，并使用您在之前版本的 MirrorMaker 中使用的属性文件重启 MirrorMaker：

   su - kafka /opt/kafka/bin/kafka-mirror-maker.sh \
   --consumer.config /opt/kafka/config/consumer.properties \
   --producer.config /opt/kafka/config/producer.properties \
   --num.streams=2

   consumer 属性提供源集群和 producer 属性的配置，提供目标集群配置。

   MirrorMaker 在集群之间设置连接。

3. 在目标集群中启动 ZooKeeper 和 Kafka：

   su - kafka
   /opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties

   su - kafka
   /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

4. 对于目标集群，验证主题是否被复制：

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_address> --list

流程

1. 切换到 kafka 用户：

   su - kafka

2. 使用 kafka-configs.sh 工具列出代理的所有代理日志记录器：

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_address> --describe --entity-type broker-loggers --entity-name BROKER-ID

   例如，对于代理 0 ：

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type broker-loggers --entity-name 0

   这会返回每个日志记录器的日志记录级别：TRACE, DEBUG, INFO, WARN, ERROR, 或 FATAL。

   例如：

   #...
   kafka.controller.ControllerChannelManager=INFO sensitive=false synonyms={}
   kafka.log.TimeIndex=INFO sensitive=false synonyms={}

3. 更改一个或多个代理日志记录器的日志级别。使用 --alter 和 --add-config 选项，并使用双引号将每个日志记录器及其级别指定为逗号分隔的列表。

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_address> --alter --add-config "LOGGER-ONE=NEW-LEVEL,LOGGER-TWO=NEW-LEVEL" --entity-type broker-loggers --entity-name BROKER-ID

   例如，对于代理 0 ：

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config "kafka.controller.ControllerChannelManager=WARN,kafka.log.TimeIndex=WARN" --entity-type broker-loggers --entity-name 0

   如果成功，则返回：

   Completed updating config for broker: 0.

流程

1. 切换到 kafka 用户：

   su - kafka

2. 检查 connect-log4j.properties 文件中配置的日志记录器的当前日志记录级别：

   $ cat /opt/kafka/config/connect-log4j.properties
   
   # ...
   log4j.rootLogger=INFO, stdout, connectAppender
   # ...
   log4j.logger.org.apache.zookeeper=ERROR
   log4j.logger.org.reflections=ERROR

   使用 curl 命令检查 Kafka Connect API 的 admin/loggers 端点中的日志记录级别：

   curl -s http://localhost:8083/admin/loggers/ | jq
   
   {
     "org.apache.zookeeper": {
       "level": "ERROR"
     },
     "org.reflections": {
       "level": "ERROR"
     },
     "root": {
       "level": "INFO"
     }
   }

   jq 以 JSON 格式打印输出。列表显示标准的 org 和 root 级别的日志程序，以及特定的带有改变的日志级别的日志程序。

   如果您在 Kafka Connect 中为 admin.listeners 配置配置 TLS (Transport Layer Security)身份验证，则 loggers 端点的地址是为 admin.listeners 指定的值，如 https，如 https://localhost:8083。

   您还可以获取特定日志记录器的日志级别：

   curl -s http://localhost:8083/admin/loggers/org.apache.kafka.connect.mirror.MirrorCheckpointConnector | jq
   
   {
     "level": "INFO"
   }

3. 使用 PUT 方法更改日志记录器的日志级别：

   curl -Ss -X PUT -H 'Content-Type: application/json' -d '{"level": "TRACE"}' http://localhost:8083/admin/loggers/root
   
   {
     # ...
   
     "org.reflections": {
       "level": "TRACE"
     },
     "org.reflections.Reflections": {
       "level": "TRACE"
     },
     "root": {
       "level": "TRACE"
     }
   }

   如果您更改了 root 日志程序，则默认使用了 root 日志级别的日志程序的日志级别也会被改变。

1. 在 ZooKeeper 集群中添加服务器，如 第 4.1 节 “运行多节点 ZooKeeper 集群” 所述，然后启动 ZooKeeper。
2. 请注意新服务器的 IP 地址和配置的访问端口。

3. 为服务器启动 zookeeper-shell 会话。从可以访问集群的机器中运行以下命令（如果可以访问集群，这可能是 ZooKeeper 节点或本地机器之一）。

   su - kafka
   /opt/kafka/bin/zookeeper-shell.sh <ip-address>:<zk-port>

4. 在 shell 会话中，运行 ZooKeeper 节点，输入以下行将新服务器作为投票成员添加到仲裁中：

   reconfig -add server.<positive-id> = <address1>:<port1>:<port2>[:role];[<client-port-address>:]<client-port>

   例如：

   reconfig -add server.4=172.17.0.4:2888:3888:participant;172.17.0.4:2181

   其中 <positive-id> 是新服务器 ID 4。

   对于两个端口，<port1> 2888 用于 ZooKeeper 服务器之间的通信，<port2> 3888 用于领导选举机制。

   新配置传播到 ZooKeeper 集群中的其他服务器 ; 新服务器现在是仲裁的完整成员。

1. 登录到在缩减后会保留的其中一个服务器上的 zookeeper-shell（例如，服务器 1）。

   注意

   使用为 ZooKeeper 集群配置的验证机制访问服务器。

2. 删除服务器，如 server 5。

   reconfig -remove 5

3. 取消激活您删除的服务器。

流程

1. 从 红帽客户门户网站 下载最新版本的 Red Hat AMQ Streams Cruise Control 归档。

2. 创建 /opt/cruise-control 目录：

   sudo mkdir /opt/cruise-control

3. 将 Cruise Control ZIP 文件的内容提取到新目录中：

   unzip amq-streams-<version>-cruise-control-bin.zip -d /opt/cruise-control

4. 将 /opt/cruise-control 目录的所有权更改为 kafka 用户：

   sudo chown -R kafka:kafka /opt/cruise-control

1. 停止 Kafka 代理：

   /opt/kafka/bin/kafka-server-stop.sh

2. 在 Kafka 配置文件中 (/opt/kafka/config/server.properties) 配置 Cruise Control Metrics Reporter：

   1. 将 CruiseControlMetricsReporter 类添加到 metric.reporters 配置选项。不要删除任何现有的指标报告器。

      metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter

   2. 添加以下配置选项和值：

      cruise.control.metrics.topic.auto.create=true
      cruise.control.metrics.topic.num.partitions=1
      cruise.control.metrics.topic.replication.factor=1

      这些选项允许 Cruise Control Metrics Reporter 创建 __CruiseControlMetrics 主题，其日志清理策略为 DELETE。如需更多信息，请参阅 Auto-created topics 和 Log cleanup policy for Cruise Control Metrics topic。

3. 如果需要，配置 SSL。

   1. 在 Kafka 配置文件中(/opt/kafka/config/server.properties)通过设置相关的客户端配置属性在 Cruise Control Metrics Reporter 和 Kafka 代理之间配置 SSL。

      Metrics Reporter 接受带有 cruise.control.metrics.reporter 前缀的所有特定于标准制作者的配置属性。例如： cruise.control.metrics.reporter.ssl.truststore.password。

   2. 在 Cruise Control 属性文件中(/opt/cruise-control/config/cruisecontrol.properties)通过设置相关的客户端配置属性在 Kafka 代理和 Cruise Control 服务器之间配置 SSL。

      Cruise Control 从 Kafka 中继承 SSL 客户端属性选项，并将这些属性用于所有 Cruise Control 服务器客户端。

4. 重启 Kafka 代理：

   /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

   有关在多节点集群中重启代理的详情，请参考 第 4.3 节 “执行 Kafka 代理的安全滚动重启”。

5. 对剩余的代理重复步骤 1-5。

流程

1. 编辑 Cruise Control 属性文件 (/opt/cruise-control/config/cruisecontrol.properties)。

2. 配置以下示例配置中显示的属性：

   # The Kafka cluster to control.
   bootstrap.servers=localhost:9092 1
   
   # The replication factor of Kafka metric sample store topic
   sample.store.topic.replication.factor=2 2
   
   # The configuration for the BrokerCapacityConfigFileResolver (supports JBOD, non-JBOD, and heterogeneous CPU core capacities)
   #capacity.config.file=config/capacity.json
   #capacity.config.file=config/capacityCores.json
   capacity.config.file=config/capacityJBOD.json 3
   
   # The list of goals to optimize the Kafka cluster for with pre-computed proposals
   default.goals={List of default optimization goals} 4
   
   # The list of supported goals
   goals={list of main optimization goals} 5
   
   # The list of supported hard goals
   hard.goals={List of hard goals} 6
   
   # How often should the cached proposal be expired and recalculated if necessary
   proposal.expiration.ms=60000 7
   
   # The zookeeper connect of the Kafka cluster
   zookeeper.connect=localhost:2181 8

   1

   Kafka 代理的主机和端口号（始终为端口 9092）。

   2

   Kafka 指标示例存储主题的复制因素。如果要在单节点 Kafka 和 ZooKeeper 集群中评估 Cruise Control，请将此属性设置为 1。对于生产环境，请将此属性设置为 2 或以上。

   3

   为代理资源设置最大容量限制的配置文件。使用适用于 Kafka 部署配置的文件。如需更多信息，请参阅 容量配置。

   4

   使用完全限定域名(FQDN)的以逗号分隔的默认优化目标列表。许多主要优化目标（请参见 5）已设为默认优化目标；如果需要，您可以添加或删除目标。更多信息请参阅 第 14.5 节 “优化目标概述”。

   5

   使用 FQDN 以逗号分隔的主要优化目标列表。要完全排除用于生成优化提议的目标，请将它们从列表中删除。更多信息请参阅 第 14.5 节 “优化目标概述”。

   6

   使用 FQDN 以逗号分隔的硬目标列表。七大主要优化目标已设置为硬目标；如果需要，您可以添加或删除目标。更多信息请参阅 第 14.5 节 “优化目标概述”。

   7

   刷新从默认优化目标生成的缓存的优化建议的时间间隔（以毫秒为单位）。更多信息请参阅 第 14.6 节 “优化提议概述”。

   8

   ZooKeeper 连接的主机和端口号（始终端口 2181）。

3. 启动 Cruise Control 服务器。默认情况下，服务器在端口 9092 上启动；（可选）指定不同的端口。

   cd /opt/cruise-control/
   ./kafka-cruise-control-start.sh config/cruisecontrol.properties <port_number>

4. 要验证 Cruise Control 是否正在运行，请将 GET 请求发送到 Cruise Control 服务器的 /state 端点：

   curl 'http://HOST:PORT/kafkacruisecontrol/state'