wgarunap/generic-kafka-event-producer

通用Kafka事件生产者

镜像概述和主要用途

该应用提供REST端点用于发布Kafka消息，支持多种键和值的编码格式（String、AVRO、JSON、MessagePack）。用户可配置消息的主题、头信息（任意数量的键值对）、键（Kafka消息键）和值（Kafka消息 payload），适用于需要灵活生产Kafka事件的场景。

核心功能和特性

• 支持多种编码格式：键和值可使用String、AVRO、JSON、MessagePack编码
• 灵活的消息配置：可自定义主题、头信息、键和值
• REST API接口：提供HTTP端点用于消息发布及相关查询
• Schema Registry集成：支持通过Schema Registry获取AVRO事件 schema
• 主题与Schema查询：可获取Kafka主题列表及Schema Registry中的主题列表

使用场景和适用范围

• 开发测试环境中快速生成Kafka消息
• 需要多种编码格式的Kafka事件生产场景
• 集成Schema Registry的AVRO格式消息生产
• 需通过HTTP接口灵活控制Kafka消息生产的业务场景

使用方法和配置说明

镜像获取与运行

构建镜像

sh
docker build -f Dockerfile -t generic-producer .

拉取镜像

sh
docker pull wgarunap/generic-kafka-event-producer:v1.2.0

运行容器

sh
docker run -dit -e SCHEMAREG_URL=http://schemareg.event.com:8081 -e KAFKA_BROKERS=kafka-1:9092,kafka-2:9092 -p 8000:8000 wgarunap/generic-kafka-event-producer:v1.2.0

配置参数（环境变量）

消息发布接口

AVRO格式消息发布

• 请求URL：<HOST>:<PORT>/publish
• 请求方法：POST
• 请求体示例：

json
{
   "topic":"test",
   "headers":{
      "key1":"value1",
      "key2":"value2"
   },
   "key":{
      "subject":"com.event.KeySchemaSubject", // 仅key.format=avro时必填
      "version":1, // 仅key.format=avro时必填
      "format":"avro", // 可选，默认string，支持avro/json/string/bytes/messagepack
      "body":{
         "event_id":"123e4567-e89b-12d3-a456-426655440000"
      }
   },
   "value":{
      "subject":"com.event.EventName", // 仅value.format=avro时必填
      "version":2, // 仅value.format=avro时必填
      "format":"avro", // 可选，默认string，支持avro/json/string/bytes/messagepack
      "body":{
         "meta":{
            "event_id":"123e4567-e89b-12d3-a456-426655440000",
            "trace_id":"123e4567-e89b-12d3-a456-426655440000",
            "account_id":"123e4567-e89b-12d3-a456-426655440000"
         }
      }
   }
}

JSON格式消息发布

• 请求URL：<HOST>:<PORT>/publish
• 请求方法：POST
• 请求体示例：

json
{
   "topic":"test",
   "headers":{
      "account_id":"123e4567-e89b-12d3-a456-426655440000"
   },
   "key":{
      "format":"string", // 可选，默认string
      "body": "test"
   },
   "value":{
      "format":"json",
      "body":{
         "meta":{
            "event_id":"123e4567-e89b-12d3-a456-426655440000",
            "trace_id":"123e4567-e89b-12d3-a456-426655440000",
            "account_id":"123e4567-e89b-12d3-a456-426655440000"
         }
      }
   }
}

空值消息发布

• 请求URL：<HOST>:<PORT>/publish
• 请求方法：POST
• 请求体示例：

json
{
   "topic":"test",
   "headers":{
      "subject":"com.event.EventName",
      "account_id":"123e4567-e89b-12d3-a456-426655440000"
   },
   "key":{
      "format":"string", // 可选
      "body":"test"
   }
}

查询接口

获取Kafka主题列表

• 请求URL：<HOST>:<PORT>/topics
• 请求方法：GET
• 示例命令：

sh
curl -XGET "<HOST>:<PORT>/topics"

获取Schema Registry主题列表

• 请求URL：<HOST>:<PORT>/avrosubjects
• 请求方法：GET
• 示例命令：

sh
curl -XGET "<HOST>:<PORT>/avrosubjects"

待办功能

• 增加以下键/值编码方式：Protobuf
• 添加认证层