feat: add Apache Pulsar EventBus implementation #2019
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
- Add Pulsar producer and consumer implementations - Integrate Pulsar support into main EventBus factory - Add configuration constants for Pulsar service URL and JWT token - Update environment configuration to include Pulsar option - Add comprehensive tests for Pulsar implementation - Support all existing EventBus interfaces and options - Use Exclusive subscription mode for better message ordering - Add Docker Compose configurations for Pulsar deployment This implementation allows coze-studio to use Apache Pulsar as a message queue alongside the existing NSQ, Kafka, and RocketMQ options.
- Add Pulsar support to .env.debug.example - Include PULSAR_SERVICE_URL and PULSAR_JWT_TOKEN configuration - Update COZE_MQ_TYPE comment to include pulsar option - Maintain consistency with main .env.example configuration
futeng
requested review from
fanlv,
junwen-lee,
liuyunchao-1998,
lvxinyu-1117,
hi-pender,
shentongmartin,
mrh997,
N3kox and
zhuangjie1125
as code owners
|
|
fanlv
reviewed
Sep 8, 2025
| } | ||
|
|
||
| // Create Pulsar client | ||
| fmt.Printf("[DEBUG] Creating Pulsar client with URL: %s\n", serviceURL) |
fanlv
reviewed
Sep 8, 2025
| @@ -0,0 +1,389 @@ | |||
| name: coze-studio | |||
There was a problem hiding this comment.
Hi @fanlv 您好,
- docker/docker-compose-external-pulsar.yml 这个启动的时候时候不包含 Pulsar 服务，主要用来模拟采用外部 Pulsar 作为 eventbus的场景。
- docker/docker-compose-pulsar.ym 这个是默认启动一个 Pulsar 服务，方便直接使用的场景。
希望都能保留着啊。如果分优先级的话，希望优先保留docker/docker-compose-external-pulsar.yml，这个可以一键启动使用包含 Pulsar 作为 backend eventbus 版本的 coze
…Replace all fmt.Printf/fmt.Println with logs.Debugf/logs.Errorf in producer.go- Use t.Logf/t.Errorf in test files instead of fmt.Println- Maintain consistency with default EventBus implementations (NSQ style)- Ensure proper logging practices across the codebase
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Description
Add Apache Pulsar support as a new EventBus implementation for coze-studio, providing users with a powerful, enterprise-grade message queue option that enhances coze-studio's scalability and security capabilities.
Why Pulsar?
As coze-studio grows and handles larger data volumes, the platform requires a message queue solution that can scale seamlessly while maintaining strong consistency guarantees. Apache Pulsar offers:
Changes
COZE_MQ_TYPE=pulsarConfiguration
Enable Pulsar EventBus
export COZE_MQ_TYPE="pulsar"
export MQ_NAME_SERVER="pulsar:6650"
export PULSAR_SERVICE_URL="pulsar://pulsar:6650"
export PULSAR_JWT_TOKEN="" # Optional JWT token for authentication
Testing
Breaking Changes
None. This is a purely additive feature that maintains full backward compatibility.
What type of PR is this?
feat: A new feature
Check the PR title.
(Optional) Translate the PR title into Chinese.
feat: 添加 Apache Pulsar EventBus 实现
(Optional) More detailed description for this PR(en: English/zh: Chinese).
en: This PR introduces Apache Pulsar as a new message queue backend for coze-studio's EventBus system. Users can now configure COZE_MQ_TYPE=pulsar to use Pulsar instead of NSQ, Kafka, or RocketMQ. The implementation includes JWT authentication support, Docker Compose configurations for easy deployment, and maintains full compatibility with existing EventBus interfaces. The Exclusive subscription mode ensures better message ordering for event processing, making it ideal for production environments requiring high throughput and strong consistency guarantees.
zh(optional): 此 PR 为 coze-studio 的 EventBus 系统引入了 Apache Pulsar 作为新的消息队列后端。用户现在可以通过配置 COZE_MQ_TYPE=pulsar 来使用 Pulsar 替代 NSQ、Kafka 或 RocketMQ。该实现包含 JWT 认证支持、便于部署的 Docker Compose 配置，并与现有 EventBus 接口完全兼容。独占订阅模式确保了事件处理的更好消息顺序，非常适合需要高吞吐量和强一致性保证的生产环境。
(Optional) Which issue(s) this PR fixes:
N/A - This is a new feature enhancement