使用 AMQ Streams Kafka Bridge

流程

1. 如果您还没有这样做，请将 Kafka Bridge 安装存档解压缩到任何目录中。

2. 使用配置属性作为参数运行 Kafka Bridge 脚本：

   例如：

   ./bin/kafka_bridge_run.sh --config-file=<path>/application.properties

3. 检查日志中是否存在安装是否成功。

   HTTP-Kafka Bridge started and listening on port 8080
   HTTP-Kafka Bridge bootstrap servers localhost:9092

流程

1. 使用 Kafka Bridge，为您创建的主题生成三个信息：

   curl -X POST \
     http://localhost:8080/topics/bridge-quickstart-topic \
     -H 'content-type: application/vnd.kafka.json.v2+json' \
     -d '{
       "records": [
           {
               "key": "my-key",
               "value": "sales-lead-0001"
           },
           {
               "value": "sales-lead-0002",
               "partition": 2
           },
           {
               "value": "sales-lead-0003"
           }
       ]
   }'

   • sales-lead-0001 发送到基于密钥哈希的分区。
   • Sales-lead-0002 直接发送到分区 2。
   • Sales-lead-0003 使用循环方法发送到 bridge-quickstart-topic 主题中的一个分区。

2. 如果请求成功，Kafka Bridge 会返回一个 偏移 数组，以及一个 200 代码，以及 application/vnd.kafka.v2+json 的内容类型 标头。对于每个消息，偏移 阵列描述：

   • 消息发送到的分区

   • 分区的当前消息偏移

     响应示例

     #...
     {
       "offsets":[
         {
           "partition":0,
           "offset":0
         },
         {
           "partition":2,
           "offset":0
         },
         {
           "partition":0,
           "offset":1
         }
       ]
     }

流程

1. 在名为 bridge-quickstart-consumer-group 的新消费者组中创建一个 Kafka Bridge 使用者：

   curl -X POST http://localhost:8080/consumers/bridge-quickstart-consumer-group \
     -H 'content-type: application/vnd.kafka.v2+json' \
     -d '{
       "name": "bridge-quickstart-consumer",
       "auto.offset.reset": "earliest",
       "format": "json",
       "enable.auto.commit": false,
       "fetch.min.bytes": 512,
       "consumer.request.timeout.ms": 30000
     }'

   • consumer 名为 bridge-quickstart-consumer，嵌入式数据格式被设置为 json。
   • 定义了一些基本配置设置。

   • 使用者不会自动向日志提交偏移，因为 enable.auto.commit 设置为 false。稍后，您将在此快速入门中手动提交偏移。

     如果请求成功，Kafka Bridge 会在响应正文中返回消费者 ID (instance_id)和基本 URL (base_uri)，以及 200 代码。

     响应示例

     #...
     {
       "instance_id": "bridge-quickstart-consumer",
       "base_uri":"http://<bridge_id>-bridge-service:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer"
     }

2. 复制基础 URL (base_uri)，以便在此快速入门的其他使用者操作中使用。

流程

1. 向 Kafka Bridge consumer 生成额外的信息，如 Produc ing 消息到主题和分区 中所述。

2. 将 GET 请求提交到 记录 端点：

   curl -X GET http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/records \
     -H 'accept: application/vnd.kafka.json.v2+json'

   创建并订阅 Kafka Bridge 消费者后，第一个 GET 请求会返回一个空的响应，因为轮询操作会启动重新平衡过程来分配分区。

3. 重复步骤 2，以从 Kafka Bridge 使用者检索信息。

   Kafka Bridge 返回一条消息重用数组，它包括了主题名称、键、值、分区和偏移挂载到响应正文的信息以及 200 代码。默认情况下，从最新的偏移中检索消息。

   HTTP/1.1 200 OK
   content-type: application/vnd.kafka.json.v2+json
   #...
   [
     {
       "topic":"bridge-quickstart-topic",
       "key":"my-key",
       "value":"sales-lead-0001",
       "partition":0,
       "offset":0
     },
     {
       "topic":"bridge-quickstart-topic",
       "key":null,
       "value":"sales-lead-0003",
       "partition":0,
       "offset":1
     },
   #...

   注意

   如果返回空响应，请向消费者生成更多记录，如 Produc ing 消息到主题和分区 中所述，然后重试检索消息。

流程

1. 对于 quickstart-bridge-topic 主题的分区 0，请参阅特定偏移：

   curl -X POST http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/positions \
     -H 'content-type: application/vnd.kafka.v2+json' \
     -d '{
       "offsets": [
           {
               "topic": "bridge-quickstart-topic",
               "partition": 0,
               "offset": 2
           }
       ]
   }'

   如果请求成功，Kafka Bridge 只会返回一个 204 代码。

