輪詢和串流開始位置 MSK叢集驗證網路組態可自訂的取用者群組 ID Amazon MSK 來源的自動擴展

Amazon Managed Streaming for Apache Kafka 主題作為管道中的源 EventBridge

您可以使用 EventBridge 管道從 Amazon Managed Streaming for Apache Kafka（AmazonMSK）主題接收記錄。然後，您可以選擇性地篩選或增強這些記錄，然後再將它們傳送到可用的目的地之一進行處理。設置管道時MSK，您可以選擇 Amazon 特定的設置。 EventBridge 當傳送資料至目的地時，管道會維護訊息代理程式的記錄順序。

Amazon MSK 是一項全受管服務，可用來建置和執行使用 Apache Kafka 處理串流資料的應用程式。Amazon MSK 簡化了運行 Apache 卡夫卡的集群的設置，擴展和管理。使用 AmazonMSK，您可以使用 AWS Identity and Access Management (IAM) 為多個可用區域和安全性設定應用程式。Amazon MSK 支持多個開源版本的卡夫卡。

Amazon MSK 作為源的操作類似於使用 Amazon 簡單隊列服務（AmazonSQS）或 Amazon Kinesis。 EventBridge內部輪詢來自來源的新郵件，然後同步調用目標。 EventBridge 批量讀取消息，並將這些消息作為事件有效負載提供給您的函數。批次大小上限可設定。(預設值為 100 則訊息。)

對於以 Apache Kafka 為基礎的來源， EventBridge 支援處理控制參數，例如批次處理視窗和批次大小。

EventBridge 按順序讀取每個分區的消息。 EventBridge 處理每個批次之後，它會認可該批次中訊息的偏移量。如果管道的目標針對批次中的任何訊息傳回錯誤，請 EventBridge 重試整批訊息，直到處理成功或訊息過期為止。

EventBridge 在呼叫目標時，在事件中傳送訊息批次。事件裝載包含訊息陣列。每個陣列項目都包含 Amazon MSK 主題和分區識別碼的詳細資訊，以及時間戳記和 base64 編碼訊息。

範例事件

下列範例事件顯示管道接收的資訊。您可以使用此事件來建立和篩選事件模式，或定義輸入轉換。並非所有欄位都可以篩選。如需有關所能篩選欄位的詳細資訊，請參閱 Amazon EventBridge 管道中的事件過濾。


[
  {
    "eventSource": "aws:kafka",
    "eventSourceArn": "arn:aws:kafka:sa-east-1:123456789012:cluster/vpc-2priv-2pub/751d2973-a626-431c-9d4e-d7975eb44dd7-2",    
    "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
        ]
      }
    ]
  }
]

輪詢和串流開始位置

請注意，建立和更新管道期間的串流輪詢最終會一致。

在建立管道期間，從串流開始輪詢事件可能需要幾分鐘時間。
在更新管道資源輪訓組態期間，從串流停止並重新開始輪詢事件可能需要幾分鐘時間。

這表示如果您指定 LATEST 當作串流的開始位置，管道可能會在建立或更新期間遺漏事件。若要確保沒有遺漏任何事件，請將串流開始位置指定為 TRIM_HORIZON。

MSK叢集驗證

EventBridge 需要存取 Amazon MSK 叢集、擷取記錄和執行其他任務的權限。Amazon MSK 支援多種控制用戶端對MSK叢集存取的選項。如需此身分驗證方法的詳細資訊，請參閱如何 EventBridge 選擇引導經紀人。

未驗證的存取

我們建議您只使用未經驗證的存取進行開發。只有在叢集停用以角色為IAM基礎的驗證時，未驗證的存取才能運作。

SASL/SCRAM認證

Amazon MSK 支援具有傳輸層安全性 () 加密的簡單身份驗證和安全層SASL/加密挑戰回應身份驗證機制 (/SCRAMTLS) 身份驗證。為了連接 EventBridge 到群集，您可以將身份驗證憑據（登錄憑據）存儲在 AWS Secrets Manager 密鑰中。

如需使用 Secrets Manager 詳細資訊，請參閱《Amazon Managed Streaming for Apache Kafka 開發人員指南》中的AWS Secrets Manager的使用者名稱和密碼身分驗證。

Amazon MSK 不支持SASL/PLAIN身份驗證。

IAM 角色型身分驗證

