Apache Kafka 作为来源进行直播

Apache Kafka 是开源事件流平台，支持数据管道和流分析等工作负载。你可以使用适用于 A pache Kafka 的亚马逊托管流媒体（亚马逊 MSK），也可以使用自我管理的 Apache Kafka 集群。 Amazon 用术语来说，自我管理集群是指未由托管的任何 Apache Kafka 集群。 Amazon这包括您自己管理的集群，以及由第三方提供商（例如Confluent Cloud CloudKarafka、或）托管的集群Redpanda。

有关集群其他 Amazon 托管选项的更多信息，请参阅Amazon 大数据博客上的《运行 Apache Kafka 的最佳实践》。 Amazon

Apache Kafka 作为来源的运作方式与使用亚马逊简单队列服务 (Amazon SQS) 或亚马逊 Kinesis 类似。 EventBridge在内部轮询来自源的新消息，然后同步调用目标。 EventBridge 批量读取消息，并将其作为事件负载提供给您的函数。最大批处理大小可配置。（默认值为 100 个消息。）

对于基于 Apache Kafka 的源， EventBridge 支持处理控制参数，例如批处理窗口和批处理大小。

EventBridge 当它调用你的管道时，会在事件参数中发送一批消息。事件负载包含一个消息数组。每个数组项目都包含 Apache Kafka 主题和 Apache Kafka 分区标识符的详细信息，以及时间戳和 base64 编码的消息。

示例事件

以下示例事件显示了管道接收到的信息。您可以使用此事件来创建和筛选您的事件模式，或定义输入转换。并非所有字段都可以筛选。有关可筛选字段的更多信息，请参阅亚马逊 EventBridge 管道过滤。


[
  {
    "eventSource": "SelfManagedKafka",
    "bootstrapServers": "b-2.demo-cluster-1.a1bcde.c1.kafka.us-east-1.amazonaws.com:9092,b-1.demo-cluster-1.a1bcde.c1.kafka.us-east-1.amazonaws.com:9092",
    "eventSourceKey": "mytopic-0",
    "topic": "mytopic",
    "partition": 0,
    "offset": 15,
    "timestamp": 1545084650987,
    "timestampType": "CREATE_TIME",
    "key":"abcDEFghiJKLmnoPQRstuVWXyz1234==", 
    "value":"SGVsbG8sIHRoaXMgaXMgYSB0ZXN0Lg==",
    "headers": [
      {
        "headerKey": [
          104,
          101,
          97,
          100,
          101,
          114,
          86,
          97,
          108,
          117,
          101
        ]
      }
    ]
  }
]

Apache Kafka 集群身份验证

EventBridge Pipes 支持多种通过自我管理的 Apache Kafka 集群进行身份验证的方法。请确保将 Apache Kafka 集群配置为使用支持的下列身份验证方法之一。有关 Apache Kafka 安全的更多信息，请参阅 Apache Kafka 文档的安全部分。

VPC 访问

如果您使用的是自我管理的 Apache Kafka 环境，只有您的 VPC 中的 Apache Kafka 用户才能访问您的 Apache Kafka 代理，则必须在 Apache Kafka 源中配置亚马逊虚拟私有云（亚马逊 VPC）。

SASL/SCRAM 身份验证

EventBridge Pipes 支持带有传输层安全 (TLS) 加密的简单身份验证和安全层/加盐质询响应身份验证机制 (SASL/SCRAM) 身份验证。 EventBridge Pipes 发送加密的凭据以向集群进行身份验证。有关 SASL/SCRAM 身份验证的更多信息，请参阅 RFC 5802。

EventBridge 管道支持采用 TLS 加密的 SASL/PLAIN 身份验证。使用 SASL/PLAIN 身份验证，Pi EventBridge pes 会将凭据以明文（未加密）的形式发送到服务器。

为了 SASL 身份验证，需要将登录凭证作为密钥存储在 Amazon Secrets Manager中。

双向 TLS 身份验证

双向 TLS（mTLS）在客户端和服务器之间提供双向身份验证。客户端向服务器发送证书以便服务器验证客户端，而服务器又向客户端发送证书以便客户端验证服务器。

在自我管理的 Apache Kafka 中，Pip EventBridge es 充当客户端。您可以配置客户端证书（作为 Secrets Manager 中的密钥）来向 Apache Kafka 代理对 Pip EventBridge es 进行身份验证。客户端证书必须由服务器信任存储中的证书颁发机构 (CA) 签名。

Apache Kafka 集群向 Pipes 发送服务器证书，用于使用 Pip EventBridge es 对 Apache Kafka 代理进行身份验证。 EventBridge 服务器证书可以是公有 CA 证书。也可以是私有 CA/自签名证书。公有 CA 证书必须由 Pip EventBridge es 信任存储区中的 CA 签名。对于私有 CA/自签名证书，您可以配置服务器根 CA 证书（作为 Secrets Manager 中的密钥）。 EventBridge Pipes 使用根证书来验证 Apache Kafka 代理。

有关 mTLS 的更多信息，请参阅为作为源的 Amazon MSK 引入双向 TLS 身份验证。

配置客户端证书密钥

CLIENT_CERTIFICATE_TLS_AUTH 密钥需要证书字段和私有密钥字段。对于加密的私有密钥，密钥需要私有密钥密码。证书和私有密钥必须采用 PEM 格式。

注意

EventBridge Pipes 支持 PBES1（但不是 PBES2）私钥加密算法。

证书字段必须包含证书列表，首先是客户端证书，然后是任何中间证书，最后是根证书。每个证书都必须按照以下结构在新行中启动：


-----BEGIN CERTIFICATE-----  
        <certificate contents>
