开发 Kafka 客户端应用程序

在 RHEL 上使用 AMQ Streams 时，保护与 Kafka 集群的客户端连接的一般概述如下：

有关设置代理的更多信息，请参阅在 RHEL 上使用 AMQ Streams。

使用配置属性文件在 Kafka 中配置监听程序。要为 Kafka 代理配置安全连接，您可以在此文件中为 TLS、SASL 和其他与安全相关的配置设置相关属性。

以下是在 Kafka 代理的 server.properties 配置文件中指定的 TLS 侦听器配置示例，其密钥存储和 truststore 格式为：

listeners = listener_1://0.0.0.0:9093, listener_2://0.0.0.0:9094
listener.security.protocol.map = listener_1:SSL, listener_2:PLAINTEXT
ssl.keystore.type = PKCS12
ssl.keystore.location = /path/to/keystore.p12
ssl.keystore.password = <password>
ssl.truststore.type = PKCS12
ssl.truststore.location = /path/to/truststore.p12
ssl.truststore.password = <password>
ssl.client.auth = required
authorizer.class.name = kafka.security.auth.SimpleAclAuthorizer.
super.users = User:superuser

listener 属性指定每个监听程序名称，以及代理侦听的 IP 地址和端口。protocol 映射告知 listener_1 侦听器将 SSL 协议用于使用 TLS 加密的客户端。listener_2 为不使用 TLS 加密的客户端提供 PLAINTEXT 连接。密钥存储包含代理的私钥和证书。truststore 包含用于验证客户端应用程序身份的可信证书。ssl.client.auth 属性强制执行客户端身份验证。

Kafka 集群使用简单的授权。授权器设置为 SimpleAclAuthorizer。定义了单个超级用户，在所有监听器上没有限制访问。AMQ Streams 支持 Kafka SimpleAclAuthorizer 和自定义授权器插件。

如果我们使用 listener.name.<name_of_listener> 为配置属性添加前缀，则配置特定于该监听程序。

这只是一个示例配置。某些配置选项特定于监听器类型。如果您使用 OAuth 2.0 或 Open Policy Agent (OPA)，还必须在特定监听器中配置对授权服务器或 OPA 服务器的访问。您可以根据特定要求和环境创建监听程序。

有关监听程序配置的更多信息，请参阅 Apache Kafka 文档。

在以下示例中，第一个 ACL 授予名为 my-topic 的特定主题的读和描述权限。resource.patternType 设置为 literal，这意味着资源名称必须完全匹配。

第二个 ACL 为名为 my-group 的特定消费者组授予读取权限。resource.patternType 设置为 prefix，这意味着资源名称必须与前缀匹配。

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add \
--allow-principal User:my-user --operation Read --operation Describe --topic my-topic --resource-pattern-type literal \
--allow-principal User:my-user --operation Read --group my-group --resource-pattern-type prefixed

在 OpenShift 上使用 AMQ Streams 时，保护与 Kafka 集群的客户端连接的一般概述如下：

有关设置代理的更多信息，请参阅在 OpenShift 中配置 AMQ Streams。

当您使用 AMQ Streams 部署 Kafka 自定义资源时，您可以在 Kafka spec 中添加监听程序配置。使用监听程序配置来保护 Kafka 中的连接。要为 Kafka 代理配置安全连接，请在监听器级别上为 TLS、SASL 和其他与安全相关的配置设置相关属性。

外部监听程序提供对 OpenShift 集群外部的 Kafka 集群的客户端访问。AMQ Streams 创建监听器服务和 bootstrap 地址，以根据配置启用对 Kafka 集群的访问。例如，您可以创建使用以下连接机制的外部监听程序：

以下是 Kafka 资源的 nodeport 侦听器配置示例：

apiVersion: {KafkaApiVersion}
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    listeners:
      - name: plaintext
        port: 9092
        type: internal
        tls: false
        configuration:
          useServiceDnsDomain: true
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls
      - name: external
        port: 9094
        type: route
        tls: true
        authentication:
          type: tls
    authorization:
      type: simple
      superUsers:
        - CN=superuser
  # ...

