使用 Kafka 事件生产者

Code Engine Kafka 订阅会监听 Kafka 实例中出现的新消息。 为一组主题创建预订时，应用程序或作业将针对其中一个主题中显示的每条新消息接收单独的事件。 每个项目最多可创建 100 个 Kafka 订阅。

虽然您可以使用任何 Kafka 实例，但本主题中的示例使用的是 IBM® Event Streams for IBM Cloud® 服务。Event Streams 是 IBM 提供的一项事件流服务，用于处理 Kafka 事件。 有关此服务的更多信息，请参阅 Event Streams 文档。

设置 Kafka 事件生产者

您可以配置您的 Kafka 消息发布者，使其向 Code Engine 和 Kafka 事件订阅发送消息。 利用您的 Code Engine Kafka 事件订阅，在收到 Kafka 消息时触发应用程序或作业。

首先，为事件流服务创建 Event Streams 服务实例。 虽然您可以使用控制台或 CLI，但以下步骤描述了如何使用 CLI 设置 Event Streams 事件生产者。

设置 Event Streams CLI 环境

  1. 下载并安装 IBM Cloud CLI。 登录到 IBM Cloud CLI。

    ibmcloud login

  2. 下载并安装 Code Engine CLI。

    ibmcloud plugin install code-engine -f

  3. 要使用 Event Streams 服务来创建 Kafka 实例，请下载并安装 Event Streams CLI。

    ibmcloud plugin install event-streams -f

  4. 登录到 IBM Cloud 帐户并以资源组为目标。 通过运行以下命令将资源组设定为目标。 要获取您的资源组列表，请运行 ibmcloud resource groups 命令。

    ibmcloud target -g <resource_group>

配置您的 Kafka 实例

  1. 为 Event Streams创建服务实例。 Event Streams CLI 服务的名称为 messagehub。 对于此示例，请创建名为 myeventstream 的 Event Streams 服务实例。

    ibmcloud resource service-instance-create myeventstream messagehub lite us-south

  2. 创建服务密钥以向服务实例提供凭证。

    ibmcloud resource service-key-create myeventstream-key Manager --instance-name myeventstream

    示例输出

    Creating service key of service instance myeventstream under account <user_account>...
OK
Service key crn:v1:bluemix:public:messagehub:us-south:a/e43abfcbd191404cb17ef650e9681dd3:c0736069-3f4a-438a-b614-6846877d692d:resource-key:4c8edfdb-abcd-abcd-abcd-abcdabcdabcd was created.
Name:          myeventstream-key
ID:            crn:v1:bluemix:public:messagehub:us-south:a/e43abfcbd191404cb17ef650e9681dd3:c0736069-3f4a-438a-b614-6846877d692d:resource-key:4c8edfdb-abcd-abcd-abcd-abcdabcdabcd
Created At:    Mon Mar 21 18:36:09 UTC 2022
State:         active
Credentials:
            api_key:                  abcdeH9tu3qE5Sn8VbJfcDEWtjR_l0iPisB3abcdefgh
            apikey:                   abcdeH9tu3qE5Sn8VbJfcDEWtjR_l0iPisB3abcdefgh
            iam_apikey_description:   Auto-generated for key crn:v1:bluemix:public:messagehub:us-south:a/e43abfcbd191404cb17ef650e9681dd3:c0736069-3f4a-438a-b614-6846877d692d:resource-key:4c8edfdb-abcd-abcd-abcd-abcdabcdabcd
            iam_apikey_name:          myeventstream-key
            iam_role_crn:             crn:v1:bluemix:public:iam::::serviceRole:Manager
            iam_serviceid_crn:        crn:v1:bluemix:public:iam-identity::a/e43abfcbd191404cb17ef650e9681dd3::serviceid:ServiceId-3e99caa5-b174-4f04-9845-5c5d783b8bc7
            instance_id:              c0736069-3f4a-438a-b614-6846877d692d
            kafka_admin_url:          https://abcdabcdabcdabcd.svc07.us-south.eventstreams.cloud.ibm.com
            kafka_brokers_sasl:       [broker-1-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 broker-2-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 broker-5-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 broker-3-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 broker-4-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 broker-0-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093]
            kafka_http_url:           https://abcdabcdabcdabcd.svc07.us-south.eventstreams.cloud.ibm.com
            password:                 abcdeH9tu3qE5Sn8VbJfcDEWtjR_l0iPisB3abcdefgh
            user:                     token

    记下服务密钥的 user，password 和 kafka-brokers_sasl 的值。 当您设置 Code Engine Kafka 预订时，需要此信息。 password 和 apikey 的值在 Event Streams 服务实例的服务密钥中相同。 您还可以使用 ibmcloud resource service-key myeventstream-key 命令来检索服务密钥信息。

  3. 相对于 Event Streams 服务实例，初始化 Event Streams 插件。

    ibmcloud es init --instance-name myeventstream

  4. 创建 Event Streams 主题。

    ibmcloud es topic-create kafka-topic1

设置 Code Engine 样本应用程序以生成 Kafka 消息

对于此场景，让我们使用 Code Engine 应用程序来充当 Kafka 消息的事件生产者。 此应用程序的用途是连接到 Event Streams 实例并发送 Kafka 消息。 此应用程序使用 Code Engine Kafka 发送方样本应用程序 来发送 Kafka 消息。 此样本发送方映像需要 BROKERS 环境变量以及包含 password 凭证的私钥。 您可以通过控制台或使用 CLI 创建此应用程序。

向 Event Streams 发送 Kafka 消息时，请务必指定 Content-Type 标头。 指定此头，以便使用者可以接收具有预期内容类型的消息; 例如，application/json。

使用 Kafka 样本所需的凭证创建私钥

在创建 Code Engine 应用程序以发送 Kafka 消息之前，请创建包含所需凭证的 Code Engine 私钥。

准备工作

为简化此场景，请创建一个密钥“kafka-subscription-secret”，用于存储“Kafka”发送方示例应用和“Kafka”事件订阅所需的凭据，其中“”事件订阅使用“Kafka”接收方示例。 示例发送应用 Kafka 和事件订阅 Code Engine Kafka 需要这些凭据，才能与服务实例 Event Streams 进行通信。 虽然在创建“Kafka”发送者应用和事件订阅之前，并不要求您先创建此密钥，但此操作可以简化后续步骤。

