EasyKafka

https://github.com/studeyang/easykafka
Admin
EasyKafka is a lightweight enhancement framework based on Spring Kafka that simplifies message...
Tokens:12,431
Snippets:94
Trust Score:7.7
License:MIT
Update:5 months ago
Show doc for...
Tokens:
Raw
### Full EasyKafka Configuration Example

Source: https://github.com/studeyang/easykafka/blob/master/README.md

A comprehensive example demonstrating the integration of EasyKafka configurations within a Spring Boot application. This includes Kafka cluster definitions, producer and consumer specific settings, and runtime configurations.

```yaml
spring:
  application:
    name: order-service

easykafka:
  init:
    kafkaCluster:
      - cluster: order
        brokers: order-kafka.example.com:9092
        tag: BASE
      - cluster: order
        brokers: order-kafka-gray.example.com:9092
        tag: GRAY
      - cluster: payment
        brokers: pay-kafka.example.com:9092
    producer:
      - beanName: orderProducer
        config:
          retries: 3
          batch.size: 32768
          linger.ms: 5
          acks: all
    consumer:
      - beanName: paymentConsumer
        cluster: payment
        config:
          session.timeout.ms: 30000

  runtime:
    producer:
      partitionSize: 1000
      async:
        corePoolSize: 5
        maxPoolSize: 10
        keepAliveSeconds: 120
        queueCapacity: 500
    consumer:
      groupIdPrefix: "order-service-"
```

--------------------------------

### Configure Kafka Cluster

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Configure your Kafka cluster connection details in the application's YAML file. This example sets up a cluster named 'send'.

```yaml
easykafka:
  init:
    kafkaCluster:
      - cluster: send
        brokers: kafka.example.com:9092
```

--------------------------------

### EasyKafka Module Structure

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Outlines the directory structure of the EasyKafka project, detailing the core client module, Spring Boot starter, and example applications for producers and consumers.

```text
easykafka/
├── easykafka-client/             # 核心模块：生产者/消费者逻辑与注解
├── easykafka-spring-boot-starter/ # Spring Boot 自动配置入口
├── example-sdk/                  # 消息定义包（模拟真实 SDK jar）
├── example-producer/             # 生产者示例应用
└── example-consumer/             # 消费者示例应用
```

--------------------------------

### Automatic Retry Mechanism Example

Source: https://context7.com/studeyang/easykafka/llms.txt

Demonstrates the automatic retry mechanism for message sending failures. The framework retries up to 3 times with exponential backoff. The `onFail` callback is invoked only after all retries are exhausted.

```java
// 重试参数（RetryableSender 硬编码）：
// - 最大重试次数：3 次
// - 初始等待时间：100ms
// - 退避倍数：2x（100ms → 200ms → 400ms，上限 5000ms）

// 验证重试行为的示例（消费端故意使处理抛异常，触发生产端重试观测）
OrderCreatedEvent event = new OrderCreatedEvent();
event.setOrderId("ORD-RETRY-TEST");

EventPublisher.publish(event, new MessageCallback() {
    @Override
    public void onSuccess() {
        // 重试成功后仅调用一次
        log.info("消息最终发送成功（可能经历重试）");
    }

    @Override
    public void onFail(Exception e) {
        // 3 次重试全部失败后调用
        log.error("消息发送最终失败，需人工介入", e);
        // 写入补偿表 / 触发告警
        compensationService.save(event, e);
    }
});

```

--------------------------------

### Maven Build and Test Commands

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Common Maven commands for building, testing, and packaging EasyKafka modules. Includes instructions for running all tests, specific module tests, and tests for individual classes or methods.

```bash
# 构建所有模块
mvn clean install

# 运行所有测试
mvn clean test

# 运行指定模块的测试
mvn clean test -f easykafka-client/pom.xml

# 运行指定测试类
mvn test -Dtest=JsonUtilsTest

# 运行指定测试方法
mvn test -Dtest=JsonUtilsTest#someMethod

# 打包（CI 环境使用）
mvn -B package --file pom.xml
```

--------------------------------

### Maven Build and Test Commands

Source: https://context7.com/studeyang/easykafka/llms.txt

Common Maven commands for building, testing, and packaging the EasyKafka project. Includes commands for running all tests, specific modules, test classes, or test methods, and for CI builds.

```bash
# 构建所有模块
mvn clean install

# 运行所有测试
mvn clean test

# 运行指定模块的测试
mvn clean test -f easykafka-client/pom.xml

# 运行指定测试类
mvn test -Dtest=JsonUtilsTest

# 运行指定测试方法
mvn test -Dtest=JsonUtilsTest#someMethod

# CI 打包
mvn -B package --file pom.xml

```

--------------------------------

### Add Consumer Dependencies

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Include the EasyKafka starter and the shared message definition package for consumers.

```xml
<!-- easykafka -->
<dependency>
    <groupId>io.github.studeyang</groupId>
    <artifactId>easykafka-spring-boot-starter</artifactId>
    <version>${easykafka.version}</version>
</dependency>

<!-- 消息定义包（与生产者共享） -->
<dependency>
    <groupId>io.github.studeyang</groupId>
    <artifactId>example-sdk</artifactId>
    <version>${sdk.version}</version>
</dependency>
```

--------------------------------

### Include EasyKafka Starter Dependency

Source: https://context7.com/studeyang/easykafka/llms.txt

Add the EasyKafka Spring Boot Starter to your project's pom.xml to enable automatic configuration and bean registration. Consumer applications also need to include the shared message definition package.

