12.2. Kafka Bridge quickstart

流程

1. 打开 application.properties 文件，并检查是否定义了默认的 HTTP 相关设置 ：

   http.enabled=true
   http.host=0.0.0.0
   http.port=8080

   Copy to Clipboard Toggle word wrap

   这会将 Kafka Bridge 配置为侦听端口 8080 上的请求。

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

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

   Copy to Clipboard Toggle word wrap

流程

1. 使用 kafka-topics.sh 实用程序创建 Kafka 主题：

   bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic bridge-quickstart-topic --partitions 3 --replication-factor 1 --config retention.ms=7200000 --config segment.bytes=1073741824

   Copy to Clipboard Toggle word wrap

   指定三个分区。

2. 验证是否已创建该主题：

   bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic bridge-quickstart-topic

   Copy to Clipboard Toggle word wrap

3. 使用 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"
           }
       ]
   }'

   Copy to Clipboard Toggle word wrap
   • sales-lead-0001 发送到基于密钥哈希的分区。
   • Sales-lead-0002 直接发送到分区 2。
   • Sales-lead-0003 使用 round-robin 方法发送到 bridge-quickstart-topic 主题中的分区。

4. 如果请求成功，Kafka Bridge 会返回一个 偏移 数组，以及 200 (OK)代码和 application/vnd.kafka.v2+json 的 content-type 标头。对于每个消息，偏移 数组描述：

   • 消息发送到的分区

   • 分区的当前消息偏移

     响应示例

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

     Copy to Clipboard Toggle word wrap

流程

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
     }'

   Copy to Clipboard Toggle word wrap
   • 消费者名为 bridge-quickstart-consumer，嵌入式数据格式被设置为 json。

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

     注意

     如果您没有在请求正文中指定消费者名称，则 Kafka Bridge 会生成一个随机消费者名称。

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

     响应示例

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

     Copy to Clipboard Toggle word wrap

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'

   Copy to Clipboard Toggle word wrap

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

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

   Kafka Bridge 返回一条信息 数组，它带有 200 (OK)代码，它包括主题名称、键、值、分区和偏移来来包括 200 (OK)代码。默认情况下，消息从最新的偏移中检索。

   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
     },
   #...

   Copy to Clipboard Toggle word wrap
   注意

   如果返回空响应，请生成更多记录到消费者，如 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
           }
       ]
   }'

   Copy to Clipboard Toggle word wrap

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

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'

   Copy to Clipboard Toggle word wrap

   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
           }
       ]
   }'

   Copy to Clipboard Toggle word wrap

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