使用控制台中的 Kafka 样本所需的凭证创建私钥

要从控制台创建 kafka-subscription-secret 私钥，请转至 私钥和配置映射，然后单击 创建 并选择要创建的私钥。 有关更多信息，请参阅 从控制台创建私钥。

  • 使用 user 的值指定 username 键，该值在 Event Streams 服务实例中的服务凭证详细信息中列出。 对于 Event Streams 服务实例，此值为 token。 需要此密钥才能在 Code Engine Kafka 事件预订与 Kafka 消息代理之间进行认证。
  • 使用 apikey 的值指定 password 键，该值在 Event Streams 服务实例中的服务凭证详细信息中列出。 此密钥对于发送方样本是必需的，用于启用 Code Engine Kafka 事件预订与 Kafka 消息代理之间的通信。
使用 CLI 创建具有 Kafka 样本所需的凭证的私钥

要使用 CLI 创建 kafka-subscription-secret 私钥，请为 password 和 username 添加字面值环境变量。 有关更多信息，请参阅 使用 CLI 创建私钥。

  • 使用 user 的值指定 username 键，该值在 Event Streams 服务实例中的服务凭证详细信息中列出。 对于 Event Streams 服务实例，此值为 token。 需要此密钥才能在 Code Engine Kafka 事件预订与 Kafka 消息代理之间进行认证。

  • 使用 apikey 的值指定 password 键，该值在 Event Streams 服务实例中的服务凭证详细信息中列出。 此密钥对于发送方样本是必需的，用于启用 Code Engine Kafka 事件预订与 Kafka 消息代理之间的通信。

    ibmcloud ce secret create --name kafka-subscription-secret --from-literal password=<value_of_apikey> --from-literal username=<value_of_user>

    例如，

    ibmcloud ce secret create --name kafka-subscription-secret --from-literal password=abcdeH9tu3qE5Sn8VbJfcDEWtjR_l0iPisB3abcdefgh --from-literal username=token

创建 Code Engine 应用程序以发送事件

创建一个 Code Engine 应用程序，用于连接到您的 Event Streams 实例，并向 Kafka 消息的接收方（ Kafka 消费者）生成（发送） Kafka 消息。

创建 Code Engine 应用程序以从控制台发送事件

要从控制台创建 kafka-sender-app 应用程序，请完成以下步骤。

  1. 名为 kafka-sender-app 的 创建 Code Engine 应用程序，其中包含以下信息。

    1. 请参阅此应用程序的 icr.io/codeengine/kafka-sender 容器映像。 此映像是从 sender.go 构建的，可从 样本中获取 IBM Cloud Code Engine GitHub 存储库。 此样本发送方应用程序需要 password 和 BROKERS 的值。
    2. 在 环境变量 (可选) 部分中，添加以下环境变量。
      1. 添加字面值环境变量 BROKERS。 对于此密钥的值，请指定 Event Streams 服务实例的服务凭证详细信息中列出的一个或多个代理主机。
      2. 添加另一个环境变量以 引用完整私钥 kafka-subscription-secret。 此私钥包含 password 的凭证。
    3. (可选) 在 资源和缩放 部分中，指定 1 作为最小实例数，以便应用程序始终具有正在运行的实例，并且不会缩放到零。 在查看日志时，将应用程序配置为始终具有正在运行的实例非常有用。 如果您正在生产环境中运行，请考虑保留应用程序的运行实例的成本，或者是否希望 Code Engine 自动缩放为零。 缺省情况下，应用程序在未使用时将缩放为零。
    4. 单击 创建 以创建和部署应用程序。

  2. 确认此应用程序处于 ready 状态。

创建 Code Engine 应用程序以使用 CLI 发送事件

要使用 CLI 创建 kafka-sender-app 应用程序，请使用以下命令。

  1. 名为 kafka-sender-app 的 创建 Code Engine 应用程序，其中包含以下信息。

    • 指定 --image 选项以引用 icr.io/codeengine/kafka-sender 容器映像。 此映像是从 sender.go 构建的，可从 样本中获取 IBM Cloud Code Engine GitHub 存储库。 此样本发送方应用程序需要存储在 kafka-subscription-secret 中的 password 凭证，并且需要 BROKERS 环境变量。
    • 指定 --env-from-secret 选项以引用包含 password 凭证的完整密钥 kafka-subscription-secret。
    • 指定 --env 选项以添加字面值环境变量 BROKERS，并提供在 Event Streams 服务实例的服务凭证详细信息中列出的其中一个代理主机的名称。 但是，如果要指定多个代理主机名，请使用格式 --env BROKERS-broker1,broker2,broker3。
    • (可选) 指定 --min-scale=1 选项，以便应用程序始终具有正在运行的实例，并且不会缩放到零。 在查看日志时，将应用程序配置为始终具有正在运行的实例非常有用。 如果您正在生产环境中运行，请考虑保留应用程序的运行实例的成本，或者是否希望 Code Engine 自动缩放为零。 缺省情况下，应用程序在未使用时将缩放为零。
    ibmcloud ce app create --name kafka-sender-app --image icr.io/codeengine/kafka-sender --env-from-secret kafka-subscription-secret --env BROKERS=broker-4-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --min-scale 1

设置 Code Engine 以接收应用程序的 Kafka 事件

要使 Code Engine 与 Kafka 事件配合使用，请通过控制台或CLI配置 Code Engine Kafka 事件订阅，以连接到 Kafka 事件代理，并监听 Kafka 事件。 此外，请设置一个 Code Engine 应用程序（或作业），用作 Kafka 事件的接收方。 Kafka 事件订阅定义了 Kafka 事件发布者（发送方）与消费者（接收方）之间的关系。

Code Engine Kafka 事件订阅连接到 Kafka 消息代理，并为每条传入的 Kafka 消息向接收方应用程序发送 HTTP Post 请求。 更多信息，请参阅 HTTP 事件的标题和正文信息。

从控制台预订应用程序的 Kafka 事件

您可以使用控制台设置 Kafka 事件订阅，以便将事件发送至 Code Engine 应用程序。