您可以使IAM用驗證連線至MSK叢集的用戶端身分。如果您的MSK叢集上的IAM驗證處於作用中狀態，且您未提供驗證的密碼，則 EventBridge 會自動預設為使用IAM驗證。若要建立和部署使用IAM者或以角色為基礎的策略，請使用主IAM控台或API。如需詳細資訊，請參IAM閱 Amazon Managed Streaming for Apache Kafka 開發人員指南中的存取控制。

若要允許連線 EventBridge 到MSK叢集、讀取記錄以及執行其他必要動作，請將下列權限新增至管道的執行角色。


{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "kafka-cluster:Connect",
                "kafka-cluster:DescribeGroup",
                "kafka-cluster:AlterGroup",
                "kafka-cluster:DescribeTopic",
                "kafka-cluster:ReadData",
                "kafka-cluster:DescribeClusterDynamicConfiguration"
            ],
            "Resource": [
                "arn:aws:kafka:region:account-id:cluster/cluster-name/cluster-uuid",
                "arn:aws:kafka:region:account-id:topic/cluster-name/cluster-uuid/topic-name",
                "arn:aws:kafka:region:account-id:group/cluster-name/cluster-uuid/consumer-group-id"
            ]
        }
    ]
}

您可以將這些許可的範圍設定為特定叢集、主題和群組。如需詳細資訊，請參閱 Amazon Apache MSK Kafka 受管串流開發人員指南中的 Amazon Kaf ka 動作。

相互TLS認證

相互TLS（mTLS）提供客戶端和服務器之間的雙向身份驗證。用戶端會將憑證傳送至伺服器以供伺服器驗證用戶端，而伺服器會將憑證傳送至用戶端以供用戶端驗證伺服器。

對於 AmazonMSK， EventBridge 充當客戶端。您可以設定用戶端憑證 (在 Secrets Manager 中做為密碼)，以便 EventBridge 與MSK叢集中的代理程式進行驗證。客户憑證必須由伺服器信任存放區中的憑證授權機構 (CA) 簽署。MSK叢集會傳送伺服器憑證 EventBridge 來驗證代理程式 EventBridge。伺服器憑證必須由 AWS 信任存放區中的 CA 簽署。

Amazon MSK 不支援自我簽署的伺服器憑證，因為 Amazon 中的所有代理程式都MSK使用 Amazon 信任服務簽署的公有憑證CAs，這些憑證預設會 EventBridge 信任。

如需有關 Amazon m TLS 的詳細資訊MSK，請參閱 Amazon Managed Streaming for Apache Kafka 開發人員指南中的相互TLS身份驗證。

設定 m TLS 密碼

CLIENT_ CERTIFICATE TLS _ AUTH 密碼需要憑證欄位和私密金鑰欄位。若為加密的私有金鑰，機密需要私有金鑰密碼。憑證和私密金鑰都必須是PEM格式。

注意

EventBridge 支援 PBES1(但不支援PBES2) 私密金鑰加密演算法。

憑證欄位必須包含憑證清單，以用戶端憑證開頭，隨後則是任何中繼憑證，並以根憑證結尾。每個憑證均必須以新的一行開始，結構如下：


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

Secrets Manager 支援高達 65,536 個位元組的機密，此空間足以容納長憑證鏈。

私密金鑰必須為 PKCS#8 格式，並具有下列結構：


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

對於已加密的私有金鑰，請使用下列結構：


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

下列範例顯示使用加密私密金鑰進行 m TLS 驗證的密碼內容。若為加密的私有金鑰，您可以在機密中包含私有金鑰密碼。


{
 "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-----"
}

如何 EventBridge 選擇引導經紀人

EventBridge 根據叢集上可用的驗證方法，以及您是否提供驗證密碼，來選擇啟動程序代理程式。如果您為 m TLS 或SASL/提供密碼SCRAM，則 EventBridge 會自動選擇該驗證方法。如果您未提供密碼，請 EventBridge選擇叢集上作用中的最強式驗證方法。以下是選 EventBridge 擇代理的優先順序，從最強到最弱的身份驗證：

mTLS（為 m 提供的秘密TLS）
SASL/SCRAM（提供的秘密SASL/SCRAM）
SASLIAM（未提供密碼，且IAM身份驗證處於活動狀態）
未驗證 TLS (未提供密碼，且IAM驗證未啟用)
純文字 (未提供秘密，且驗證與未IAM驗證均未TLS啟動)

注意

如果 EventBridge 無法連接到最安全的代理類型，則不會嘗試連接到其他（較弱的）代理類型。如果您想 EventBridge 要選擇較弱的代理程式類型，請停用叢集上所有較強的驗證方法。