监听器 属性配置了三个监听程序： 纯文本、tls 和 external。外部监听程序 是 nodeport 类型，它使用 TLS 进行加密和验证。当使用 Cluster Operator 创建 Kafka 集群时，会自动生成 CA 证书。您可以将集群 CA 添加到客户端应用程序的信任存储中，以验证 Kafka 代理的身份。另外，您可以将 AMQ Streams 配置为使用代理或监听程序级别的自己的证书。当客户端应用程序需要不同的安全配置时，可能需要在监听器级别使用证书。在监听器级别使用证书也会添加额外的控制层和安全性。

Kafka 集群使用简单的授权。authorization 属性类型设置为 simple。定义了单个超级用户，在所有监听器上没有限制访问。AMQ Streams 支持 Kafka SimpleAclAuthorizer 和自定义授权器插件。

这只是一个示例配置。某些配置选项特定于监听器类型。如果您使用 OAuth 2.0 或 Open Policy Agent (OPA)，还必须在特定监听器中配置对授权服务器或 OPA 服务器的访问。您可以根据特定要求和环境创建监听程序。

如需有关监听器配置的更多信息，请参阅 GenericKafkaListener 模式参考。

在以下示例中，第一个 ACL 授予名为 my-topic 的特定主题的读和描述权限。resource.patternType 设置为 literal，这意味着资源名称必须完全匹配。

第二个 ACL 为名为 my-group 的特定消费者组授予读取权限。resource.patternType 设置为 prefix，这意味着资源名称必须与前缀匹配。

apiVersion: {KafkaUserApiVersion}
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  # ...
  authorization:
    type: simple
    acls:
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operations:
          - Read
          - Describe
      - resource:
          type: group
          name: my-group
          patternType: prefix
        operations:
          - Read

配置客户端应用程序使用的安全协议，以匹配 Kafka 代理监听程序中配置的协议。例如，对 TLS 身份验证使用 SSL （安全套接字层），或使用 TLS 加密对 SASL (Simple Authentication and Security Layer)验证 SASL_SSL。在您的客户端配置中添加 truststore 和 keystore，它支持访问 Kafka 集群所需的身份验证机制。

如果使用 TLS 身份验证，您的 Kafka 客户端配置需要一个信任存储和密钥存储来连接到 Kafka 集群。如果您使用 SASL SCRAM-SHA-512，则通过交换用户名和密码凭证而不是数字证书来执行身份验证，因此不需要密钥存储。SCRAM-SHA-512 是一种更加轻量级的机制，但它不像使用基于证书的身份验证一样安全。

您可以创建一个 config.properties 文件来指定客户端应用程序使用的身份验证凭证。

在以下示例中，security.protocol 设置为 SSL，以启用客户端和代理之间的 TLS 身份验证和加密。

ssl.truststore.location 和 ssl.truststore.password 属性指定信任存储的位置和密码。ssl.keystore.location 和 ssl.keystore.password 属性指定密钥存储的位置和密码。

使用 PKCS thePublic-Key Cryptography Standards 文件格式。您还可以使用 base64 编码的 PEM (Privacy Enhanced Mail)格式。

bootstrap.servers = my-cluster-kafka-bootstrap:9093
security.protocol = SSL
ssl.truststore.location = /path/to/ca.p12
ssl.truststore.password = truststore-password
ssl.keystore.location = /path/to/user.p12
ssl.keystore.password = keystore-password
client.id = my-client

在以下示例中，security.protocol 设置为 SASL_SSL，以便在客户端和服务器之间使用 TLS 加密启用 SASL 身份验证。如果您只需要身份验证而不是加密，您可以使用 SASL 协议。进行身份验证的 SASL 机制是 SCRAM-SHA-512。可以使用不同的身份验证机制。sasl.jaas.config 属性指定身份验证凭据。

bootstrap.servers = my-cluster-kafka-bootstrap:9093
security.protocol = SASL_SSL
sasl.mechanism = SCRAM-SHA-512
sasl.jaas.config = org.apache.kafka.common.security.scram.ScramLoginModule required \
  username = "user" \
  password = "secret";
ssl.truststore.location = path/to/truststore.p12
ssl.truststore.password = truststore_password
ssl.truststore.type = PKCS12
client.id = my-client

您可以组合 SSL 配置和密码套件，以进一步保护客户端应用程序和 Kafka 集群之间的基于 TLS 的通信。在 Kafka 代理配置中指定支持的 TLS 版本和密码套件。如果要限制它们使用的 TLS 版本和密码套件，您还可以将配置添加到客户端。客户端上的配置应该只使用代理上启用的协议和密码套件。