```xml
<!-- pom.xml -->
<dependency>
    <groupId>io.github.studeyang</groupId>
    <artifactId>easykafka-spring-boot-starter</artifactId>
    <version>${easykafka.version}</version>
</dependency>

<!-- 消费方还需引入消息定义包（与生产方共享） -->
<dependency>
    <groupId>io.github.studeyang</groupId>
    <artifactId>example-sdk</artifactId>
    <version>${sdk.version}</version>
</dependency>
```

--------------------------------

### Add EasyKafka Dependency

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Include the EasyKafka Spring Boot starter dependency in your project to use its features.

```xml
<dependency>
    <groupId>io.github.studeyang</groupId>
    <artifactId>easykafka-spring-boot-starter</artifactId>
    <version>${easykafka.version}</version>
</dependency>
```

--------------------------------

### Producer Runtime Configuration

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Configure producer runtime settings like partition size and asynchronous thread pool parameters. Ensure thread pool settings are appropriate for your application's load.

```yaml
easykafka:
  runtime:
    producer:
      partitionSize: 500           # 批量发送时的分区大小
      async:
        corePoolSize: 3            # 核心线程数
        maxPoolSize: 5             # 最大线程数
        keepAliveSeconds: 60       # 空闲线程存活时间（秒）
        queueCapacity: 100         # 队列容量
        rejectedHandler: java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy
        threadNamePrefix: kafka-async-producer-
```

--------------------------------

### Publishing Messages with Callbacks

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Illustrates how to use EventPublisher to send messages and handle success or failure asynchronously using callbacks.

```APIDOC
## Sending Messages with Callbacks

`EventPublisher.publish` supports callbacks for asynchronous execution upon successful or failed message sending.

### Full Callback (MessageCallback)

```java
EventPublisher.publish(event, new MessageCallback() {
    @Override
    public void onSuccess() {
        log.info("Message sent successfully");
    }

    @Override
    public void onFail(Exception e) {
        log.error("Message sending failed", e);
        // Compensating operations can be performed here
    }
});
```

### Success Only Callback (SuccessCallback)

```java
EventPublisher.publish(event, new SuccessCallback() {
    @Override
    public void onSuccess() {
        log.info("Message sent successfully");
    }
});
```

### Failure Only Callback (FailCallback)

```java
EventPublisher.publish(event, new FailCallback() {
    @Override
    public void onFail(Exception e) {
        log.error("Message sending failed, performing alert", e);
    }
});
```

### Batch Sending

```java
List<OrderCreatedEvent> events = buildEvents();
EventPublisher.publish(events);
```
```

--------------------------------

### Generic Publishing with ObjectPublisher

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Explains how to use ObjectPublisher for sending messages that do not inherit from Event or when a dynamic Topic needs to be specified.

```APIDOC
## Generic Publishing (ObjectPublisher)

Use `ObjectPublisher` when the message object does not inherit from `Event` or when you need to dynamically specify the Topic.

### Sending to a Specific Topic

```java
ObjectPublisher.publish(myObject, new TopicMetadata("send", "custom-topic"));
```

### Sending with Full Metadata

```java
MessageMetadata metadata = new MessageMetadataBuilder()
    .topicMetadata("send", "custom-topic")
    .messageKey("key-123")
    .messageHeader("requestId", "req-456")
    .messageTag(Tag.GRAY)
    .build();

ObjectPublisher.publish(myObject, metadata);
```

### Sending with Callbacks

```java
ObjectPublisher.publish(myObject, metadata, new MessageCallback() {
    public void onSuccess() { ... }
    public void onFail(Exception e) { ... }
});
```
```

--------------------------------

### Send Message using EventPublisher

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Send a message by creating an instance of your event class, setting its properties, and calling `EventPublisher.publish()`.

```java
import io.github.open.easykafka.client.EventPublisher;

OrderCreatedEvent event = new OrderCreatedEvent();
event.setOrderId("ORD-123");
event.setUserId("USER-456");

EventPublisher.publish(event);
```

--------------------------------

### Configure Kafka Clusters in application.yml

Source: https://context7.com/studeyang/easykafka/llms.txt

Define Kafka cluster information in application.yml under `easykafka.init.kafkaCluster`. This configuration automatically registers Producer Beans and Listener Container Factories. Each cluster entry requires a logical identifier (`cluster`), broker addresses (`brokers`), and an optional environment tag (`tag`).

```yaml
# application.yml
spring:
  application:
    name: order-service

easykafka:
  init:
    kafkaCluster:
      - cluster: order          # 逻辑集群标识，与 @Topic / @EventHandler 中的 cluster 对应
        brokers: order-kafka.example.com:9092
        tag: BASE               # BASE（默认）= 基线环境
      - cluster: order
        brokers: order-kafka-gray.example.com:9092
        tag: GRAY               # GRAY = 灰度环境；未配置时框架自动复制 BASE 配置
      - cluster: payment
        brokers: pay-kafka.example.com:9092

  runtime:
    producer:
      partitionSize: 1000
      async:
        corePoolSize: 5
        maxPoolSize: 10
        keepAliveSeconds: 120
        queueCapacity: 500
        threadNamePrefix: kafka-async-producer-
    consumer:
      groupIdPrefix: "order-service-"  # 最终 groupId = groupIdPrefix + topicName
```

--------------------------------

### Multi-Cluster Configuration for EasyKafka

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Configure EasyKafka to connect to multiple Kafka clusters simultaneously by defining them in the application's YAML configuration.

