kafka-fix

[MallanagoudaB]

Summary by CodeRabbit

• Bug Fixes

  • Improved error handling, standardized error messages, enhanced logging, and reliable cleanup to prevent resource leaks during health checks.

• New Features

  • Optional end-to-end validation: health-check can now perform an optional send/receive verification in addition to topic validation; default behavior remains lightweight.

• Chores

  • Package version bumped to 0.0.6.

[coderabbitai]

Walkthrough

Adds an async, promise-based Kafka health-check that ensures a topic exists (creating it if missing), optionally performs a producer→consumer round-trip, centralizes cleanup to close consumer/client exactly once, expands logging/error handling, and updates check(kafkaUrl, topicName, groupId, sendReceive = false). (≤50 words)

Changes

Cohort / File(s)	Summary
Kafka health-check logic health-check/services/kafka.js	Replaces callback-based flow with a linear promise/async sequence. Adds ensureTopicExists (metadata lookup + create), centralized cleanup to close consumer/client exactly once, expanded error handling and logging, conditional sendReceive path (skip or perform producer→consumer round-trip), and updates signature to async function check(kafkaUrl, topicName, groupId, sendReceive = false).
Package version health-check/package.json	Bumps package version from 0.0.5 → 0.0.6. No functional or dependency changes.

Sequence Diagram(s)

sequenceDiagram
    actor Caller
    participant Module as kafka.js
    participant Client as Kafka Client
    participant Metadata as Metadata API
    participant Producer
    participant Consumer

    Caller->>Module: check(kafkaUrl, topicName, groupId, sendReceive)
    activate Module

    Module->>Client: create client
    Client-->>Module: ready

    Module->>Metadata: loadMetadataForTopics / ensureTopicExists
    Metadata-->>Module: topic exists / created

    alt sendReceive = false
        Module-->>Caller: resolve (topic validated)
    else sendReceive = true
        Module->>Producer: init producer & wait ready
        Producer-->>Module: ready
        Module->>Producer: send health message
        Producer-->>Module: ack

        Module->>Consumer: init consumer & subscribe
        Consumer-->>Module: ready
        Module->>Consumer: poll for message (with timeout)
        Consumer-->>Module: message received

        rect rgb(230,245,255)
            note right of Module: centralized cleanup — close consumer, close client (exactly once)
        end

        Module-->>Caller: resolve (round-trip verified)
    end

    deactivate Module

Loading

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~20 minutes

• Focus areas:
  • ensureTopicExists metadata lookup and topic-creation correctness.
  • Centralized cleanup semantics (ensure single close and single resolution).
  • sendReceive branch: producer send, consumer poll/timeout, event/error handlers and proper resource teardown.

Possibly related PRs

• added-group-id #342 — Updates health-check/services/kafka.js with groupId usage and consumer/topic handling; likely overlaps with groupId/sendReceive and cleanup changes.
• kafka-topic-change #341 — Modifies the same Kafka health-check to accept dynamic topic handling; related to topic validation/refactor.

Suggested reviewers

• VISHNUDAS-tunerlabs
• priyanka-TL

Poem

🐰 In tunnels of code where the topics once hid,

I hop in with metadata, tidy and tid.
A producer drops carrots, a consumer will cheer,
Cleanup tucks tails as the logs make it clear.
Hooray — healthy topics and messages near! 🥕✨

Pre-merge checks and finishing touches

❌ Failed checks (1 inconclusive)

Check name	Status	Explanation	Resolution
Title Check	❓ Inconclusive	The PR title "kafka-fix" is a vague, single-word descriptor that uses a non-specific term ("fix") without clarifying what aspect of Kafka was addressed. While the title is technically related to the changeset since the modifications involve the Kafka health-check service, it fails to convey meaningful information about the actual changes. A developer scanning commit history would not understand from this title that the PR involves refactoring callback-based logic to promises, adding topic creation capabilities, implementing conditional send/receive paths, or improving error handling. The title is overly broad and lacks the specificity needed to communicate the primary change.	Consider revising the title to be more descriptive and specific. Examples might include "Refactor Kafka health check to use promise-based flow" or "Add topic creation and conditional messaging to Kafka health check." The improved title should clearly communicate the main change so that teammates can understand the PR's purpose without needing to read the full diff.