创建 Code Engine 应用程序以从控制台接收 Kafka 事件

  1. 创建 Code Engine 应用程序，以充当 Kafka 消息的事件使用者并接收 Kafka 事件。 例如，创建名为 kafka-receiver-app 的应用程序，该应用程序使用 icr.io/codeengine/kafka-receiver 映像。 此映像是从 receiver.go 构建的，可从 样本中获取 IBM Cloud Code Engine GitHub 存储库。 此样本不需要任何环境变量。
  2. 部署此应用程序后，请确认其处于 ready 状态。

使用控制台时，在创建 Kafka 事件订阅之前，用于接收 Kafka 事件的应用程序或作业并不一定需要已经存在。 但是，如果创建事件预订时应用程序或作业不存在，那么预订的状态将反映使用者不存在。 您必须在预订处于就绪状态之前创建应用程序或作业，并且可以通过此预订接收事件。

从控制台为应用程序创建 Code Engine Kafka 事件预订

Kafka 事件订阅定义了 Kafka 事件发布者（发送方）与消费者（接收方）之间的关系。

准备工作

完成以下步骤以从控制台为应用程序创建 Kafka 事件预订。

  1. 从 Code Engine "项目" 页面，转至您的项目。
  2. 在“概述”页面中，单击 事件预订。
  3. 在“事件预订”页面中，单击 创建 以创建预订。
  4. 从“创建事件预订”页面中，选择 Event Streams / Kafka 磁贴以指定事件类型。
  5. 对于 常规，请提供 Event Streams / Kafka 预订的名称。 点击 “下一步”继续。
  6. 有关消息代理的详细信息，
    1. 指定用于接收消息的 Kafka 消息代理主机，这些消息队列中的消息将通过此订阅作为事件接收。 要获取有关代理主机，主题和访问凭证的信息，请在 Event Streams 控制台中查看服务实例的服务凭证详细信息。 例如，为 myeventstream-key 服务实例的消息代理主机指定 "broker-0-abcdabcdabcdabcd.kafka.svc01.us-south.eventstreams.cloud.ibm.com:9093", "broker-1-abcdabcdabcdabcd.kafka.svc01.us-south.eventstreams.cloud.ibm.com:9093"。 您可以在 Kafka_brokers_sasl 字段的 Event Streams 中找到服务实例的代理程序。
    2. 单击 配置 以配置对消息代理的访问权。 要从 Code Engine 向 Kafka 或 Event Streams 实例进行认证，您需要提供消息代理访问私钥。
      • 您可以创建新密钥，选择现有密钥，或者如果不需要凭证来访问消息代理，请选择 None。
      • 要创建私钥，请单击 创建。 提供私钥的名称以及 username 和 password 的值。 username 和 password 的值必须与 Kafka 或 Event Streams 实例的服务凭证中的值匹配。 例如，username 的值是 user 的值，此值在 Event Streams 服务实例中的服务凭证详细信息中列出。 password 的值是在 Event Streams 服务实例的服务凭证详细信息中列出的 apikey 的值。
      • 对于此示例，请使用先前创建的现有 kafka-subscription-secret 私钥。
    3. 指定消息队列的现有主题的名称。 例如，kafka-topic1。 要获取有关服务实例的现有主题的信息，请转至 Event Streams 控制台中的服务实例并查看 主题。
    4. (可选) 指定使用者组。 Kafka 消息的使用者可以分组为 使用者组。 如果您正在使用使用者组，那么主题配置将控制到使用者组中的使用者的消息流。 每当向使用者组添加使用者或从该使用者组中除去使用者时，该主题中的消息流可能会更改。 此操作可能会导致现有使用者不再接收来自该主题的消息。
    5. 点击 “下一步”继续。
  7. 对于 事件使用者，指定 Code Engine 应用程序以接收事件。 请注意，您可以从已定义的应用程序和作业的列表中进行选择，也可以为尚未创建的应用程序 (或作业) 提供名称。 使用控制台创建事件预订时，应用程序或作业不需要存在。 但是，创建预订时，该预订的状态反映使用者不存在。 您必须在预订处于就绪状态之前创建应用程序或作业，并且可以通过此预订接收事件。 对于此示例，请使用引用 icr.io/codeengine/kafka-receiver 映像的 kafka-receiver-app 应用程序。 如果应用程序不存在，请提供应用程序的名称，并在创建 Kafka 预订后 创建应用程序。 (仅限应用程序) 您可以选择指定路径。 默认情况下，事件会路由到目标应用程序的根 URL。 您可以通过指定路径将事件发送到应用程序中的其他目标。 例如，如果预订路径指定 /events，那么会将事件发送到 https://<base application URL>/events。 点击 “下一步”继续。
  8. 对于 摘要，请查看 Kafka 事件预订的设置，并根据需要进行更改。 准备就绪后，单击 创建 以创建 Kafka 预订。

从控制台将事件发送到接收应用程序

现在，您已创建了引用 kafka-receiver-app 应用程序的 Kafka 事件订阅，请使用 kafka-sender-app 将消息事件发送至接收应用程序。

  1. 开始将接收方应用程序记录到 从控制台查看应用程序日志 以查看事件。

  2. (可选) 开始将发送方应用程序记录到 从控制台查看应用程序日志，以查看有关发送的事件的信息。

  3. 启动日志记录后，使用 curl 调用 kafka-sender-app 应用程序，并指定 kafka-sender-app 的公共 URL、主题名称和要发送的信息数量。 您可以从应用程序的“域映射”选项卡中获取此应用程序的公共 URL。 例如，

    curl "<public_URL_of_Kafka_sender_app>?topic=<your_topic_name>&num=<number_of_messages_to_produce>"

请确保将该值用引号括起来，以确保将其视为单个字符串。

使用 CLI 预订应用程序的 Kafka 事件

您可以使用命令行界面（CLI）设置 Kafka 事件订阅，以便将事件发送至 Code Engine 应用程序。

事件以 HTTP POST请求的形式发送给应用程序。 有关 Kafka 事件所包含信息的更多详情，请参阅 HTTP 中关于事件的标头和正文信息。 如果您的事件被发送到一个 Code Engine 任务，该任务会将事件作为环境变量接收。 有关 Kafka 订阅的环境变量的更多信息，请参阅 “事件的环境变量”。