```yaml
easykafka:
  init:
    kafkaCluster:
      - cluster: order     # 订单集群
        brokers: order-kafka.example.com:9092
      - cluster: payment   # 支付集群
        brokers: pay-kafka.example.com:9092
```

--------------------------------

### Cluster Configuration Reference

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Provides a reference for configuring Kafka clusters, including logical cluster names, broker addresses, and environment tags.

```APIDOC
## Cluster Configuration Reference

### Cluster Configuration

```yaml
easykafka:
  init:
    kafkaCluster:
      - cluster: send              # Logical cluster name (corresponds to @Topic/@EventHandler)
        brokers: 127.0.0.1:9092    # Broker addresses, comma-separated
        tag: BASE                  # Tag: BASE (default) or GRAY
```

| Property | Type | Required | Default Value | Description |
|---|---|---|---|---|
| `cluster` | String | Yes | - | Logical cluster identifier. |
| `brokers` | String | Yes | - | List of Broker addresses in `host:port` format. |
| `tag` | Tag | No | BASE | Environment tag (BASE / GRAY). |
```

--------------------------------

### Multi-Cluster and Gray Routing

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Details EasyKafka's support for multiple Kafka clusters within a single application and its built-in gray routing capabilities.

```APIDOC
## Multi-Cluster and Gray Routing

EasyKafka supports integrating multiple Kafka clusters in the same application and provides built-in gray routing capabilities.

### Multi-Cluster Configuration

```yaml
easykafka:
  init:
    kafkaCluster:
      - cluster: order     # Order cluster
        brokers: order-kafka.example.com:9092
      - cluster: payment   # Payment cluster
        brokers: pay-kafka.example.com:9092
```

### Gray Routing Configuration

Configure gray nodes with `tag: gray` for a cluster. The framework automatically routes based on the current environment.

```yaml
easykafka:
  init:
    kafkaCluster:
      - cluster: send
        brokers: kafka-prod.example.com:9092       # Baseline environment
      - cluster: send
        brokers: kafka-gray.example.com:9092
        tag: gray                                   # Gray environment
```

> **Note**: If no gray nodes are configured, the framework duplicates the baseline configuration for gray, which does not affect normal startup.

### Manually Specifying Routing Tag

```java
// Force routing to the gray cluster
MessageMetadata metadata = new MessageMetadataBuilder()
    .topicMetadata("send", "order-topic")
    .messageTag(Tag.GRAY)
    .build();

ObjectPublisher.publish(event, metadata);
```
```

--------------------------------

### Implement Full Send Callback with MessageCallback

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Provide a complete callback implementation using MessageCallback to handle both successful and failed message sends asynchronously.

```java
EventPublisher.publish(event, new MessageCallback() {
    @Override
    public void onSuccess() {
        log.info("消息发送成功");
    }

    @Override
    public void onFail(Exception e) {
        log.error("消息发送失败", e);
        // 可在此处做补偿操作
    }
});
```

--------------------------------

### Multi-Cluster and Gray Routing Configuration

Source: https://context7.com/studeyang/easykafka/llms.txt

Configure multiple Kafka clusters for an application, using 'tag: GRAY' for gray nodes. The framework automatically selects the target cluster based on the current environment. If no gray nodes are configured, it defaults to the baseline configuration.

```yaml
# application.yml — 多集群 + 灰度路由示例
easykafka:
  init:
    kafkaCluster:
      # 订单集群 - 基线
      - cluster: order
        brokers: order-kafka.example.com:9092
        tag: BASE
      # 订单集群 - 灰度
      - cluster: order
        brokers: order-kafka-gray.example.com:9092
        tag: GRAY
      # 支付集群（仅基线，框架自动补充灰度配置）
      - cluster: payment
        brokers: pay-kafka.example.com:9092

```

--------------------------------

### Generic Sending with ObjectPublisher using Full Metadata

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Send objects with detailed metadata, including topic, message key, headers, and tags, using ObjectPublisher and MessageMetadataBuilder.

```java
MessageMetadata metadata = new MessageMetadataBuilder()
    .topicMetadata("send", "custom-topic")
    .messageKey("key-123")
    .messageHeader("requestId", "req-456")
    .messageTag(Tag.GRAY)
    .build();

ObjectPublisher.publish(myObject, metadata);
```

--------------------------------

### Send Single Domain Event using EventPublisher

Source: https://context7.com/studeyang/easykafka/llms.txt

The `EventPublisher.publish(Event)` method provides a static facade for asynchronously sending `Event` messages. The framework automatically populates essential metadata such as `messageId`, `messageKey`, `messageTag`, `messageTopic`, and `messageCreateTime` based on annotations and current context.

```java
import io.github.open.easykafka.client.EventPublisher;

// 构造消息对象
OrderCreatedEvent event = new OrderCreatedEvent();
event.setOrderId("ORD-20250601-001");
event.setUserId("USER-888");
event.setAmount(299.99);
event.setRequestId("REQ-abc123");
event.setRetryCount(0);

// 发送（异步，无回调）
EventPublisher.publish(event);
// 框架自动：
//   messageId        = UUID
//   messageKey       = event.getOrderId()（@MessageKey 字段）
//   messageHeader    = {requestId: "REQ-abc123", retryCount: 0}（@MessageHeader 字段）
//   messageTopic     = {cluster: "order", name: "order-created-topic"}
//   messageTag       = 当前环境（BASE / GRAY）自动选择
//   messageCreateTime = 发送时间
```

