Add event-driven architecture integration samples - #147
Open
gayaldassanayake wants to merge 4 commits into
Open
Conversation
Six runnable Ballerina samples covering common EDA patterns under integrator-default-profile/event-integration/: - pubsub_kafka: Pub/Sub with Kafka - cqrs_kafka: CQRS with Kafka and PostgreSQL - outbox_cdc_rabbitmq: Transactional Outbox with PostgreSQL CDC and RabbitMQ - competing_consumers_rabbitmq: Competing Consumers with RabbitMQ - saga_solace: Saga choreography with Solace PubSub+ - dlq_asb: Dead-Letter Queue and Retry with Azure Service Bus Each sample follows the WSO2 Integrator UI project structure (config.bal, connections.bal, main.bal) and uses named listeners. All samples are version 1.0.0 and include a Deploy to Devant button in their README.
Contributor
|
Warning Review limit reached
Next review available in: 59 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (13)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 18
🧹 Nitpick comments (1)
integrator-default-profile/event-integration/pubsub_kafka/main.bal (1)
16-16: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse a named HTTP listener.
The HTTP listener is currently anonymous, while the PR objective requires named listeners. Declare
httpListenerseparately and attach the service to it, matchingkafkaListener.Proposed refactor
-service /publish on new http:Listener(8090) { +listener http:Listener httpListener = new (8090); + +service /publish on httpListener {🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@integrator-default-profile/event-integration/pubsub_kafka/main.bal` at line 16, Update the HTTP service declaration to use a named listener: declare an http:Listener variable named httpListener separately on port 8090, then attach the service /publish to httpListener, matching the existing kafkaListener pattern.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In
`@integrator-default-profile/event-integration/competing_consumers_rabbitmq/config.bal`:
- Around line 3-4: Update the configurable defaults for rabbitmqUser and
rabbitmqPassword so deployment configurations do not ship with guest
credentials; retain guest/guest only through an explicitly local-only
configuration and require other environments to supply both values via
secret-backed configuration.
In
`@integrator-default-profile/event-integration/competing_consumers_rabbitmq/main.bal`:
- Line 13: Make the HTTP listener in main.bal configurable instead of
hard-coding port 8090, using the configuration origin and preserving 8090 as the
default; update README.md lines 18-20 to document the supported --http.port
override syntax for launching additional instances.
- Around line 23-29: Update the rabbitmq:ServiceConfig annotation for the
rabbitmq:Service to set autoAck to false, then explicitly acknowledge each
delivery with basicAck only after onMessage completes thumbnail generation
successfully. Handle delivery failures and acknowledgement call errors so
unsuccessful processing is not marked complete, using the service’s existing
RabbitMQ listener and message-handling symbols.
- Line 11: Update the rabbitmqListener initialization to create a
rabbitmq:ConnectionConfiguration using rabbitmqUser and rabbitmqPassword, then
pass that configuration through the listener constructor so consumers
authenticate with the same credentials as the publisher.
In `@integrator-default-profile/event-integration/cqrs_kafka/main.bal`:
- Around line 31-44: Update onConsumerRecord so each match/event is deduplicated
using its unique id before modifying leaderboard totals. Add or reuse a
uniqueness constraint/table and perform the idempotency check atomically with
the leaderboard upsert, ensuring Kafka redelivery skips already-processed events
while new events increment totals exactly once.
- Around line 24-29: Align the endpoint with the documented /matches/record path
by renaming the resource function add in
integrator-default-profile/event-integration/cqrs_kafka/main.bal (lines 24-29)
to record. No direct change is needed in
integrator-default-profile/event-integration/cqrs_kafka/README.md (line 7),
since the implementation will match its existing documentation and curl example.
In `@integrator-default-profile/event-integration/dlq_asb/main.bal`:
- Around line 26-35: Update the notification sender’s send call to serialize
job.toJsonString() as byte[] and set contentType to asb:JSON, matching the
byte-array JSON parsing in onMessage. Keep the existing consumer flow and
settlement behavior unchanged.
- Around line 17-22: Update the asbListener initialization to propagate the
fallible asb:Listener constructor result by using Ballerina’s check mechanism
with new, preserving the existing listener configuration.
In `@integrator-default-profile/event-integration/dlq_asb/README.md`:
- Around line 3-20: Update the README’s prerequisites and setup instructions for
the notifications queue to explicitly provision it with maxDeliveryCount set to
5, or revise the documented down-path behavior to match the queue’s actual
default configuration. Ensure the documented queue settings and the sample’s
auto-DLQ behavior are consistent.
In `@integrator-default-profile/event-integration/outbox_cdc_rabbitmq/config.bal`:
- Around line 6-8: Add configurable rabbitmqUsername, rabbitmqPassword, and
rabbitmqVirtualHost settings in config.bal alongside the existing RabbitMQ
settings, then update the RabbitMQ client initialization in connections.bal to
pass all three through a rabbitmq:ConnectionConfiguration when creating
rabbitmq:Client.
In `@integrator-default-profile/event-integration/outbox_cdc_rabbitmq/main.bal`:
- Line 42: Update the registration info log in the affected event handler to
remove the raw user.email field, while retaining the “User registered” message
and user.id identifier.
- Around line 21-28: Configure the postgresqlCdcListener declaration to set
options.eventProcessingFailureHandlingMode to cdc:FAIL, ensuring errors returned
by onCreate publishMessage halt CDC processing for retry. Validate offset
behavior with at least one outbox row and confirm a broker publish failure does
not advance the offset or skip that row.
- Line 46: Update the CDC ServiceConfig annotation to derive the qualified table
filter from the configurable dbName value instead of hardcoding accounts.
Preserve the public.outbox schema/table suffix so overrides keep the listener
and filter aligned.
In `@integrator-default-profile/event-integration/outbox_cdc_rabbitmq/README.md`:
- Line 3: Update the outbox CDC RabbitMQ sample to provision a durable queue
bound to the user-events exchange for UserRegistered messages and publish them
as persistent messages, ensuring the documented lossless-delivery guarantee is
backed by the sample’s topology. If the implementation cannot manage that queue
and persistence, reword the README guarantee and document the required external
queue topology instead.
In `@integrator-default-profile/event-integration/pubsub_kafka/main.bal`:
- Around line 11-14: Update the kafkaListener configuration to set offsetReset
to kafka:OFFSET_RESET_EARLIEST, preserving the existing groupId and topics
settings so fresh consumer groups process previously published events.
In `@integrator-default-profile/event-integration/saga_solace/config.bal`:
- Around line 1-4: Remove the default admin values from the configurable Solace
credentials in config.bal, requiring them to be injected for deployed
environments; if local defaults are retained, explicitly enforce that the
configuration is local-only before deployment. Update the associated README
prerequisites to explain the required credential injection or local-only
restriction.
In `@integrator-default-profile/event-integration/saga_solace/README.md`:
- Around line 30-33: Update the “Inject compensation” curl example in the README
to call the exposed POST /trips/book endpoint and set the request destination to
"fail" so it triggers compensation; do not document the nonexistent
/trips/failCar endpoint.
- Around line 14-20: Update the Solace queue setup block in the README to create
and subscribe the missing car-on-hotel-booked queue to trip/hotel/booked,
matching the queue configured by main.bal while preserving the existing
provisioning commands.
---
Nitpick comments:
In `@integrator-default-profile/event-integration/pubsub_kafka/main.bal`:
- Line 16: Update the HTTP service declaration to use a named listener: declare
an http:Listener variable named httpListener separately on port 8090, then
attach the service /publish to httpListener, matching the existing kafkaListener
pattern.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 5c310199-993f-49e3-8aac-3bf15c785750
📒 Files selected for processing (36)
integrator-default-profile/event-integration/competing_consumers_rabbitmq/Ballerina.tomlintegrator-default-profile/event-integration/competing_consumers_rabbitmq/Dependencies.tomlintegrator-default-profile/event-integration/competing_consumers_rabbitmq/README.mdintegrator-default-profile/event-integration/competing_consumers_rabbitmq/config.balintegrator-default-profile/event-integration/competing_consumers_rabbitmq/connections.balintegrator-default-profile/event-integration/competing_consumers_rabbitmq/main.balintegrator-default-profile/event-integration/cqrs_kafka/Ballerina.tomlintegrator-default-profile/event-integration/cqrs_kafka/Dependencies.tomlintegrator-default-profile/event-integration/cqrs_kafka/README.mdintegrator-default-profile/event-integration/cqrs_kafka/config.balintegrator-default-profile/event-integration/cqrs_kafka/connections.balintegrator-default-profile/event-integration/cqrs_kafka/main.balintegrator-default-profile/event-integration/dlq_asb/Ballerina.tomlintegrator-default-profile/event-integration/dlq_asb/Dependencies.tomlintegrator-default-profile/event-integration/dlq_asb/README.mdintegrator-default-profile/event-integration/dlq_asb/config.balintegrator-default-profile/event-integration/dlq_asb/connections.balintegrator-default-profile/event-integration/dlq_asb/main.balintegrator-default-profile/event-integration/outbox_cdc_rabbitmq/Ballerina.tomlintegrator-default-profile/event-integration/outbox_cdc_rabbitmq/Dependencies.tomlintegrator-default-profile/event-integration/outbox_cdc_rabbitmq/README.mdintegrator-default-profile/event-integration/outbox_cdc_rabbitmq/config.balintegrator-default-profile/event-integration/outbox_cdc_rabbitmq/connections.balintegrator-default-profile/event-integration/outbox_cdc_rabbitmq/main.balintegrator-default-profile/event-integration/pubsub_kafka/Ballerina.tomlintegrator-default-profile/event-integration/pubsub_kafka/Dependencies.tomlintegrator-default-profile/event-integration/pubsub_kafka/README.mdintegrator-default-profile/event-integration/pubsub_kafka/config.balintegrator-default-profile/event-integration/pubsub_kafka/connections.balintegrator-default-profile/event-integration/pubsub_kafka/main.balintegrator-default-profile/event-integration/saga_solace/Ballerina.tomlintegrator-default-profile/event-integration/saga_solace/Dependencies.tomlintegrator-default-profile/event-integration/saga_solace/README.mdintegrator-default-profile/event-integration/saga_solace/config.balintegrator-default-profile/event-integration/saga_solace/connections.balintegrator-default-profile/event-integration/saga_solace/main.bal
- pubsub_kafka: add offsetReset EARLIEST to kafka listener; extract named http listener - cqrs_kafka: fix README endpoint path /matches/record → /matches/add - saga_solace: add missing car-on-hotel-booked queue to setup block; fix compensation curl (was /trips/failCar, now /trips/book with destination=fail) - outbox_cdc_rabbitmq: remove email field from user registration log - competing_consumers_rabbitmq: make HTTP port configurable (enables documented second-instance launch); propagate credentials to RabbitMQ listener
5 tasks
gayaldassanayake
marked this pull request as ready for review
July 27, 2026 04:12
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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
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.
Summary
integrator-default-profile/event-integration/covering common EDA patternsconfig.bal,connections.bal,main.bal) with named listeners and version1.0.0pubsub_kafkacqrs_kafkaoutbox_cdc_rabbitmqcompeting_consumers_rabbitmqsaga_solacedlq_asbTest plan
bal buildpasses cleanly for all six samplespubsub_kafka: POST to/publish/pageview→ log showsPageViewed event publishedandUpdating recommendationscqrs_kafka: POST match results → GET/leaderboard/topreturns correct aggregated scoresoutbox_cdc_rabbitmq: POST/users/register→ CDC captures outbox row → RabbitMQ receives routed eventcompeting_consumers_rabbitmq: POST resize jobs → queue distributes to consumer instancessaga_solace:destination=Paris→ full saga completes;destination=fail→ compensation chain firesdlq_asb:recipient=alice→ complete;recipient=down→ abandon ×5 → auto-DLQ;recipient=invalid→ immediate dead-letter