创建 Code Engine 应用程序以使用 CLI 接收 Kafka 事件

准备工作

  1. 创建 Code Engine 应用程序，以充当 Kafka 消息的事件使用者并接收 Kafka 事件。 例如，创建名为 kafka-receiver-app2 的应用程序，该应用程序使用 icr.io/codeengine/kafka-receiver 映像。 此映像是从 receiver.go 构建的，可从 样本中获取 IBM Cloud Code Engine GitHub 存储库。 此样本不需要任何环境变量。 您可以选择指定 --min-scale=1 选项，以便应用程序始终具有正在运行的实例，并且不会缩放到零。 在查看日志时，将应用程序配置为始终具有正在运行的实例非常有用。 如果您正在生产环境中运行，请考虑保留应用程序的运行实例的成本，或者是否希望 Code Engine 自动缩放为零。 缺省情况下，应用程序在未使用时将缩放为零。

    ibmcloud ce app create -n kafka-receiver-app2 --image icr.io/codeengine/kafka-receiver --min-scale 1

    默认情况下，事件会路由到目标应用程序的根 URL。 您可以使用 --path 选项将事件发送到应用程序中的其他目标。 例如，如果预订指定 --path /event，那么会将事件发送到 https://<base application URL>/events。

  2. 部署此应用程序后，运行 app get 命令以确认应用程序处于 ready 状态。

    ibmcloud ce app get -n kafka-receiver-app2

使用 CLI 为应用程序创建 Code Engine Kafka 事件预订

您可以使用 CLI 创建一个 Kafka 事件订阅，该订阅定义了 Kafka 事件发布者（发送方）与消费者（接收方）之间的关系。

  1. 使用 ibmcloud ce sub kafka create 命令为 Kafka 事件创建 Code Engine Kafka 事件预订。 使用先前创建的 kafka-subscription-secret 私钥来访问消息代理。 根据 Kafka 资源的服务凭证信息指定代理信息。 对于此示例，您可以从 ibmcloud resource service-key myeventstream-key 命令的输出中获取代理信息。 请注意，您必须为主题的每个代理程序指定 --broker 选项。 --destination 选项指定接收事件的 Code Engine 资源。

    ibmcloud ce sub kafka create --name mykafkasubscription --destination kafka-receiver-app2 --secret kafka-subscription-secret --topic kafka-topic1 --broker broker-3-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-5-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker  broker-0-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-1-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-4-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-2-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093

  2. 显示 Kafka 事件订阅的详细信息。

    ibmcloud ce sub kafka get -n mykafkasubscription

    示例输出

    Getting Kafka event subscription 'mykafkasubscription'...
OK
Name:          mykafkasubscription
[...]
Destination Type:                 app
Destination:                      kafka-receiver-app2
Brokers:
broker-3-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
broker-5-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
broker-0-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
broker-1-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
broker-4-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
broker-2-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
Consumer Group:                   knative-kafka-source-a4072fe1-1dfa-4470-9d07-bf7a0ff8e340
Topics:
kafka-topic1
Secret key reference (user):      kafka-subscription-secret.username
Secret key reference (password):  kafka-subscription-secret.password
Ready:                            true
Conditions:
Type                     OK    Age  Reason
ConnectionEstablished    true  24s
InitialOffsetsCommitted  true  24s
Ready                    true  24s
Scheduled                true  24s
SinkProvided             true  24s
Events:
Type     Reason           Age  Source                  Messages
Normal   FinalizerUpdate  26s  kafkasource-controller  Updated "mykafkasubscription" finalizers

使用 CLI 将事件发送到接收应用程序

现在，您已创建了引用 kafka-receiver-app 应用程序的 Kafka 事件订阅，请使用 kafka-sender-app 将消息事件发送至接收应用程序。

  1. URL 使用带有 kafka-sender-app 选项的 ibmcloud ce app get 命令，使用 --output url 选项查找应用程序的 URL。

    ibmcloud ce app get -n kafka-sender-app --output url

    示例输出

    https://kafka-sender-app.abcdabcdabc.us-south.codeengine.appdomain.cloud

  2. 运行 Kafka 事件生产者应用程序 kafka-sender-app 以将事件发送到目标 Code Engine 应用程序。 使用 curl 调用 kafka-sender-app 应用程序，并指定主题的值和消息数。 使用 ibmcloud ce app get 命令的输出结果来查找应用程序的公共 URL。 请确保将该值用引号括起来，以确保将其视为单个字符串。

    curl "<public_URL_of_Kafka_sender_app>?topic=<your_topic_name>&num=<number_of_messages_to_produce>"

    例如，

    curl "https://kafka-sender-app.abcdabcdabc.us-south.codeengine.appdomain.cloud?topic=kafka-topic1&num=1"

  3. 查看日志中的事件。 使用有效的代理程序，主题和访问私钥创建 Kafka 事件预订时，如果您有 Kafka 应用程序 (例如 kafka-sender-app) 生成有关该主题的消息，那么您可以在日志中看到接收 Kafka 消息的目标 Code Engine 应用程序 (例如 kafka-receiver-app) 的事件。 使用 Kafka 接收方应用程序 (icr.io/codeengine/kafka-receiver) 时，请在日志中搜索接收方应用程序的 Event data 以查看接收到的消息。

    ibmcloud ce app logs -n kafka-receiver-app2

    示例输出

    Getting logs for all instances of application 'kafka-receiver-app2'...