--------------------------------

### EasyKafka Producer Send Flow

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Illustrates the sequence of operations when publishing an event using EasyKafka's producer. Highlights the use of an asynchronous thread pool, message conversion, and a sender chain with retry and reporting capabilities.

```text
EventPublisher.publish(event)
    │
    ▼
DefaultMessagePublisher（异步线程池）
    │
    ▼
SendMessageConverter（Event → SendMessage）
    │
    ▼
Sender 责任链（装饰器模式）
    ├── ReportableSender（记录发送结果日志）
    ├── RetryableSender（失败重试，最多 3 次，指数退避）
    └── ProducerSender（构建 ProducerRecord，调用 Kafka）
    │
    ▼
ProducerContainer.getProducer(cluster, tag)
    │
    ▼
StringKafkaProducer → Kafka Broker
    │
    ▼
MessageCallback.onSuccess() / onFail()
```

--------------------------------

### Cluster Configuration Reference

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Defines the configuration parameters for connecting to Kafka clusters, including cluster name, broker addresses, and environment tags.

```yaml
easykafka:
  init:
    kafkaCluster:
      - cluster: send              # 逻辑集群名称（与 @Topic/@EventHandler 对应）
        brokers: 127.0.0.1:9092    # Broker 地址，多个用逗号分隔
        tag: BASE                  # 标签：BASE（默认）或 GRAY
```

--------------------------------

### Gray Routing Configuration for Kafka Clusters

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Set up gray routing by configuring gray nodes within your Kafka cluster definitions. The framework automatically routes based on the current environment.

```yaml
easykafka:
  init:
    kafkaCluster:
      - cluster: send
        brokers: kafka-prod.example.com:9092       # 基线环境
      - cluster: send
        brokers: kafka-gray.example.com:9092
        tag: gray                                   # 灰度环境
```

--------------------------------

### Publish Event with Callbacks

Source: https://context7.com/studeyang/easykafka/llms.txt

Publish a single event with asynchronous callbacks for success and failure. Supports MessageCallback, SuccessCallback, and FailCallback interfaces.

```java
import io.github.open.easykafka.client.EventPublisher;
import io.github.open.easykafka.client.producer.callback.MessageCallback;
import io.github.open.easykafka.client.producer.callback.SuccessCallback;
import io.github.open.easykafka.client.producer.callback.FailCallback;

OrderCreatedEvent event = new OrderCreatedEvent();
event.setOrderId("ORD-20250601-002");

// 完整回调（成功 + 失败）
EventPublisher.publish(event, new MessageCallback() {
    @Override
    public void onSuccess() {
        log.info("[OrderCreatedEvent] 消息发送成功，orderId={}", event.getOrderId());
    }

    @Override
    public void onFail(Exception e) {
        log.error("[OrderCreatedEvent] 消息发送失败，orderId={}", event.getOrderId(), e);
        // 可触发告警、写入补偿表等操作
    }
});

// 仅关注成功
EventPublisher.publish(event, (SuccessCallback) () ->
    log.info("发送成功"));

// 仅关注失败
EventPublisher.publish(event, (FailCallback) e ->
    log.error("发送失败，进行告警", e));
```

--------------------------------

### Publish Generic Object with Topic Metadata

Source: https://context7.com/studeyang/easykafka/llms.txt

Publish generic objects (not extending Event) or when dynamic topic specification is needed. Non-String objects are JSON serialized. Use this for simple object publishing.

```java
import io.github.open.easykafka.client.ObjectPublisher;
import io.github.open.easykafka.client.model.TopicMetadata;
import java.util.Arrays;
import java.util.List;

// 发送 List 类型消息（EventPublisher 不支持）
List<String> payload = Arrays.asList("item-1", "item-2", "item-3");
ObjectPublisher.publish(payload, new TopicMetadata("order", "order-sync-topic"));
// 实际发送的 Kafka value: ["item-1","item-2","item-3"]
```

--------------------------------

### Message Headers with @MessageHeader and @Header

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Demonstrates how to use @MessageHeader to add fields to Kafka message headers and @Header to retrieve them in the consumer.

```APIDOC
## Message Headers

### Producer Side
Annotate fields in your message class with `@MessageHeader` to have their values written to Kafka Record Headers.

```java
@Topic(cluster = "send", name = "order-topic")
public class OrderCreatedEvent extends Event {
    @MessageHeader
    private String requestId;

    @MessageHeader
    private String traceId;
}
```

### Consumer Side
Use the `@Header` annotation in your Kafka handler method to retrieve values from message headers.

```java
@KafkaHandler
public void handle(OrderCreatedEvent event,
                   @Header(value = "requestId", required = false) String requestId,
                   @Header(value = "traceId", required = false) String traceId) {
    log.info("requestId={}, traceId={}", requestId, traceId);
}
```
```

--------------------------------

### Implement Success-Only Send Callback with SuccessCallback

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Use SuccessCallback when you only need to be notified upon successful message delivery.

```java
EventPublisher.publish(event, new SuccessCallback() {
    @Override
    public void onSuccess() {
        log.info("消息发送成功");
    }
});
```

--------------------------------

### Publish Generic Object with Full Message Metadata

Source: https://context7.com/studeyang/easykafka/llms.txt

Publish generic objects with detailed message metadata, including Key, Headers, and routing tags (BASE/GRAY), using MessageMetadataBuilder for precise control. Includes an asynchronous callback.

