Make KafkaProcess Async #76
Conversation
WalkthroughThe changes introduce asynchronous lifecycle management for Kafka processing components, update related service and test logic to support async patterns, and simplify the example application to a minimal topic handler that logs and throws an exception. Additionally, a new Docker Compose configuration is added for orchestrating a Redpanda Kafka cluster with a console UI. A new logging method and consumer close state handling are also introduced. Changes
Sequence Diagram(s)sequenceDiagram
participant User
participant KafkaService
participant KafkaProcess
User->>KafkaService: StartAsync()
loop For each process
KafkaService->>KafkaProcess: Start(cancellationToken) (awaited)
end
User->>KafkaService: StopAsync()
loop For each process
KafkaService->>KafkaProcess: Stop() (awaited)
end
sequenceDiagram
participant Kafka
participant Program
Kafka->>Program: Message on "test" topic
Program->>Program: Log message value
Program->>Program: Throw exception
Poem
Note ⚡️ AI Code Reviews for VS Code, Cursor, WindsurfCodeRabbit now has a plugin for VS Code, Cursor and Windsurf. This brings AI code reviews directly in the code editor. Each commit is reviewed immediately, finding bugs before the PR is raised. Seamless context handoff to your AI code agent ensures that you can easily incorporate review feedback. Note ⚡️ Faster reviews with cachingCodeRabbit now supports caching for code and dependencies, helping speed up reviews. This means quicker feedback, reduced wait times, and a smoother review experience overall. Cached data is encrypted and stored securely. This feature will be automatically enabled for all accounts on May 30th. To opt out, configure ✨ Finishing Touches
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. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (2)
examples/Examples/Program.cs (1)
22-28: Consider making the handler async for consistencyThe handler is throwing an exception synchronously, which could be inconsistent with the rest of the asynchronous implementation. Consider making it async for better alignment with the overall async pattern.
-app.MapTopic("test", (KafkaContext context) => { +app.MapTopic("test", async (KafkaContext context) => { Console.WriteLine("Test topic received message: " + context.Value); throw new Exception("Test exception"); });tools/docker-compose.yml (1)
65-71: Consider documenting the commented volume driver optionsThe commented volume driver options suggest there might be an alternative configuration for persistent storage, but it's not clear why they're commented out or when they should be used.
volumes: redpanda: driver: local - #driver_opts: - #o: bind - #type: rw - #device: /redpanda_data + # Uncomment the following options to use a bind mount instead of a Docker volume + # driver_opts: + # o: bind + # type: rw + # device: /path/to/your/redpanda_data
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
Cache: Disabled due to data retention organization setting
Knowledge Base: Disabled due to data retention organization setting
📒 Files selected for processing (6)
examples/Examples/Program.cs(1 hunks)src/MinimalKafka/KafkaProcess.cs(4 hunks)src/MinimalKafka/KafkaService.cs(1 hunks)test/MinimalKafka.Tests/KafkaProcessTests.cs(1 hunks)test/MinimalKafka.Tests/KafkaServiceTests.cs(2 hunks)tools/docker-compose.yml(1 hunks)
🧰 Additional context used
🧬 Code Graph Analysis (1)
src/MinimalKafka/KafkaService.cs (1)
src/MinimalKafka/KafkaProcess.cs (4)
Task(6-6)Task(7-7)Task(32-64)Task(66-70)
🔇 Additional comments (7)
test/MinimalKafka.Tests/KafkaProcessTests.cs (1)
141-143: Proper async exception handling implementedThe test has been correctly updated to use async/await pattern for testing the now-asynchronous
Startmethod. The lambda is properly declared as async and the assertion correctly usesThrowAsyncfor asynchronous exception verification.test/MinimalKafka.Tests/KafkaServiceTests.cs (2)
49-49: Correctly modified to await the process start verificationThe test assertion now properly awaits the asynchronous
Startmethod call verification, which aligns with the updated interface that returns aTask.
66-66: Correctly modified to await the process stop verificationThe test assertion now properly awaits the asynchronous
Stopmethod call verification, which aligns with the updated interface that returns aTask.src/MinimalKafka/KafkaProcess.cs (3)
6-7: Interface updated correctly to support async patternThe interface changes properly transition from synchronous to asynchronous method signatures by returning
Taskinstead ofvoid, aligning with modern .NET async patterns.
32-32: Implementation properly converted to async methodThe
Startmethod is correctly marked asasyncand returnsTask, matching the interface specification.
50-50: Handler now awaited correctlyThe handler invocation is now properly awaited, which ensures that exceptions thrown during asynchronous execution will be properly propagated and caught in the catch block.
tools/docker-compose.yml (1)
1-60: Docker Compose configuration looks good for Redpanda setupThe Docker Compose configuration properly sets up a Redpanda instance with console UI, using appropriate network configurations and port mappings. This will provide a suitable Kafka-compatible environment for testing the asynchronous Kafka process implementation.
There was a problem hiding this comment.
Actionable comments posted: 1
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
Cache: Disabled due to data retention organization setting
Knowledge Base: Disabled due to data retention organization setting
📒 Files selected for processing (6)
src/MinimalKafka/Helpers/Logging.cs(1 hunks)src/MinimalKafka/KafkaConsumer.cs(2 hunks)src/MinimalKafka/KafkaDelegateFactory.cs(1 hunks)src/MinimalKafka/KafkaProcess.cs(4 hunks)src/MinimalKafka/KafkaService.cs(1 hunks)test/MinimalKafka.Tests/KafkaProcessTests.cs(1 hunks)
✅ Files skipped from review due to trivial changes (1)
- src/MinimalKafka/Helpers/Logging.cs
🚧 Files skipped from review as they are similar to previous changes (3)
- src/MinimalKafka/KafkaService.cs
- src/MinimalKafka/KafkaProcess.cs
- test/MinimalKafka.Tests/KafkaProcessTests.cs
⏰ Context from checks skipped due to timeout of 90000ms (1)
- GitHub Check: Analyze
🔇 Additional comments (2)
src/MinimalKafka/KafkaConsumer.cs (2)
55-55: Good addition of flag to track consumer state.Adding the
_isClosedboolean flag provides a mechanism to track the consumer's state, which helps prevent multiple close operations.
90-96: Excellent implementation to prevent duplicate consumer disposal.The check for
_isClosedprevents potential issues from attempting to close the consumer multiple times, which could happen in asynchronous contexts. This complements the PR's goal of making KafkaProcess asynchronous by ensuring proper resource cleanup.
This fixes the issue when an exception is thrown it is ignored
Summary by CodeRabbit
New Features
Refactor
Tests
Chores