OK
kafka-receiver-app2-00001-deployment-66976f7988-9xttm/user-container:
2022/03/31 22:19:45 Listening on port 8080
2022/03/31 22:19:46 ----------
2022/03/31 22:19:46 Path: /
2022/03/31 22:19:46 Header: Accept-Encoding=[gzip]
2022/03/31 22:19:46 Header: Ce-Id=[partition:0/offset:167]
2022/03/31 22:19:46 Header: Ce-Source=[/apis/v1/namespaces/glxo4k7nj7d/kafkasources/mykafkasubscription#kafka-topic1]
2022/03/31 22:19:46 Header: Ce-Specversion=[1.0]
2022/03/31 22:19:46 Header: Ce-Subject=[partition:0#167]
2022/03/31 22:19:46 Header: Ce-Time=[2022-03-31T22:19:36.499Z]
2022/03/31 22:19:46 Header: Ce-Type=[dev.knative.kafka.event]
2022/03/31 22:19:46 Header: Content-Length=[8]
2022/03/31 22:19:46 Header: Forwarded=[for=172.30.208.213;proto=http, for=127.0.0.6]
2022/03/31 22:19:46 Header: K-Proxy-Request=[activator]
2022/03/31 22:19:46 Header: Traceparent=[00-b033708685c715a7c2384cdf05797785-65540b0937e9b0ce-00]
2022/03/31 22:19:46 Header: User-Agent=[Go-http-client/1.1]
2022/03/31 22:19:46 Header: X-B3-Parentspanid=[e1a785d7fdbead6c]
2022/03/31 22:19:46 Header: X-B3-Sampled=[1]
2022/03/31 22:19:46 Header: X-B3-Spanid=[abcde9901e6bf83f]
2022/03/31 22:19:46 Header: X-B3-Traceid=[abcde490a426573772fa0bf60caf5ddb]
2022/03/31 22:19:46 Header: X-Envoy-Attempt-Count=[1]
2022/03/31 22:19:46 Header: X-Forwarded-For=[172.30.208.213, 127.0.0.6, 127.0.0.6]
2022/03/31 22:19:46 Header: X-Forwarded-Proto=[http]
2022/03/31 22:19:46 Header: X-Request-Id=[abcdeb4e-c5ac-abcd-abcd-60e6278abcde]
2022/03/31 22:19:46 Event data: test1: 1

    请注意，应用程序的日志信息仅持续一个小时。 有关查看应用程序 (或作业) 日志的更多信息，请参阅 查看日志。

传递到应用程序的 Kafka 事件的头和主体信息

发送到应用程序的所有事件均以 HTTP POST消息的形式接收。 事件包含某些 HTTP 标头，这些标头可帮助您快速确定事件的关键信息，而无需查看事件正文（业务逻辑）。 有关更多信息，请参阅 CloudEvents 规范。

应用程序的头

事件头文件
头 描述
ce-id 事件的唯一标识，除非事件已重放，在这种情况下，将为其分配相同的标识。
ce-source 用于指示此事件在事件生产者中的起源位置的 URI 引用。 对于 Kafka 事件，此头采用以下格式: /apis/v1/namespaces/[PROJECT_SUBDOMAIN]/kafkasources/[KAFKA_SUBSCRIPTION_NAME]#[TOPIC_NAME]。
ce-specversion CloudEvents 规范的版本。 该值始终为 1.0。
ce-subject Kafka 分区号和偏移量。 例如，partition:0#1 指的是分区 0 和偏移量 1。
ce-time 生成该事件的时间。
ce-type 事件类型。 对于 Kafka 事件，这是 dev.knative.kafka.event。

示例输出

Ce-Id=[partition:0/offset:0]
Ce-Source=[/apis/v1/namespaces/ewgz38l13ts/kafkasources/mykafkasubscription#kafka-topic1]
Ce-Specversion=[1.0]
Ce-Subject=[partition:0#0]
Ce-Time=[2021-09-27T16:39:01.36Z]
Ce-Type=[dev.knative.kafka.event]

HTTP 应用程序主体

HTTP 主体包含 Kafka 消息，格式由您在创建或更新订阅时指定。

设置 Code Engine 以接收作业的 Kafka 事件

要使 Code Engine 与 Kafka 事件配合使用，请通过控制台或CLI配置 Code Engine Kafka 事件订阅，以连接到 Kafka 事件代理，并监听 Kafka 事件。 此外，请设置一个 Code Engine 任务（或应用程序），用作 Kafka 事件的接收方。 Kafka 事件订阅定义了 Kafka 事件发布者（发送方）与消费者（接收方）之间的关系。

从控制台预订作业的 Kafka 事件

您可以使用控制台设置 Kafka 事件订阅，以便将事件发送至 Code Engine 作业。

为作业创建事件预订时，将为触发的每个事件创建作业运行。 此作业运行具有与作业相关的环境变量。 Code Engine Kafka 事件订阅会连接到您的 Kafka 消息代理，并发送与该作业相关的环境变量。 有关 Kafka 发送的环境变量的更多信息，请参阅 “事件的环境变量”。

创建 Code Engine 作业以从控制台接收 Kafka 事件

  1. 创建 Code Engine 作业，以充当 Kafka 消息的事件使用者并接收 Kafka 事件。 例如，创建名为 kafka-receiver-job 的作业，该作业使用样本 icr.io/codeengine/codeengine 映像。 此映像是根据 codeengine.go 构建的，可从 样本 IBM Cloud Code Engine GitHub 存储库获取。 此样本不需要任何环境变量。
  2. 创建此作业后，请确认其处于 ready 状态。

使用控制台时，在创建 Kafka 事件订阅之前，用于接收 Kafka 事件的应用程序或作业并不一定需要已经存在。 但是，如果创建事件预订时应用程序或作业不存在，那么预订的状态将反映使用者不存在。 您必须在预订处于就绪状态之前创建应用程序或作业，并且可以通过此预订接收事件。

从控制台为作业创建 Code Engine Kafka 事件预订

Kafka 事件订阅定义了 Kafka 事件发布者（发送方）与消费者（接收方）之间的关系。

准备工作

完成以下步骤以从控制台为应用程序创建 Kafka 事件预订。

  1. 从 Code Engine "项目" 页面，转至您的项目。
  2. 在“概述”页面中，单击 事件预订。
  3. 在“事件预订”页面中，单击 创建 以创建预订。
  4. 从“创建事件预订”页面中，选择 Event Streams / Kafka 磁贴以指定事件类型。
  5. 对于 常规，请提供 Event Streams / Kafka 预订的名称。 点击 “下一步”继续。
  6. 有关消息代理的详细信息，
    1. 指定用于接收消息的 Kafka 消息代理主机，这些消息队列中的消息将通过此订阅作为事件接收。 要获取有关代理主机，主题和访问凭证的信息，请在 Event Streams 控制台中查看服务实例的服务凭证详细信息。 例如，为 myeventstream-key 服务实例的消息代理主机指定 "broker-0-abcdabcdabcdabcd.kafka.svc01.us-south.eventstreams.cloud.ibm.com:9093", "broker-1-abcdabcdabcdabcd.kafka.svc01.us-south.eventstreams.cloud.ibm.com:9093"。
    2. 单击 配置 以配置对消息代理的访问权。 要从 Code Engine 向 Kafka 或 Event Streams 实例进行认证，您需要提供消息代理访问私钥。
      • 您可以创建新密钥，选择现有密钥，或者如果不需要凭证来访问消息代理，请选择 None。
      • 要创建私钥，请单击 创建。 提供私钥的名称以及 username 和 password 的值。 username 和 password 的值必须与 Kafka 或 Event Streams 实例的服务凭证中的值匹配。 例如，username 的值是 user 的值，此值在 Event Streams 服务实例中的服务凭证详细信息中列出。 password 的值是在 Event Streams 服务实例的服务凭证详细信息中列出的 apikey 的值。
      • 对于此示例，请使用先前创建的现有 kafka-subscription-secret 私钥。
    3. 指定消息队列的现有主题的名称。 例如，kafka-topic1。 要获取有关服务实例的现有主题的信息，请转至 Event Streams 控制台中的服务实例并查看 主题。
    4. (可选) 指定使用者组。 Kafka 消息的使用者可以分组为 使用者组。 如果您正在使用使用者组，那么主题配置将控制到使用者组中的使用者的消息流。 每当向使用者组添加使用者或从该使用者组中除去使用者时，该主题中的消息流可能会更改。 此操作可能会导致现有使用者不再接收来自该主题的消息。
    5. 点击 “下一步”继续。
  7. 对于 事件使用者，指定 Code Engine 作业以接收事件。 请注意，您可以从已定义的作业和应用程序的列表中进行选择，也可以为尚未创建的作业 (或应用程序) 提供名称。 使用控制台创建事件预订时，应用程序或作业不需要存在。 但是，创建预订时，该预订的状态反映使用者不存在。 您必须在预订处于就绪状态之前创建作业 (或应用程序)，并且可以通过此预订接收事件。 对于此示例，选择 job 作为组件类型，并将引用 icr.io/codeengine/codeengine 图像的 kafka-receiver-job 作业用作接收事件的组件。 如果作业不存在，请提供作业的名称，并在创建 Kafka 预订后 创建作业。 (仅限应用程序) 您可以选择指定路径。 点击 “下一步”继续。
  8. 对于 摘要，请查看 Kafka 事件预订的设置，并根据需要进行更改。 准备就绪后，单击 创建 以创建 Kafka 预订。

从控制台向接收作业发送事件

现在，您已创建了引用 kafka-receiver-job 作业的 Kafka 事件订阅，请使用 kafka-sender-app 将消息事件发送至接收作业。

  1. 开始将接收方作业记录到 从控制台查看作业日志 以查看事件。

  2. (可选) 开始将发送方应用程序记录到 从控制台查看应用程序日志，以查看有关发送的事件的信息。

  3. 启动日志记录后，使用 curl 调用 kafka-sender-app 应用程序，并指定 kafka-sender-app 的公共 URL、主题名称和要发送的信息数量。 您可以从应用程序的“域映射”选项卡中获取此应用程序的公共 URL。 例如，

    curl "<public_URL_of_Kafka_sender_app>?topic=<your_topic_name>&num=<number_of_messages_to_produce>"

请确保将该值用引号括起来，以确保将其视为单个字符串。

使用有效的代理程序，主题和访问私钥创建 Kafka 预订时，如果您的 Kafka 作业生成有关该主题的消息 (例如 kafka-sender-app)，那么您可以在日志中看到接收 Kafka 消息 (例如 kafka-receiver-job) 的 Code Engine 作业的事件。 使用 Kafka 接收方作业 (icr.io/codeengine/codeengine) 时，请在接收方作业的日志中搜索 CE_DATA 以查看接收到的消息。

使用 CLI 预订作业的 Kafka 事件

您可以使用命令行界面 (CLI) 配置一个 Kafka 事件订阅，以便将事件发送到 Code Engine 任务。

为作业创建事件预订时，将为触发的每个事件创建作业运行。 此作业运行具有与作业相关的环境变量。 Code Engine Kafka 事件订阅会连接到您的 Kafka 消息代理，并发送与该作业相关的环境变量。 有关 Kafka 发送的环境变量的更多信息，请参阅 “事件的环境变量”。

创建 Code Engine 作业以使用 CLI 接收 Kafka 事件

准备工作

  1. 创建 Code Engine 作业，以充当 Kafka 消息的事件使用者并接收 Kafka 事件。 例如，创建名为 kafka-receiver-job 的作业，该作业使用 icr.io/codeengine/codeengine 映像。 此映像是根据 codeengine.go 构建的，可从 样本 IBM Cloud Code Engine GitHub 存储库获取。

    ibmcloud ce job create -n kafka-receiver-job --image icr.io/codeengine/codeengine

  2. (可选) 创建此作业后，运行 job get 命令以查看有关此作业的信息。

    ibmcloud ce job get -n kafka-receiver-job

使用 CLI 为作业创建 Code Engine Kafka 事件预订

您可以使用 CLI 创建一个 Kafka 事件订阅，该订阅定义了 Kafka 事件发布者（发送方）与消费者（接收方）之间的关系。

  1. 使用 ibmcloud ce sub kafka create 命令为 Kafka 事件创建 Code Engine Kafka 事件预订。 使用先前创建的 kafka-subscription-secret 私钥来访问消息代理。 根据 Kafka 资源的服务凭证信息指定代理信息。 对于此示例，您可以从 ibmcloud resource service-key myeventstream-key 命令的输出中获取代理信息。 请注意，您必须为主题的每个代理程序指定 --broker 选项。 --destination 选项指定接收事件的 Code Engine 资源。 使用接收作业时，还必须指定 --destination-type 选项以指定资源是作业，因为此选项的缺省值为 app。

    ibmcloud ce sub kafka create --name mykafkasubscription-withjob --destination-type job --destination kafka-receiver-job --secret kafka-subscription-secret --topic kafka-topic1 --broker broker-3-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-5-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker  broker-0-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-1-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-4-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-2-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093

  2. 显示 Kafka 事件订阅的详细信息。

    ibmcloud ce sub kafka get -n mykafkasubscription-withjob

    示例输出

    Getting Kafka event subscription 'mykafkasubscription-withjob'...
OK
Name:          mykafkasubscription-withjob
[...]
Destination Type:                 job
Destination:                      kafka-receiver-job
Brokers:
broker-3-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
broker-5-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
broker-0-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
broker-1-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
broker-4-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
broker-2-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
Consumer Group:                   knative-kafka-source-a4072fe1-1dfa-4470-9d07-bf7a0ff8e340
Topics:
kafka-topic1
Secret key reference (user):      kafka-subscription-secret.username
Secret key reference (password):  kafka-subscription-secret.password
Ready:                            true
[...]

使用 CLI 将事件发送到接收作业

现在，您已创建了引用 kafka-receiver-job 应用程序的 Kafka 事件订阅，请使用 kafka-sender-app 将消息事件发送至接收应用程序。

  1. URL kafka-sender-app 使用 ibmcloud ce app get 命令查找应用程序的 URL。

    ibmcloud ce app get -n kafka-sender-app --output url

    示例输出

    https://kafka-sender-app.abcdabcdabc.us-south.codeengine.appdomain.cloud

  2. 运行 Kafka 事件生产者应用程序 kafka-sender-app 以将事件发送到目标 Code Engine 作业。 使用 curl 调用 kafka-sender-app 应用程序，并指定主题的值和消息数。 使用 ibmcloud ce app get 命令的输出，查找事件生成应用程序的公共 URL。 请确保将该值用引号括起来，以确保将其视为单个字符串。 例如，

    curl "<public_URL_of_Kafka_sender_app>?topic=<your_topic_name>&num=<number_of_messages_to_produce>"

    例如，

    curl "https://kafka-sender-app.abcdabcdabc.us-south.codeengine.appdomain.cloud?topic=kafka-topic1&num=1"

  3. 查看日志中的事件。 使用有效的代理程序，主题和访问私钥创建 Kafka 事件预订时，如果您有 Kafka 应用程序生成有关该主题的消息 (例如 kafka-sender-app)，那么您可以在目标 Code Engine 作业的日志中看到接收 Kafka 消息的事件，例如 kafka-receiver-job。 对于使用 curl 发送的每条消息，Kafka 事件将触发相同数目的作业运行。 要查看发送到作业的事件，请使用 ibmcloud ce jobrun logs 命令。

    1. 使用 ibmcloud ce jobrun list 命令列出 kafka-receiver-job 作业的作业运行。
        ibmcloud ce jobrun list --job kafka-receiver-job
    ```
2. 使用 **`ibmcloud ce jobrun logs`** 命令来获取特定作业运行的日志。

```txt {: pre}
    ibmcloud ce jobrun logs -n kafka-receiver-job-abcde
    ```
    示例输出

    ```txt {: screen}
    Getting logs for all instances of job run 'kafka-receiver-job-abcde'...
    Getting jobrun 'kafka-receiver-job-abcde'...
    Getting instances of jobrun 'kafka-receiver-job-abcde'...
    OK
    kafka-receiver-job-abcde-0-0/kafka-receiver-job:
    Hello from helloworld! I'm a batch job! Index: 0
    Hello World from:
    . ___  __  ____  ____
    ./ __)/  \(    \(  __)
    ( (__(  O )) D ( ) _)
    .\___)\__/(____/(____)
    .____  __ _   ___  __  __ _  ____
    (  __)(  ( \ / __)(  )(  ( \(  __)
    .) _) /    /( (_ \ )( /    / ) _)
    (____)\_)__) \___/(__)\_)__)(____)
    Some Env Vars:
    --------------
    CE_DATA=test1: 2
    CE_DOMAIN=us-south.codeengine.appdomain.cloud
    CE_ID=partition:0/offset:249
    CE_JOB=kafka-receiver-job
    CE_JOBRUN=kafka-receiver-job-abcde
    CE_SOURCE=/apis/v1/namespaces/p99k7iy919d/kafkasources/kafkasub-job-ui#kafka-topic1
    CE_SPECVERSION=1.0
    CE_SUBDOMAIN=p99k7iy919d
    CE_SUBJECT=partition:0#249
    CE_TIME=2022-06-21T12:19:24.06Z
    CE_TYPE=dev.knative.kafka.event
    HOME=/root
    HOSTNAME=kafka-receiver-job-abcde-0-0
    JOB_INDEX=0
    KUBERNETES_PORT=tcp://172.21.0.1:443
    KUBERNETES_PORT_443_TCP=tcp://172.21.0.1:443
    KUBERNETES_PORT_443_TCP_ADDR=172.21.0.1
    KUBERNETES_PORT_443_TCP_PORT=443
    KUBERNETES_PORT_443_TCP_PROTO=tcp
    KUBERNETES_SERVICE_HOST=172.21.0.1
    KUBERNETES_SERVICE_PORT=443
    KUBERNETES_SERVICE_PORT_HTTPS=443
    PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
    PWD=/
    SHLVL=1
    z=Set env var 'SHOW' to see all variables
    ```
使用 Kafka 接收方作业 (`icr.io/codeengine/codeengine`) 时，请在接收方作业的日志中搜索 `CE_DATA` 以查看接收到的消息。

请注意，作业运行的日志信息仅持续一个小时。 有关查看应用程序或作业的日志的更多信息，请参阅 [查看日志](/docs/codeengine?topic=codeengine-logging)。

传递到作业的 Kafka 事件的环境变量

传递到作业的所有事件都将作为环境变量接收。 这些环境变量包含前缀 CE_，并且基于 CloudEvents spec。

每个事件都包含一些公共环境变量，每次将事件传递到作业时都会显示这些变量。 每个事件中的实际变量集可以包含更多选项。 有关更多信息，请参阅 CloudEvent 属性。

下表描述了特定于 Kafka 事件的环境变量。

事件环境变量
变量 描述
CE_DATA 事件的数据 (主体)。
CE_DOMAIN URL 应用程序（和项目）的域名部分。
CE_ID 事件的唯一标识，除非事件已重放，在这种情况下，将为其分配相同的标识。
CE_SOURCE 用于指示此事件在事件生产者中的起源位置的 URI 引用。 对于 Kafka 事件，此头采用以下格式: /apis/v1/namespaces/[PROJECT_SUBDOMAIN]/kafkasources/kafkasub#[TOPIC_NAME]。
CE_SPECVERSION CloudEvents 规范的版本。 该值始终为 1.0。
CE_SUBDOMAIN 与应用程序（和项目）相关的 URL 的子域部分。 如果您熟悉 Kubernetes，CE_SUBDOMAIN 将映射到与您的项目相关联的 Kubernetes 命名空间。
CE_SUBJECT Kafka 分区号和偏移量。 例如，partition:0#1 指的是分区 0 和偏移量 1。
CE_TIME 生成该事件的时间。
CE_TYPE 事件类型。 对于 Kafka 事件，这是 dev.knative.kafka.event。

示例输出

CE_DATA={"message":"This is a test message #","message_number":1}
CE_DOMAIN=us-south.codeengine.appdomain.cloud
CE_ID=partition:0/offset:46
CE_SOURCE=/apis/v1/namespaces/ewgz38l13ts/kafkasources/mykafkasubscription-job#kafka-topic1
CE_SPECVERSION=1.0
CE_SUBDOMAIN=ewgz38l13ts
CE_SUBJECT=partition:0#46
CE_TIME=2021-09-27T18:02:17.7Z
CE_TYPE=dev.knative.kafka.event

查看和更新 Kafka 事件预订

您可以查看有关您的 Kafka 活动订阅的详细信息，或更新该订阅。

从控制台查看和更新 Kafka 事件预订

  • 查看有关事件预订的信息

    1. 从 Code Engine "项目" 页面，转至您的项目。
    2. 从“概述”页面，单击 事件预订 以查看已定义的预订的列表。

  • 更新事件预订

    1. 请前往您的订阅页面。 要浏览至预订页面，请转至“事件预订”页面，然后单击要更新的预订的名称。
    2. 更新预订。 例如，将 Kafka 预订的主题更改为其他主题。 从 消息代理详细信息 选项卡中，从“主题”部分中除去现有主题，然后添加新主题的名称。
    3. 单击保存以保存更改。

使用 CLI 查看和更新 Kafka 事件预订

  • 要使用 CLI 查看有关事件预订的信息，请使用 ibmcloud ce subscription kafka get 命令。

    ibmcloud ce sub kafka get -n mykafkasubscription

    示例输出

    Getting Kafka event subscription 'mykafkasubscription'...
OK
Name:               mykafkasubscription
ID:                 abcdefgh-abcd-abcd-abcd-1a2b3c4d5e6f
Project Name:       myproject
Project ID:         01234567-abcd-abcd-abcd-abcdabcd1111
Age:                2m4s
Created:            2022-06-18T16:59:12-04:00
Destination Type:                 app
Destination:                      kafka-receiver-app2
Brokers:
broker-3-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
broker-5-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
broker-0-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
broker-0-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
broker-4-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
broker-0-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
Consumer Group:                   knative-kafka-source-c577b304-dccd-40c8-bb62-138c39f6112a
Topics:
kafka-topic1
Secret key reference (user):      kafka-subscription-secret.username
Secret key reference (password):  kafka-subscription-secret.password
Ready:                            true
Conditions:
Type                     OK    Age  Reason
ConnectionEstablished    true  53m
InitialOffsetsCommitted  true  53m
Ready                    true  52m
Scheduled                true  52m
SinkProvided             true  53m
Events:
Type     Reason           Age                Source                  Messages
Normal   FinalizerUpdate  53m                kafkasource-controller  Updated "mykafkasubscription" finalizers

  • 要使用 CLI 更新事件预订，请使用 ibmcloud ce subscription kafka update 命令。 以下示例更新主题名称。

    ibmcloud ce sub kafka update -n mykafkasubscription --topic kafka-topic2

    您可以使用 ibmcloud ce subscription kafka update 命令来更新 Kafka 订阅的值。 但是，不能使用此命令修改使用者组的值。 如果您想更新订阅以引用另一个主题，请在更新订阅之前确保 Kafka 主题已存在。

删除 Kafka 事件订阅

当您不再需要 Kafka 订阅时，可以将其删除。

删除订阅时，系统会使用 Event Streams 服务实例的服务凭据，将消费者组从 Event Streams 服务实例中移除。 如果在删除订阅时，服务凭据已被删除或已失效，则无法从 Event Streams 服务实例中移除消费者组。 您对 Code Engine Kafka 事件订阅的删除请求失败。

从控制台删除 Kafka 预订

  1. 从 Code Engine "项目" 页面，转至您的项目。
  2. 从“概述”页面，单击 事件预订 以查看已定义的预订的列表。
  3. 从预订列表中，删除要从应用程序或作业中除去的预订。

如果删除与预订关联的应用程序或作业，那么不会删除该预订。 如果重新创建应用程序或作业 (或者具有相同名称的另一个应用程序或作业)，那么您的预订将与该应用程序或作业重新连接。

使用 CLI 删除 Kafka 预订

您可以通过运行 ibmcloud ce subscription kafka delete 命令来删除 Kafka 预订。

例如，使用以下命令来删除名为 mykafkasubscription 的 Kafka 预订。

ibmcloud ce subscription kafka delete --name mykafkasubscription

如果删除与预订关联的应用程序或作业，那么不会删除该预订。 相反，它将进入 false 的就绪状态，因为预订取决于应用程序或作业的可用性。 如果重新创建应用程序或作业 (或具有相同名称的其他应用程序或作业)，那么您的预订将重新连接，并且“就绪”状态为 true。

定义其他事件属性

创建预订时，可以定义要包含在生成的任何事件中的其他 CloudEvent 属性。 这些属性与事件交付中的任何其他 CloudEvent 属性类似。 如果选择指定现有 CloudEvent 属性的名称，那么它将覆盖事件中包含的原始值。

要定义其他属性，请将 --extension 选项与 ibmcloud ce subscription kafka create CLI 命令配合使用。

有关更多信息，请参阅 我可以使用其他 CloudEvents 规范吗?