```java
import io.github.open.easykafka.client.ObjectPublisher;
import io.github.open.easykafka.client.model.MessageMetadata;
import io.github.open.easykafka.client.model.MessageMetadataBuilder;
import io.github.open.easykafka.client.model.Tag;

String message = "{\"source\":\"legacy-system\",\"data\":\"...\" }";

MessageMetadata metadata = new MessageMetadataBuilder()
        .topicMetadata("order", "order-sync-topic")  // 指定集群和 Topic
        .messageKey("ORD-20250601-099")               // Kafka Record Key
        .messageHeader("requestId", "REQ-xyz")        // Header（String 类型）
        .messageHeader("retryCount", 0)               // Header（Integer 类型）
        .messageTag(Tag.BASE)                         // 强制路由到基线集群
        .build();

ObjectPublisher.publish(message, metadata, new MessageCallback() {
    @Override
    public void onSuccess() {
        log.info("legacy 消息同步成功");
    }

    @Override
    public void onFail(Exception e) {
        log.error("legacy 消息同步失败", e);
    }
});
```

--------------------------------

### EasyKafka Consumer Flow

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Describes the process of consuming messages with EasyKafka. Covers annotation discovery, listener container factory creation, and message routing to specific handler methods based on message type.

```text
Kafka Broker
    │
    ▼
@EventHandler（标注在消费者类上）
    │
    ▼
EventHandlerAnnotationBeanPostProcessor
（发现 Handler，注册元数据到 ListenerContainer）
    │
    ▼
KafkaListenerContainerFactoryRegistrar
（为每个集群创建 ConcurrentKafkaListenerContainerFactory
  命名规则：{cluster}KafkaListenerContainerFactory
  AckMode：RECORD，自定义反序列化器）
    │
    ▼
@KafkaHandler 方法（根据消息类型自动路由到对应方法）
```

--------------------------------

### Force Routing to Gray Cluster

Source: https://context7.com/studeyang/easykafka/llms.txt

Use ObjectPublisher with MessageMetadataBuilder to explicitly send messages to a gray cluster. This is useful for validating gray releases. Ensure MessageMetadata is built correctly with the desired topic and message tag.

```java
// 强制路由到灰度集群（适用于灰度发布验证）
import io.github.open.easykafka.client.ObjectPublisher;
import io.github.open.easykafka.client.model.MessageMetadataBuilder;
import io.github.open.easykafka.client.model.Tag;

MessageMetadata grayMetadata = new MessageMetadataBuilder()
        .topicMetadata("order", "order-created-topic")
        .messageTag(Tag.GRAY)   // 强制发往灰度集群
        .build();

ObjectPublisher.publish(orderCreatedEvent, grayMetadata);

```

--------------------------------

### Generic Sending with ObjectPublisher and Callback

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Combine generic sending with callbacks for asynchronous handling of send results when using ObjectPublisher.

```java
ObjectPublisher.publish(myObject, metadata, new MessageCallback() {
    public void onSuccess() { ... }
    public void onFail(Exception e) { ... }
});
```

--------------------------------

### Batch Sending of Events

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Efficiently send multiple events in a single batch operation using EventPublisher.publish with a list of events.

```java
List<OrderCreatedEvent> events = buildEvents();
EventPublisher.publish(events);
```

--------------------------------

### Consumer Runtime Configuration

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Set a prefix for consumer group IDs to help organize and identify consumers within your application. The final group ID is formed by concatenating this prefix with the topic name.

```yaml
easykafka:
  runtime:
    consumer:
      groupIdPrefix: "my-service-" # 消费者组 ID 前缀
```

--------------------------------

### Kafka Event Handler with @EventHandler

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Shows how to declare a class as a Kafka message processor using the @EventHandler annotation, which is a wrapper around @KafkaListener.

```APIDOC
## Kafka Event Handler

Annotate a consumer class with `@EventHandler` to declare it as a Kafka message processor. This annotation wraps the underlying `@KafkaListener`.

```java
@Service
@EventHandler(
    cluster = "send",
    topics = "order-topic",
    concurrency = "3"
)
public class OrderEventHandler {

    @KafkaHandler
    public void handleCreated(OrderCreatedEvent event) { ... }

    @KafkaHandler
    public void handleCancelled(OrderCancelledEvent event) { ... }
}
```

### @EventHandler Properties

| Property | Type | Required | Default Value | Description |
|---|---|---|---|---|
| `cluster` | String | Yes | - | The Kafka cluster name to consume from. |
| `topics` | String | Yes | - | The Topic name to subscribe to. |
| `groupId` | String | No | Auto-generated | The consumer group ID. Defaults to `{groupIdPrefix}{topics}`. |
| `concurrency` | String | No | "1" | The number of concurrent consumer threads. |
| `id` | String | No | Auto-generated | The Listener container ID. |
```

--------------------------------

### Implement Failure-Only Send Callback with FailCallback

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Implement FailCallback to handle only message sending failures, allowing for error logging or compensation logic.

```java
EventPublisher.publish(event, new FailCallback() {
    @Override
    public void onFail(Exception e) {
        log.error("消息发送失败，进行告警", e);
    }
});
```

--------------------------------

### EventPublisher.publish(Event, MessageCallback)

Source: https://context7.com/studeyang/easykafka/llms.txt

Sends an event and triggers a callback asynchronously upon completion. Supports MessageCallback (success+failure), SuccessCallback (success only), and FailCallback (failure only).