-----END CERTIFICATE-----

Secrets Manager 支持最多包含 65536 字节的密钥，这为长证书链提供了充足的空间。

私有密钥必须采用 PKCS #8 格式，并具有以下结构：


-----BEGIN PRIVATE KEY-----  
         <private key contents>
-----END PRIVATE KEY-----

对于加密的私有密钥，请使用以下结构：


-----BEGIN ENCRYPTED PRIVATE KEY-----  
          <private key contents>
-----END ENCRYPTED PRIVATE KEY-----

以下示例显示使用加密私有密钥进行 mTLS 身份验证的密钥内容。对于加密的私有密钥，可以在密钥中包含私有密钥密码。


{
 "privateKeyPassword": "testpassword",
 "certificate": "-----BEGIN CERTIFICATE-----
MIIE5DCCAsygAwIBAgIRAPJdwaFaNRrytHBto0j5BA0wDQYJKoZIhvcNAQELBQAw
...
j0Lh4/+1HfgyE2KlmII36dg4IMzNjAFEBZiCRoPimO40s1cRqtFHXoal0QQbIlxk
cmUuiAii9R0=
-----END CERTIFICATE-----
-----BEGIN CERTIFICATE-----
MIIFgjCCA2qgAwIBAgIQdjNZd6uFf9hbNC5RdfmHrzANBgkqhkiG9w0BAQsFADBb
...
rQoiowbbk5wXCheYSANQIfTZ6weQTgiCHCCbuuMKNVS95FkXm0vqVD/YpXKwA/no
c8PH3PSoAaRwMMgOSA2ALJvbRz8mpg==
-----END CERTIFICATE-----",
 "privateKey": "-----BEGIN ENCRYPTED PRIVATE KEY-----
MIIFKzBVBgkqhkiG9w0BBQ0wSDAnBgkqhkiG9w0BBQwwGgQUiAFcK5hT/X7Kjmgp
...
QrSekqF+kWzmB6nAfSzgO9IaoAaytLvNgGTckWeUkWn/V0Ck+LdGUXzAC4RxZnoQ
zp2mwJn2NYB7AZ7+imp0azDZb+8YG2aUCiyqb6PnnA==
-----END ENCRYPTED PRIVATE KEY-----"
}

配置服务器根 CA 证书密钥

如果您的 Apache Kafka 代理使用 TLS 加密（具有由私有 CA 签名的证书），则创建此密钥。您可以将 TLS 加密用于 VPC、SASL/SCRAM、SASL/PLAIN 或 mTLS 身份验证。

服务器根 CA 证书密钥需要一个字段，其中包含 PEM 格式的 Apache Kafka 代理的根 CA 证书。以下示例显示密钥的结构。


{
     "certificate": "-----BEGIN CERTIFICATE-----       
  MIID7zCCAtegAwIBAgIBADANBgkqhkiG9w0BAQsFADCBmDELMAkGA1UEBhMCVVMx
  EDAOBgNVBAgTB0FyaXpvbmExEzARBgNVBAcTClNjb3R0c2RhbGUxJTAjBgNVBAoT
  HFN0YXJmaWVsZCBUZWNobm9sb2dpZXMsIEluYy4xOzA5BgNVBAMTMlN0YXJmaWVs
  ZCBTZXJ2aWNlcyBSb290IENlcnRpZmljYXRlIEF1dG...
  -----END CERTIFICATE-----"

网络配置

如果您使用的是使用私有 VPC 连接的自我管理的 Apache Kafka 环境，则 EventBridge 必须有权访问与您的 Apache Kafka 代理关联的亚马逊虚拟私有云（亚马逊 VPC）资源。要访问您的 Apache Kafka 集群的 VPC， EventBridge 需要您的源子网的出站互联网访问权限。对于公有子网，它必须是托管 NAT 网关。对于私有子网，它可以是 NAT 网关，也可以是您自己的 NAT。确保此 NAT 具有公共 IP 地址，可以连接到互联网。

使用以下规则配置您的 Amazon VPC 安全组（至少）：

入站规则 - 允许为源指定之安全组的 Apache Kafka 代理端口（9092 对应纯文本，9094 对应 TLS，9096 对应 SASL，9098 对应 IAM）上的所有流量。
出站规则 – 允许所有目标的端口 443 上的所有流量传输。允许为源指定之安全组的 Apache Kafka 代理端口（9092 对应纯文本，9094 对应 TLS，9096 对应 SASL，9098 对应 IAM）上的所有流量。

使用 Apache Kafka 来源进行消费者自动扩展

最初创建 Apache Kafka 源时， EventBridge 会分配一个使用者来处理 Kafka 主题中的所有分区。每个使用者都使用多个并行运行的处理器来处理增加的工作负载。此外，还可以根据工作负载 EventBridge 自动增加或缩小使用者的数量。要保留每个分区中的消息顺序，使用者的最大数量为主题中每个分区一个使用者。

每隔一分钟， EventBridge 评估主题中所有分区的消费偏移延迟。如果延迟太高，则表示分区接收消息的速度 EventBridge 快于处理消息的速度。如有必要，在主题中 EventBridge 添加或删除消费者。增加或移除使用者的扩缩过程会在评估完成后的三分钟内进行。

如果您的目标超负荷，则 EventBridge 减少消费者的数量。此操作通过减少使用者可以检索和发送到函数的消息数来减少函数的工作负载。

Javascript 在您的浏览器中被禁用或不可用。

要使用 Amazon Web Services 文档，必须启用 Javascript。请参阅浏览器的帮助页面以了解相关说明。

Amazon MSK 主题

Amazon SQS 队列