✅ Passed checks (2 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.

✨ Finishing touches

• 📝 Generate docstrings

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment
• Commit unit tests in branch KafkaFix

---

📜 Recent review details

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between 71dc90c and f00bc08.

📒 Files selected for processing (2)

• health-check/package.json (1 hunks)
• health-check/services/kafka.js (1 hunks)

🧰 Additional context used

🧠 Learnings (1)

📚 Learning: 2025-08-20T10:33:46.359Z

Learnt from: MallanagoudaB
PR: ELEVATE-Project/utils#342
File: health-check/services/kafka.js:49-54
Timestamp: 2025-08-20T10:33:46.359Z
Learning: In the ELEVATE-Project/utils health-check module, validation for Kafka configuration including groupId is performed in the parent file (health-check/index.js) in the validateHealthConfig function, not in the individual service files like health-check/services/kafka.js.

Applied to files:

• health-check/services/kafka.js

🪛 Biome (2.1.2)

health-check/services/kafka.js

[error] 46-133: Promise executor functions should not be async.

(lint/suspicious/noAsyncPromiseExecutor)

🔇 Additional comments (4)

health-check/package.json (1)

3-3: Version bump looks appropriate.

The minor version increment from 0.0.5 to 0.0.6 aligns with the functional improvements made to the Kafka health check module.

health-check/services/kafka.js (3)

14-36: Topic existence check looks good.

The ensureTopicExists function properly verifies topic existence and creates it with appropriate settings for a health check (1 partition, 1 replication factor).

---

75-78: Good design: Optional send/receive path.

The conditional path that completes after topic validation when sendReceive is false is a sensible optimization, allowing lightweight health checks without the overhead of producer/consumer setup.

---

47-49: Verify: Process-specific topics may cause proliferation.

Appending the process ID to topic and group names creates unique topics per process. Combined with automatic topic creation, this will result in multiple health-check topics accumulating in Kafka (one per process that runs the health check).

Confirm whether this is the intended behavior, or if health-check topics should be shared across processes. If topics should be shared, remove the pidSuffix from uniqueTopicName (keeping it only for uniqueGroupId to avoid consumer conflicts).

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[MallanagoudaB]

@coderabbitai full review

[coderabbitai]

✅ Actions performed

Full review triggered.

[coderabbitai]

Actionable comments posted: 4

🧹 Nitpick comments (1)

health-check/services/kafka.js (1)

47-49: Consider alternative to PID for uniqueness in containerized environments.

Using process.pid as a suffix may not guarantee uniqueness in containerized environments (Docker, Kubernetes) where PIDs are namespaced and commonly reused. If multiple instances or rapid restarts occur, topic/group collisions could happen.

Consider using a combination of hostname and timestamp, or a UUID:

-const pidSuffix = `-${process.pid}`
+const pidSuffix = `-${require('os').hostname()}-${Date.now()}`

Or if guaranteed uniqueness is critical:

-const pidSuffix = `-${process.pid}`
+const pidSuffix = `-${uuidv4()}`

📜 Review details

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between 71dc90c and 59d5503.

📒 Files selected for processing (1)

• health-check/services/kafka.js (1 hunks)

🧰 Additional context used

🧠 Learnings (2)

📓 Common learnings

Learnt from: MallanagoudaB
PR: ELEVATE-Project/utils#342
File: health-check/services/kafka.js:49-54
Timestamp: 2025-08-20T10:33:46.359Z
Learning: In the ELEVATE-Project/utils health-check module, validation for Kafka configuration including groupId is performed in the parent file (health-check/index.js) in the validateHealthConfig function, not in the individual service files like health-check/services/kafka.js.

📚 Learning: 2025-08-20T10:33:46.359Z

Learnt from: MallanagoudaB
PR: ELEVATE-Project/utils#342
File: health-check/services/kafka.js:49-54
Timestamp: 2025-08-20T10:33:46.359Z
Learning: In the ELEVATE-Project/utils health-check module, validation for Kafka configuration including groupId is performed in the parent file (health-check/index.js) in the validateHealthConfig function, not in the individual service files like health-check/services/kafka.js.

Applied to files:

• health-check/services/kafka.js

🧬 Code graph analysis (1)

health-check/services/kafka.js (1)

health-check/index.js (1)

• kafka (37-37)

🪛 Biome (2.1.2)

health-check/services/kafka.js

[error] 46-133: Promise executor functions should not be async.

(lint/suspicious/noAsyncPromiseExecutor)

🔇 Additional comments (1)

health-check/services/kafka.js (1)

14-36: Original review comment is incorrect—code follows the documented kafka-node API.

The concerns raised are based on misunderstanding the kafka-node library:

1. Line 22 metadata access: results?.[1]?.metadata is the correct and documented way to access metadata from kafka-node's loadMetadataForTopics response. It's not fragile; it's the library's standard response structure.

2. replicationFactor: 1: This is intentional for a health-check topic. Health checks don't require the fault tolerance of production topics, and a single replica simplifies the health check logic.

The code is correct as written and requires no changes.

Likely an incorrect or invalid review comment.

[VISHNUDAS-tunerlabs]

Reviewed - on oct 30