```APIDOC
## `EventPublisher.publish(Event, MessageCallback)` — Send with Callback

Sends a message and triggers a callback in an asynchronous thread after sending. Supports three types of interfaces: `MessageCallback` (success+failure), `SuccessCallback` (success only), and `FailCallback` (failure only).

### Method
```java
EventPublisher.publish(event, callback);
```

### Parameters
- **event** (Event) - The event object to publish.
- **callback** (MessageCallback | SuccessCallback | FailCallback) - The callback to execute upon message sending.

### Request Example
```java
// Full callback (success + failure)
EventPublisher.publish(event, new MessageCallback() {
    @Override
    public void onSuccess() {
        log.info("[OrderCreatedEvent] Message sent successfully, orderId={}", event.getOrderId());
    }

    @Override
    public void onFail(Exception e) {
        log.error("[OrderCreatedEvent] Message sending failed, orderId={}", event.getOrderId(), e);
        // Trigger alerts, write compensation tables, etc.
    }
});

// Focus only on success
EventPublisher.publish(event, (SuccessCallback) () ->
    log.info("Sent successfully"));

// Focus only on failure
EventPublisher.publish(event, (FailCallback) e ->
    log.error("Sending failed, triggering alert", e));
```
```

--------------------------------

### ObjectPublisher.publish(Object, TopicMetadata)

Source: https://context7.com/studeyang/easykafka/llms.txt

Publishes generic objects that do not inherit from Event, or when a dynamic Topic needs to be specified. Non-String objects are JSON serialized.

```APIDOC
## `ObjectPublisher.publish(Object, TopicMetadata)` — Generic Object Send (Simple Mode)

Use `ObjectPublisher` when the message object does not inherit from `Event` (e.g., `List`, plain POJO, `String`) or when the target Topic needs to be dynamically specified. Non-String type messages will be JSON serialized first.

### Method
```java
ObjectPublisher.publish(object, topicMetadata);
```

### Parameters
- **object** (Object) - The object to publish.
- **topicMetadata** (TopicMetadata) - Metadata specifying the target topic.

### Request Example
```java
// Send List type message (not supported by EventPublisher)
List<String> payload = Arrays.asList("item-1", "item-2", "item-3");
ObjectPublisher.publish(payload, new TopicMetadata("order", "order-sync-topic"));
// Actual Kafka value sent: ["item-1","item-2","item-3"]
```
```

--------------------------------

### Implement Message Handler

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Implement a message handler by creating a Spring service annotated with `@EventHandler`, specifying the cluster and topics to subscribe to. Use `@KafkaHandler` to define methods for specific message types.

```java
import io.github.open.easykafka.client.annotation.EventHandler;
import org.springframework.kafka.annotation.KafkaHandler;
import org.springframework.stereotype.Service;

@Service
@EventHandler(cluster = "send", topics = "order-topic")
public class OrderEventHandler {

    @KafkaHandler
    public void handle(OrderCreatedEvent event) {
        System.out.println("收到订单创建事件: " + event.getOrderId());
    }
}
```

--------------------------------

### ObjectPublisher.publish(Object, MessageMetadata)

Source: https://context7.com/studeyang/easykafka/llms.txt

Publishes generic objects with full metadata control using MessageMetadataBuilder for Key, Header, and routing tags.

```APIDOC
## `ObjectPublisher.publish(Object, MessageMetadata)` — Generic Object Send (Full Metadata)

Control the Key, Header, and routing tags (BASE / GRAY) precisely by chaining metadata using `MessageMetadataBuilder`.

### Method
```java
ObjectPublisher.publish(object, metadata, callback);
```

### Parameters
- **object** (Object) - The object to publish.
- **metadata** (MessageMetadata) - The metadata for the message.
- **callback** (MessageCallback) - Optional callback for success/failure.

### Request Example
```java
String message = "{\"source\":\"legacy-system\",\"data\":\"...\" }";

MessageMetadata metadata = new MessageMetadataBuilder()
        .topicMetadata("order", "order-sync-topic")  // Specify cluster and Topic
        .messageKey("ORD-20250601-099")               // Kafka Record Key
        .messageHeader("requestId", "REQ-xyz")        // Header (String type)
        .messageHeader("retryCount", 0)               // Header (Integer type)
        .messageTag(Tag.BASE)                         // Force route to baseline cluster
        .build();

ObjectPublisher.publish(message, metadata, new MessageCallback() {
    @Override
    public void onSuccess() {
        log.info("Legacy message sync successful");
    }

    @Override
    public void onFail(Exception e) {
        log.error("Legacy message sync failed", e);
    }
});
```
```

--------------------------------

### Batch Publish Events

Source: https://context7.com/studeyang/easykafka/llms.txt

Publish multiple events in a single call, suitable for scenarios where events are batched before notifying downstream systems. Each message is retried independently.

```java
import io.github.open.easykafka.client.EventPublisher;
import java.util.ArrayList;
import java.util.List;

List<OrderCreatedEvent> events = new ArrayList<>();
for (int i = 1; i <= 5; i++) {
    OrderCreatedEvent e = new OrderCreatedEvent();
    e.setOrderId("ORD-" + i);
    e.setUserId("USER-" + i);
    events.add(e);
}

// 批量异步发送，每条消息独立重试
EventPublisher.publish(events);
```

--------------------------------

### Manually Specify Routing Tag for Messages

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Force message routing to a specific cluster environment, such as the gray cluster, by setting the message tag using MessageMetadataBuilder.

