kq: 支持 kafka.Writer BatchTimeout/BatchSize/BatchBytes 配置

[zhuud]

• 新增 WithBatchTimeout 选项，支持配置批量发送超时时间
• 新增 WithBatchSize 选项，支持配置批量发送消息数量
• 新增 WithBatchBytes 选项，支持配置批量发送字节大小
• 补充完整的单元测试，覆盖单个配置和组合配置场景

涉及文件：

• kq/pusher.go: 添加三个配置选项和对应的 PushOption 函数
• kq/pusher_test.go: 添加对应的单元测试用例

Greptile Overview

Updated On: 2025-11-07 08:35:18 UTC

Greptile Summary

This PR adds three new configuration options to the Kafka pusher component to expose native kafka-go writer batch configuration settings. The changes introduce WithBatchTimeout, WithBatchSize, and WithBatchBytes options that allow users to control when message batches are flushed based on time intervals, message counts, or byte sizes respectively. These new options follow the existing functional options pattern used throughout the kq package, maintaining consistency with other configuration functions like WithBalancer and WithChunkSize. The implementation adds the new fields to the pushOptions struct and applies them to the underlying kafka.Writer during initialization, only setting values when they are positive to avoid overriding defaults with zero values.

Important Files Changed

Filename	Score	Overview
kq/pusher.go	4/5	Adds three new batch configuration options with proper validation, but has a potential conflict where syncPush overrides batchSize
kq/pusher_test.go	5/5	Comprehensive test coverage for new batch options including individual, combined, and unit tests

Confidence score: 4/5

• This PR introduces useful Kafka batch configuration options with minimal risk to existing functionality
• Score reduced by one point due to a potential configuration conflict where syncPush unconditionally overrides user-configured batchSize
• Pay close attention to the interaction between WithSyncPush and WithBatchSize options in kq/pusher.go

Sequence Diagram

sequenceDiagram
    participant User
    participant Pusher
    participant kafka.Writer
    participant ChunkExecutor

    User->>Pusher: NewPusher(addrs, topic, opts)
    Pusher->>kafka.Writer: new kafka.Writer
    Note over Pusher: Apply batch options (timeout, size, bytes)
    alt syncPush enabled
        Note over Pusher: Set BatchSize=1, no ChunkExecutor
    else async push (default)
        Pusher->>ChunkExecutor: NewChunkExecutor with options
    end
    Pusher-->>User: return Pusher instance

    User->>Pusher: Push(ctx, value) or PushWithKey(ctx, key, value)
    Note over Pusher: Create kafka.Message with key/value
    Note over Pusher: Inject trace context via OpenTelemetry
    
    alt has ChunkExecutor (async)
        Pusher->>ChunkExecutor: Add(msg, len(value))
        Note over ChunkExecutor: Buffer messages until batch ready
        ChunkExecutor->>kafka.Writer: WriteMessages(ctx, batch...)
    else no ChunkExecutor (sync)
        Pusher->>kafka.Writer: WriteMessages(ctx, msg)
    end
    
    kafka.Writer-->>Pusher: return error/nil
    Pusher-->>User: return error/nil

    User->>Pusher: Close()
    alt has ChunkExecutor
        Pusher->>ChunkExecutor: Flush()
        ChunkExecutor->>kafka.Writer: WriteMessages(ctx, remaining_batch...)
    end
    Pusher->>kafka.Writer: Close()
    kafka.Writer-->>Pusher: return error/nil
    Pusher-->>User: return error/nil

Loading

[greptile-apps]

Additional Comments (1)

1. kq/pusher.go, line 81-83 (link)

   logic: Setting producer.BatchSize = 1 unconditionally overrides any user-configured batchSize when syncPush is enabled. This may not be the intended behavior if a user wants both sync push and a custom batch size. Should syncPush override user-configured batchSize, or should it only be set to 1 if no custom batchSize was provided?

_{2 files reviewed, 1 comment}

_{Edit Code Review Agent Settings | Greptile}