在以下示例中，SSL 使用 security.protocol 进行 Kafka 代理和客户端应用程序之间的通信。您可以将密码套件指定为用逗号分开的列表。ssl.cipher.suites 属性是客户端允许使用的密码套件的逗号分隔列表。

security.protocol: "SSL"
ssl.enabled.protocols: "TLSv1.3", "TLSv1.2"
ssl.protocol: "TLSv1.3"
ssl.cipher.suites: "TLS_AES_256_GCM_SHA384"

ssl.enabled.protocols 属性指定可用于集群及其客户端之间安全通信的可用 TLS 版本。在本例中，TLSv1.3 和 TLSv1.2 都已启用。ssl.protocol 属性为所有连接设置默认 TLS 版本，且必须从启用的协议中选择它。默认情况下，客户端使用 TLSv1.3 进行通信。如果客户端只支持 TLSv1.2，它仍然可以连接到代理并使用该支持的版本进行通信。同样，如果配置位于客户端，且代理只支持 TLSv1.2，客户端将使用支持的版本。

Apache Kafka 支持的密码套件取决于您使用的 Kafka 版本和底层环境。检查提供最高安全性级别的最新支持的密码套件。

您不必为客户端应用程序中的 ACLS 明确配置任何内容。ACL 由 Kafka 代理在服务器端强制执行。当客户端向服务器发送请求来生成或消耗数据时，服务器会检查 ACL，以确定客户端（用户）是否被授权执行所请求的操作。如果客户端被授权，则会处理请求；否则请求将被拒绝，并返回错误。但是，客户端仍必须经过身份验证，并使用适当的安全协议启用与 Kafka 集群的安全连接。

如果您在 Kafka 代理上使用访问控制列表(ACL)，请确保正确设置了 ACL，以限制对您要控制的主题和操作的访问。如果您使用 Open Policy Agent (OPA)策略来管理访问，在策略中配置授权规则，因此您不需要为 Kafka 代理指定 ACL。OAuth 2.0 提供了一些灵活性：您可以使用 OAuth 2.0 供应商来管理 ACL；或使用 OAuth 2.0 和 Kafka 的简单 授权来管理 ACL。

使用 OAuth 2.0 开放标准进行 AMQ Streams 授权，通过 OAuth 2.0 供应商强制控制授权。OAuth 2.0 为应用提供了一种安全的方法来访问存储在其他系统中的用户数据。授权服务器可能会向客户端应用程序发出访问令牌，以授予 Kafka 集群的访问权限。

以下步骤描述了设置和使用 OAuth 2.0 进行令牌验证的一般方法：

如果您在 Kafka 代理上配置了 OAuth 2.0 的监听程序，您可以将客户端应用程序设置为使用 OAuth 2.0。除了用于访问 Kafka 集群的标准 Kafka 客户端配置外，还必须为 OAuth 2.0 身份验证包含特定的配置。您还必须确保 Kafka 集群和客户端应用程序可以访问您要使用的授权服务器。

指定 SASL （简单身份验证和安全层）安全协议和机制。在生产环境中，推荐进行以下设置：

JAAS (Java 身份验证和授权服务)模块实施 SASL 机制。机制的配置取决于您使用的身份验证方法。例如，使用凭证交换您添加 OAuth 2.0 访问令牌端点、访问令牌、客户端 ID 和客户端 secret。客户端连接到授权服务器的令牌端点(URL)，以检查令牌是否仍然有效。您还需要一个 truststore，其中包含授权服务器的公钥证书，以便进行身份验证。

bootstrap.servers = my-cluster-kafka-bootstrap:9093
security.protocol = SASL_SSL
sasl.mechanism = OAUTHBEARER
# ...
sasl.jaas.config = org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
    oauth.token.endpoint.uri = "https://localhost:9443/oauth2/token" \
    oauth.access.token = <access_token> \
    oauth.client.id = "<client_id>" \
    oauth.client.secret = "<client_secret>" \
    oauth.ssl.truststore.location = "/<truststore_location>/oauth-truststore.p12" \
    oauth.ssl.truststore.password = "<truststore_password>" \
    oauth.ssl.truststore.type = "PKCS12" \

