Conversation
WalkthroughKafka health check now requires both a URL and a topic; index.js passes both values. Kafka service signature changed to accept a dynamic topic and uses it throughout. Added Kafka validation in config. Package version bumped 0.0.3 → 0.0.4 and minor formatting/style tweaks elsewhere. Changes
Sequence Diagram(s)sequenceDiagram
autonumber
participant Client
participant HealthCheck as health-check/index.js
participant KafkaSvc as services/kafka.js
participant KafkaCluster as Kafka
Client->>HealthCheck: GET /health
HealthCheck->>KafkaSvc: check(kafkaUrl, topicName)
KafkaSvc->>KafkaCluster: connect
KafkaSvc->>KafkaCluster: ensureTopicExists(topicName)
KafkaSvc->>KafkaCluster: produce(test message → topicName)
KafkaCluster-->>KafkaSvc: message consumed (or timeout)
KafkaSvc->>KafkaCluster: cleanup (disconnect)
KafkaSvc-->>HealthCheck: result (healthy/unhealthy)
HealthCheck-->>Client: HTTP 200/503 with details
rect rgba(230,245,255,0.5)
note over HealthCheck,KafkaSvc: Changed: topicName provided dynamically to KafkaSvc.check
end
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes Poem
Tip 🔌 Remote MCP (Model Context Protocol) integration is now available!Pro plan users can now connect to remote MCP servers from the Integrations page. Connect with popular remote MCPs such as Notion and Linear to add more context to your reviews and chats. 📜 Recent review detailsConfiguration used: CodeRabbit UI 💡 Knowledge Base configuration:
You can enable these sources in your CodeRabbit configuration. 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
✨ Finishing Touches
🧪 Generate unit tests
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. CodeRabbit Commands (Invoked using PR/Issue comments)Type 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: 1> [!CAUTION]
Some comments are outside the diff and can’t be posted inline due to platform limits.
🔭 Outside diff range comments (1)
health-check/services/kafka.js (1)
49-127: Flaky flow: producing before consuming risks missing the message. Add input validation and single-settle guard; create the consumer before sending.Current order creates the consumer after the message is produced, so with
fromOffset: falsethe consumer can miss the produced message. Also, there’s no guard against multiple resolves, and the Promise executor is markedasyncunnecessarily. The patch below addresses all three:
- Validate
topicNameupfront.- Remove
asyncfrom the Promise executor.- Create the consumer before sending to ensure it sees the message.
- Add a single-settle guard and consistent cleanup.
-async function check(kafkaUrl,topicName) { - return new Promise(async (resolve) => { +function check(kafkaUrl, topicName) { + return new Promise((resolve) => { - console.log(`[Kafka Health Check] Connecting to Kafka at ${kafkaUrl}`); - const client = new kafka.KafkaClient({ kafkaHost: kafkaUrl }); + if (!topicName || typeof topicName !== 'string' || topicName.trim().length === 0) { + console.error('[Kafka Health Check] Missing `topicName`'); + return resolve(false); + } + + console.log(`[Kafka Health Check] Connecting to Kafka at ${kafkaUrl}`); + const client = new kafka.KafkaClient({ kafkaHost: kafkaUrl }); - try { - await ensureTopicExists(client, topicName); - } catch (err) { - client.close(); - return resolve(false); - } + ensureTopicExists(client, topicName).then(() => { + const messageId = `health-check-${uuidv4()}`; + const payloads = [{ topic: topicName, messages: messageId }]; + + const producer = new kafka.Producer(client); + let settled = false; + const settle = (ok) => { + if (settled) return; + settled = true; + clearTimeout(timeout); + consumer.close(true, () => { + // Close producer if available, then client + try { + producer.close(() => client.close()); + } catch (_) { + client.close(); + } + resolve(ok); + }); + }; - const messageId = `health-check-${uuidv4()}`; - const payloads = [ - { - topic: topicName, - messages: messageId, - }, - ]; + const consumer = new kafka.Consumer( + client, + [{ topic: topicName, partition: 0 }], + { autoCommit: true, fromOffset: false } + ); - const producer = new kafka.Producer(client); + const timeout = setTimeout(() => { + console.error('[Kafka Health Check] Timed out waiting for message'); + settle(false); + }, 10000); - producer.on('ready', () => { - console.log(`[Kafka Health Check] Producer ready. Sending message: ${messageId}`); - - producer.send(payloads, (err, data) => { - if (err) { - console.error('[Kafka Health Check] Error sending message:', err); - client.close(); - return resolve(false); - } - - console.log('[Kafka Health Check] Message sent:', data); - - const consumer = new kafka.Consumer( - client, - [{ topic: topicName, partition: 0 }], - { - autoCommit: true, - fromOffset: false, - } - ); - - const timeout = setTimeout(() => { - console.error('[Kafka Health Check] Timed out waiting for message'); - consumer.close(true, () => { - client.close(); - resolve(false); - }); - }, 10000); - - consumer.on('message', (message) => { - console.log('[Kafka Health Check] Received message:', message.value); - if (message.value === messageId) { - clearTimeout(timeout); - consumer.close(true, () => { - client.close(); - resolve(true); - }); - } - }); - - consumer.on('error', (err) => { - console.error('[Kafka Health Check] Consumer error:', err); - clearTimeout(timeout); - consumer.close(true, () => { - client.close(); - resolve(false); - }); - }); - }); - }); + consumer.on('message', (message) => { + console.log('[Kafka Health Check] Received message:', message.value); + if (message.value === messageId) { + settle(true); + } + }); + + consumer.on('error', (err) => { + console.error('[Kafka Health Check] Consumer error:', err); + settle(false); + }); + + producer.on('ready', () => { + console.log(`[Kafka Health Check] Producer ready. Sending message: ${messageId}`); + producer.send(payloads, (err, data) => { + if (err) { + console.error('[Kafka Health Check] Error sending message:', err); + return settle(false); + } + console.log('[Kafka Health Check] Message sent:', data); + }); + }); + + producer.on('error', (err) => { + console.error('[Kafka Health Check] Producer error:', err); + settle(false); + }); + }).catch((err) => { + client.close(); + return resolve(false); + }); - producer.on('error', (err) => { - console.error('[Kafka Health Check] Producer error:', err); - client.close(); - return resolve(false); - }); }); }
🧹 Nitpick comments (2)
health-check/package.json (1)
11-19: Consider removing duplicate Kafka clients to reduce footprint.Both kafka-node and kafkajs are listed as dependencies, but this package (health-check) exclusively uses kafka-node. If kafkajs isn’t used anywhere in this package, consider removing it to shrink install size and reduce maintenance.
health-check/services/kafka.js (1)
11-47: Optional: reduce metadata scope when checking topics.Using loadMetadataForTopics([]) fetches metadata for all topics. If broker ACLs and topic counts are large, consider requesting metadata only for the target topic (e.g., loadMetadataForTopics([topicName])) and handling the “unknown topic” case to decide on creation. This reduces payload and latency.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (3)
health-check/index.js(1 hunks)health-check/package.json(1 hunks)health-check/services/kafka.js(3 hunks)
🔇 Additional comments (2)
health-check/package.json (1)
3-3: Version bump looks good.Package version updated to 0.0.4 — no issues.
health-check/index.js (1)
38-38: All kafka.check invocations updated
Verified that the sole call site in health-check/index.js (line 38) passes bothurlandtopic; no other references exist.
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
2 participants
Summary by CodeRabbit