2. 将 GET 请求提交到 记录 端点：

   curl -X GET http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/records \
     -H 'accept: application/vnd.kafka.json.v2+json'

   Kafka Bridge 从您想到的偏移返回信息。

3. 通过向同一分区的最后偏移恢复默认消息检索行为。此时，使用 位置/端点。

   curl -X POST http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/positions/end \
     -H 'content-type: application/vnd.kafka.v2+json' \
     -d '{
       "partitions": [
           {
               "topic": "bridge-quickstart-topic",
               "partition": 0
           }
       ]
   }'

   如果请求成功，Kafka Bridge 会返回另一个 204 代码。

流程

1. 编辑 Kafka Bridge 安装存档提供的 application.properties 文件。

   使用属性文件指定 Kafka 和 HTTP 相关的属性。

   1. 配置标准 Kafka 相关属性，包括特定于 Kafka 用户和制作者的属性。

      使用：

      • kafka.bootstrap.servers 用于定义到 Kafka 集群的主机/端口连接
      • kafka.producer.acks 来为 HTTP 客户端提供确认

      • kafka.consumer.auto.offset.reset 以确定如何管理 Kafka 中偏移的重置

        有关配置 Kafka 属性的更多信息，请参阅 Apache Kafka 网站

   2. 配置 HTTP 相关的属性，以启用对 Kafka 集群的 HTTP 访问。

      例如：

      bridge.id=my-bridge
      http.host=0.0.0.0
      http.port=8080 1
      http.cors.enabled=true 2
      http.cors.allowedOrigins=https://strimzi.io 3
      http.cors.allowedMethods=GET,POST,PUT,DELETE,OPTIONS,PATCH 4

      1

      Kafka Bridge 侦听端口 8080 的默认 HTTP 配置。

      2

      设置为 true 以启用 CORS。

      3

      以逗号分隔的 CORS 源列表。您可以使用 URL 或 Java 正则表达式。

      4

      CORS 允许的 HTTP 方法的逗号分隔列表。

2. 保存配置文件。

流程

1. 编辑 Kafka Bridge 安装存档提供的 application.properties 文件。

   使用 bridge.tracing 属性启用您要使用的追踪。

   启用 OpenTelemetry 的示例配置

   #bridge.tracing=jaeger 1
   bridge.tracing=opentelemetry 2

   1

   启用 OpenTracing 的属性（已弃用）。此处保留注释。

   2

   启用 OpenTelemetry 的属性会被取消注释。

   启用追踪后，您可以在运行 Kafka Bridge 脚本时初始化追踪。

2. 保存配置文件。

3. 设置用于追踪的环境变量。

   OpenTelemetry 的环境变量

   OTEL_SERVICE_NAME=my-tracing-service 1
   OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 2

   1

   OpenTelemetry tracer 服务的名称。

   2

   基于 gRPC 的 OTLP 端点，侦听端口 4317。

   OpenTracing 的环境变量

   JAEGER_SERVICE_NAME=my-jaeger-service 1
   JAEGER_AGENT_HOST=localhost 2
   JAEGER_AGENT_PORT=6831 3

   1

   OpenTracing Jaeger tracer 服务的名称。

   2

   用于与侦听 span 的 Jaeger 代理通信的主机名。

   3

   与 Jaeger 代理通信的端口。端口 6831 由 Jaeger 代理公开。

4. 使用为追踪启用的属性运行 Kafka Bridge 脚本：

   在启用了 OpenTelemetry 的情况下运行 Kafka Bridge

   ./bin/kafka_bridge_run.sh --config-file=<path>/application.properties

   Kafka Bridge 的内部消费者和制作者现在已启用用于追踪。