```java
// 强制路由到灰度集群
MessageMetadata metadata = new MessageMetadataBuilder()
    .topicMetadata("send", "order-topic")
    .messageTag(Tag.GRAY)
    .build();

ObjectPublisher.publish(event, metadata);
```

--------------------------------

### Generic Sending with ObjectPublisher to a Specific Topic

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Utilize ObjectPublisher to send objects that do not extend Event or when a dynamic Topic needs to be specified.

```java
ObjectPublisher.publish(myObject, new TopicMetadata("send", "custom-topic"));
```

--------------------------------

### Define Message Class with @Topic

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Define your message class by extending `Event` and annotating it with `@Topic` to specify the Kafka cluster and topic name.

```java
import io.github.open.easykafka.client.annotation.Topic;
import io.github.open.easykafka.client.message.Event;

@Topic(cluster = "send", name = "order-topic")
public class OrderCreatedEvent extends Event {
    private String orderId;
    private String userId;
    // getter/setter ...
}
```

--------------------------------

### Define Message Key with @MessageKey

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Annotate a field in your message class with `@MessageKey` to designate it as the Kafka Record Key. This ensures messages with the same key are routed to the same partition.

```java
@Topic(cluster = "send", name = "order-topic")
public class OrderCreatedEvent extends Event {
    @MessageKey
    private String orderId;  // 同一 orderId 的消息路由到同一分区
}
```

--------------------------------

### EventPublisher.publish(Collection<Event>)

Source: https://context7.com/studeyang/easykafka/llms.txt

Publishes multiple events of the same or different types in a single call. Suitable for scenarios where events are batched and then notified to downstream systems.

```APIDOC
## `EventPublisher.publish(Collection<Event>)` — Batch Send

Sends multiple `Event` messages of the same or different types in one call, suitable for scenarios where events are batched before notifying downstream systems.

### Method
```java
EventPublisher.publish(events);
```

### Parameters
- **events** (Collection<Event>) - A collection of event objects to publish.

### Request Example
```java
List<OrderCreatedEvent> events = new ArrayList<>();
for (int i = 1; i <= 5; i++) {
    OrderCreatedEvent e = new OrderCreatedEvent();
    e.setOrderId("ORD-" + i);
    e.setUserId("USER-" + i);
    events.add(e);
}

// Batch asynchronous send, each message retried independently
EventPublisher.publish(events);
```
```

--------------------------------

### Specify Message Key with @MessageKey Annotation

Source: https://context7.com/studeyang/easykafka/llms.txt

Annotate a field with `@MessageKey` to designate its value as the Kafka Record Key. This ensures that messages with the same key are consistently routed to the same partition, preserving order for related messages.

```java
import io.github.open.easykafka.client.annotation.Topic;
import io.github.open.easykafka.client.annotation.MessageKey;
import io.github.open.easykafka.client.message.Event;

@Topic(cluster = "payment", name = "payment-topic")
public class PaymentCompletedEvent extends Event {

    @MessageKey
    private String paymentId;    // 相同 paymentId 的消息始终路由到同一 Kafka 分区

    private String orderId;
    private Double amount;
    private String status;
}
```

--------------------------------

### @EventHandler Annotation

Source: https://context7.com/studeyang/easykafka/llms.txt

Registers a consumer for specific Kafka topics within a cluster. This annotation wraps `@KafkaListener` and automatically configures the necessary Kafka listener container factory and consumer group.

```APIDOC
## `@EventHandler` Annotation — Register Consumer

Annotate a Spring Bean class to declare it as a message handler for a specified cluster and topic. This internally wraps `@KafkaListener`, and the framework automatically creates a `ConcurrentKafkaListenerContainerFactory` and registers the consumer group.

### Usage
```java
@Service
@EventHandler(cluster = "clusterName", topics = "topicName")
public class MyEventHandler implements IEventHandler {
    // ... handler methods ...
}
```

### Parameters
- **cluster** (string) - The Kafka cluster name.
- **topics** (string[]) - An array of topic names to subscribe to.
- **concurrency** (string, optional) - The number of concurrent consumer threads. Defaults to 1.
- **groupId** (string, optional) - The consumer group ID. If not provided, it's auto-generated as `groupIdPrefix + topicName`.

### Handler Method
Annotate methods within the handler class with `@KafkaHandler` to process specific message types. You can also inject headers using `@Header`.

### Example
```java
import io.github.open.easykafka.client.annotation.EventHandler;
import io.github.open.easykafka.client.message.IEventHandler;
import org.springframework.kafka.annotation.KafkaHandler;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;
import java.nio.ByteBuffer;
import java.util.Optional;

@Slf4j
@Service
@EventHandler(
    cluster      = "order",
    topics       = "order-created-topic",
    concurrency  = "3"           // Concurrent consumer threads, defaults to 1
    // groupId   = "my-group"    // If not provided, auto-generated: groupIdPrefix + topicName
)
public class OrderCreatedEventHandler implements IEventHandler {

    // Handle OrderCreatedEvent type messages
    @KafkaHandler
    public void handleOrderCreated(OrderCreatedEvent event,
                                   @Header(value = "requestId", required = false) String requestId,
                                   @Header(value = "retryCount", required = false) ByteBuffer retryCount) {
        int retry = Optional.ofNullable(retryCount).map(ByteBuffer::getInt).orElse(0);
        log.info("Received order creation event orderId={}, requestId={}, retryCount={}",
                 event.getOrderId(), requestId, retry);
        // Business logic...
    }

