Refactor kafka integration samples - #133
Conversation
|
Warning Rate limit exceeded
You’ve run out of usage credits. Purchase more in the billing tab. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
📝 WalkthroughWalkthroughThis pull request adds Kafka producer and consumer connector samples for the integration platform. The changes introduce a shared 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ 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}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
integrator-default-profile/connectors/kafka_consumer_connector_sample/functions.bal (1)
3-6: ⚡ Quick winMake
processOrderreturnerror?to preserve explicit failure propagation.At Line 3, returning
error?will let the caller usecheckand keep processing/commit behavior aligned with actual per-message outcome once TODO logic is added.Proposed change
-public isolated function processOrder(OrderEvent orderEvent) { +public isolated function processOrder(OrderEvent orderEvent) returns error? { // TODO: Implement the logic to process the order log:printInfo("Processing order..", orderId = orderEvent.orderId); }And update the call site in
integrator-default-profile/connectors/kafka_consumer_connector_sample/main.bal:- processOrder(orderEvent); // Implement a processing logic under processOrder() method in functions.bal file + check processOrder(orderEvent); // Implement processing logic in functions.bal🤖 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/connectors/kafka_consumer_connector_sample/functions.bal` around lines 3 - 6, Change the processOrder function signature to return error? (public isolated function processOrder(OrderEvent orderEvent) returns error?) so failures can be propagated instead of swallowed; inside processOrder return any errors from the TODO logic (or nil on success). Then update the caller in main.bal that invokes processOrder to use check (or otherwise handle the returned error) so per-message processing/commit semantics reflect actual success/failure.
🤖 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/connectors/kafka_consumer_connector_sample/main.bal`:
- Around line 25-30: The code unconditionally calls caller->commit() after
iterating records, which will commit offsets for failed messages caught in the
"on fail error e" block and prevent redelivery; modify the logic so you only
commit offsets for successfully processed records (e.g., track the highest
successfully processed msg.offset.offset per partition during the loop and call
a targeted commit for those offsets or skip commit entirely if any record
failed), updating the commit call that currently uses caller->commit() to use
per-partition/per-offset commits or conditional commit based on the tracked
lastSuccessfulOffset; ensure references in the "on fail error e" handler
(msg.offset.offset and msg.offset.partition.partition) are used to update
failure state so you can avoid committing failed messages.
---
Nitpick comments:
In
`@integrator-default-profile/connectors/kafka_consumer_connector_sample/functions.bal`:
- Around line 3-6: Change the processOrder function signature to return error?
(public isolated function processOrder(OrderEvent orderEvent) returns error?) so
failures can be propagated instead of swallowed; inside processOrder return any
errors from the TODO logic (or nil on success). Then update the caller in
main.bal that invokes processOrder to use check (or otherwise handle the
returned error) so per-message processing/commit semantics reflect actual
success/failure.
🪄 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
Run ID: bf819f8e-a1c7-48ff-ba4a-84eaa22e6009
📒 Files selected for processing (6)
integrator-default-profile/connectors/kafka_consumer_connector_sample/functions.balintegrator-default-profile/connectors/kafka_consumer_connector_sample/main.balintegrator-default-profile/connectors/kafka_consumer_connector_sample/types.balintegrator-default-profile/connectors/kafka_producer_connector_sample/automation.balintegrator-default-profile/connectors/kafka_producer_connector_sample/config.balintegrator-default-profile/connectors/kafka_producer_connector_sample/types.bal
There was a problem hiding this comment.
Pull request overview
This PR refactors the Kafka producer/consumer connector samples to send and consume a structured OrderEvent payload (JSON), adds configurable topic support on the producer side, and improves message observability by logging message identifiers (orderId) and (on failures) Kafka offset/partition metadata.
Changes:
- Introduces an
OrderEventrecord type and uses it for producer serialization and consumer deserialization. - Makes the producer topic configurable and updates the producer automation flow to publish
OrderEventJSON. - Updates the consumer flow to parse
OrderEvent, invoke aprocessOrder()hook, and log failure context (offset/partition).
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| integrator-default-profile/connectors/kafka_producer_connector_sample/types.bal | Adds OrderEvent record type for producer sample payloads. |
| integrator-default-profile/connectors/kafka_producer_connector_sample/config.bal | Adds configurable kafkaTopic for producer sample. |
| integrator-default-profile/connectors/kafka_producer_connector_sample/automation.bal | Publishes OrderEvent JSON to Kafka and logs the sent orderId. |
| integrator-default-profile/connectors/kafka_consumer_connector_sample/types.bal | Adds public OrderEvent record type for consumer deserialization. |
| integrator-default-profile/connectors/kafka_consumer_connector_sample/main.bal | Implements listener-based consumption, parses OrderEvent, logs offsets/partitions on failure, and commits manually. |
| integrator-default-profile/connectors/kafka_consumer_connector_sample/functions.bal | Adds processOrder() hook used by the consumer service. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Purpose
This PR refactors the publisher and consumer flows of the Kafka Connector to improve maintainability and error tracking.
Goals
Approach
User stories
Release note
Documentation
Training
Certification
Marketing
Automation tests
Security checks
Samples
Related PRs
Migrations (if applicable)
Test environment
Learning