網路組態

EventBridge 必須能夠訪問與您的 Amazon MSK 集群關聯的 Amazon Virtual Private Cloud（AmazonVPC）資源。

若要存取 Amazon MSK 叢集， EventBridge 可以對來源VPC的子網路使用對外網際網路存取。對於私有子網路，它可以是NAT閘道，也可以是您自己NAT的閘道。確保NAT具有公共 IP 地址並且可以連接到互聯網。對於公共子網路，您必須使用VPC端點 (如下所述)。
EventBridge Pipes 也支援透過事件傳遞 AWS PrivateLink，可讓您將事件從位於 Amazon Virtual Private Cloud (Amazon VPC) 中的事件來源傳送至 Pipes 目標，而無需遍歷公用網際網路。您可以使用管道從 Amazon Managed Streaming for Apache Kafka (Amazon MSK)、自我管理的 Apache Kafka 以及位於私有子網路中的 Amazon MQ 來源進行輪詢，而無需部署網際網路閘道、設定防火牆規則或設定 Proxy 伺服器。您也可以使用VPC端點支援從公用子網路中的 Kafka 叢集傳遞。

若要設定VPC端點，請參閱AWS PrivateLink 使用指南中的建立VPC端點。對於服務名稱，請選取com.amazonaws.region.pipes-data。

使用下列規則 (至少) 設定 Amazon VPC 安全群組：

輸入規則 — 允許為您的來源指定的安全群組的 MSK Amazon 代理連接埠上的所有流量。
傳出規則：針對所有目的地，允許連接埠 443 上的所有流量。允許針對您的來源指定的安全群組的 MSK Amazon 代理連接埠上的所有流量。

代理連接埠包括：
- 對於明文字而言是 9092
- 為了找到 TLS
- 適用於 SASL
- 適用於 IAM

注意

您的 Amazon VPC 配置可以通過 Amazon MSK API 發現。您不需要在設定期間設定它。

可自訂的取用者群組 ID

將 Kafka 設為來源時，您可以指定取用者群組 ID。此取用者群組 ID 是您希望管道加入之 Kafka 取用者群組的現有識別符。您可以使用此功能將任何正在進行的 Apache Kafka 記錄處理設定從其他消費者遷移到。 EventBridge

如果您指定取用者群組 ID，且該取用者群組內還有其他作用中的輪詢者，則 Kafka 會將訊息分配給所有取用者。換句話說， EventBridge 不會收到有關 Apache 卡夫卡主題的所有消息。如果您想 EventBridge 要處理主題中的所有郵件，請關閉該用戶群組中的任何其他輪詢者。

此外，如果您指定用戶群組識別碼，而 Apache Kafka 找到具有相同 ID 的有效現有用戶群組，則 EventBridge 會忽略管道的StartingPosition參數。相反地，會根據用戶群組的認可偏移量 EventBridge 開始處理記錄。如果您指定用戶群組識別碼，而 Apache Kafka 找不到現有的用戶群組，請使用指 EventBridge 定的來源設定。StartingPosition

您指定的取用者群組 ID 在所有 Kafka 事件來源中必須是唯一的。使用指定的取用者群組 ID 建立管道之後，您就無法更新此值。

Amazon MSK 來源的自動擴展

當您最初建立 Amazon MSK 來源時， EventBridge 會配置一個取用者來處理 Apache Kafka 主題中的所有分割區。每個取用者都有多個並行運行的處理器以處理增加的工作負載。此外，根據工作負載， EventBridge 自動擴展或減少取用者的數量。為了保留每個分割區中的訊息順序，主題中每個分割區的取用者數上限是一個取用者。

在一分鐘間隔內， EventBridge 評估主題中所有分割區的消費者偏移延遲。如果延遲太高，則分區接收消息的速度超 EventBridge 過了處理它們的速度。如有必要，可在主題中 EventBridge 新增或移除取用者。新增或刪除取用者的擴展過程，將在三分鐘的評估期間內完成。

如果您的目標超載， EventBridge 減少消費者的數量。此動作可透過減少取用者可擷取和傳送至管道的訊息數量，減少管道的工作負載。

您的瀏覽器已停用或無法使用 Javascript。

您必須啟用 Javascript，才能使用 AWS 文件。請參閱您的瀏覽器說明頁以取得說明。

文件慣用形式

Amazon MQ 訊息代理程式

阿帕奇卡夫卡流