    // Handle another message type for the same Topic within the same Handler class
    @KafkaHandler
    public void handleOrderCancelled(OrderCancelledEvent event) {
        log.info("Received order cancellation event orderId={}", event.getOrderId());
    }
}
```
```

--------------------------------

### Define Message Topic with @Topic Annotation

Source: https://context7.com/studeyang/easykafka/llms.txt

Use the `@Topic` annotation on message classes to specify the Kafka cluster and topic name. This metadata is automatically used by the framework to construct `ProducerRecord` instances during message sending.

```java
import io.github.open.easykafka.client.annotation.Topic;
import io.github.open.easykafka.client.annotation.MessageKey;
import io.github.open.easykafka.client.annotation.MessageHeader;
import io.github.open.easykafka.client.message.Event;

// 定义一条领域事件消息
@Topic(cluster = "order", name = "order-created-topic")
public class OrderCreatedEvent extends Event {

    @MessageKey                  // 该字段值作为 Kafka Record Key，相同 key 路由到同一分区
    private String orderId;

    @MessageHeader               // 该字段值写入 Kafka Record Header，消费端可通过 @Header 读取
    private String requestId;

    @MessageHeader
    private String traceId;

    private String userId;
    private Double amount;

    // getter / setter ...
}
```

--------------------------------

### Declare Kafka Message Handler with @EventHandler

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Use the @EventHandler annotation on a consumer class to declare it as a Kafka message processor. This annotation is a wrapper around @KafkaListener.

```java
@Service
@EventHandler(
    cluster = "send",
    topics = "order-topic",
    concurrency = "3"
)
public class OrderEventHandler {

    @KafkaHandler
    public void handleCreated(OrderCreatedEvent event) { ... }

    @KafkaHandler
    public void handleCancelled(OrderCancelledEvent event) { ... }
}
```

--------------------------------

### Automatic Retries on Send Failure

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Describes the automatic retry mechanism for message sending failures, managed by the framework without requiring business code intervention.

```APIDOC
## Automatic Retries

The framework automatically retries on send failures without intervention from business code:

| Parameter | Value |
|---|---|
| Maximum Retries | 3 times |
| Initial Wait Time | 100ms |
| Backoff Multiplier | 2x |
| Maximum Wait Time | 5000ms |
| Retry Strategy | Exponential Backoff |
```

--------------------------------

### Define Message Headers with @MessageHeader

Source: https://github.com/studeyang/easykafka/blob/master/README.md

Annotate fields in your event class with @MessageHeader to have their values written to Kafka Record Headers. Consumers can retrieve these values using the @Header annotation.

```java
@Topic(cluster = "send", name = "order-topic")
public class OrderCreatedEvent extends Event {
    @MessageHeader
    private String requestId;

    @MessageHeader
    private String traceId;
}
```

```java
@KafkaHandler
public void handle(OrderCreatedEvent event,
                   @Header(value = "requestId", required = false) String requestId,
                   @Header(value = "traceId", required = false) String traceId) {
    log.info("requestId={}, traceId={}", requestId, traceId);
}
```

--------------------------------

### Register Event Handler with @EventHandler

Source: https://context7.com/studeyang/easykafka/llms.txt

Annotate a Spring Bean class to declare it as a message handler for specific clusters and topics. This annotation wraps @KafkaListener and automatically configures the consumer factory and group ID.

```java
import io.github.open.easykafka.client.annotation.EventHandler;
import io.github.open.easykafka.client.message.IEventHandler;
import org.springframework.kafka.annotation.KafkaHandler;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;
import java.nio.ByteBuffer;
import java.util.Optional;

@Slf4j
@Service
@EventHandler(
    cluster      = "order",
    topics       = "order-created-topic",
    concurrency  = "3"           // 并发消费线程数，默认 1
    // groupId   = "my-group"    // 不填则自动生成：groupIdPrefix + topicName
)
public class OrderCreatedEventHandler implements IEventHandler {

    // 处理 OrderCreatedEvent 类型消息
    @KafkaHandler
    public void handleOrderCreated(OrderCreatedEvent event,
                                   @Header(value = "requestId", required = false) String requestId,
                                   @Header(value = "retryCount", required = false) ByteBuffer retryCount) {
        int retry = Optional.ofNullable(retryCount).map(ByteBuffer::getInt).orElse(0);
        log.info("收到订单创建事件 orderId={}, requestId={}, retryCount={}",
                 event.getOrderId(), requestId, retry);
        // 业务处理逻辑...
    }

    // 同一 Handler 类处理同 Topic 下的另一种消息类型
    @KafkaHandler
    public void handleOrderCancelled(OrderCancelledEvent event) {
        log.info("收到订单取消事件 orderId={}", event.getOrderId());
    }
}
```

--------------------------------

### Add Message Headers with @MessageHeader Annotation

Source: https://context7.com/studeyang/easykafka/llms.txt

Use the `@MessageHeader` annotation on message fields to include their values in the Kafka Record Headers. Supported types include `String` and `Integer`. Consumers can access these header values using the `@Header` parameter in their handler methods.

```java
import io.github.open.easykafka.client.annotation.Topic;
import io.github.open.easykafka.client.annotation.MessageHeader;
import io.github.open.easykafka.client.message.Event;

@Topic(cluster = "order", name = "order-created-topic")
public class OrderCreatedEvent extends Event {

    @MessageHeader
    private String requestId;   // 写入 Header "requestId"

    @MessageHeader
    private Integer retryCount; // 写入 Header "retryCount"

    private String orderId;
}
```
Tokens:
Raw