有关将代理设置为使用 OAuth 2.0 的更多信息，请参阅以下指南：

使用带有 AMQ Streams 的 Open Policy Agent (OPA)策略控制器来评估您的 Kafka 集群的请求，具体取决于访问策略。开放策略代理(OPA)是一种策略引擎，用于管理授权策略。策略集中访问控制，可以动态更新，无需更改客户端应用程序。例如，您可以创建一个策略，仅允许某些用户（客户端）为特定主题生成和使用消息。

AMQ Streams 为 Kafka 授权使用 Open Policy Agent 插件作为授权器。

以下步骤描述了设置和使用 OPA 的一般方法：

如果您在 Kafka 代理上配置了 OPA 的监听程序，您可以将客户端应用程序设置为使用 OPA。在监听器配置中，您可以指定一个 URL 来连接到 OPA 服务器并授权您的客户端应用程序。除了访问 Kafka 集群的标准 Kafka 客户端配置外，还必须添加凭证以使用 Kafka 代理进行身份验证。代理通过向 OPA 服务器发送请求来评估授权策略，以检查客户端是否有必要的授权来执行请求操作。您不需要信任存储或密钥存储来保护通信，因为策略引擎强制执行授权策略。

bootstrap.servers = my-cluster-kafka-bootstrap:9093
security.protocol = SASL_SSL
sasl.mechanism = SCRAM-SHA-512
sasl.jaas.config = org.apache.kafka.common.security.scram.ScramLoginModule required \
  username = "user" \
  password = "secret";
# ...

有关将代理设置为使用 OPA 的更多信息，请参阅以下指南：

通过在代理和制作者客户端应用程序中配置事务属性，您可以确保在单个事务中处理消息。事务向消息流增加可靠性和一致性。

代理始终启用事务。您可以使用以下属性更改默认配置：

transaction.state.log.replication.factor = 3
transaction.state.log.min.isr = 2
transaction.abort.timed.out.transaction.cleanup.interval.ms = 3600000

这是生产环境的典型配置，它为内部 __transaction_state 主题创建 3 个副本。\__transaction_state 主题存储有关事务在进行中的信息。事务日志至少需要 2 个同步副本。清理间隔是检查超时事务和清理对应的事务日志之间的时间。

要为客户端配置添加事务属性，您需要为生产者和消费者设置以下属性：

transactional.id = unique-transactional-id
enable.idempotence = true
max.in.flight.requests.per.connection = 5
acks = all
transaction.timeout.ms = 30000
delivery.timeout = 25000

事务 ID 允许 Kafka 代理跟踪事务。它是制作者的唯一标识符，应当与特定的分区一起使用。如果您需要为多组分区执行事务，则需要为每个集合使用不同的事务 ID。启用 idempotence 以避免生成者实例创建重复消息。使用 idempotence 时，消息使用制作者 ID 和序列号进行跟踪。代理收到消息时，它会检查制作者 ID 和序列号。如果收到具有相同制作者 ID 和序列号的消息，代理会丢弃重复的消息。

最大动态请求数设置为 5，以便以发送的顺序处理事务。分区最多可有 5 个动态请求，而不影响消息排序。

通过将 acks 设为 all，生产者会等待来自要写入的主题分区的所有同步副本的确认，然后再将事务视为完成。这样可确保信息被大量写入（提交）到 Kafka 集群，即使代理失败，也不会丢失它们。

事务超时指定客户端在超时前必须完成事务的最长时间。交付超时指定制作者在超时前等待代理确认消息发送的最长时间。要确保在事务周期内发送消息，请将交付超时设置为小于事务超时。在指定重试次数时，请考虑网络延迟和消息吞吐量，并允许临时故障。

group.id = my-group-id
isolation.level = read_committed
enable.auto.commit = false

read_committed 隔离级别指定消费者只读取成功完成的事务的消息。消费者不会处理属于持续或失败的事务的任何消息。这样可确保消费者只读取属于完整事务的消息。

当使用事务流消息时，务必要将 enable.auto.commit 设置为 false。如果设置为 true，则消费者会定期提交偏移量，而无需考虑事务。这意味着消费者可以在事务完成前提交消息。通过将 enable.auto.commit 设置为 false，使用者仅读取并提交已完全写入并提交至该主题的消息，作为事务的一部分。