Fix Kafka commit ordering after producer writes

[officialasishkumar]

Description

• Delay Kafka consumer offset completion until downstream producer writes succeed.
• Propagate producer write failures through the processing future so single and batch commit paths do not acknowledge records that were not written to the next topic.
• Add focused KafkaMosipEventBus tests for producer-success commit ordering and producer-failure behavior.

Fixes #2305

Tests

• JAVA_HOME=/usr/lib/jvm/java-21-openjdk-amd64 PATH=/usr/lib/jvm/java-21-openjdk-amd64/bin:$PATH mvn -pl registration-processor-core -Dtest=KafkaMosipEventBusTest -DfailIfNoTests=false -Dgpg.skip=true -Dmaven.javadoc.skip=true test

Summary by CodeRabbit

• Refactor

  • Improved event processing and message-handling flow, including safer context propagation and more precise tracing/span lifecycle to ensure accurate commit/rollback behavior.

• Tests

  • Added unit tests covering producer success and failure paths and commit/resume behavior to validate processing across single and batch modes.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Repository UI

Review profile: ASSERTIVE

Plan: Pro

Run ID: 9f034dc0-2c98-482d-9e09-c569d36507c8

📥 Commits

Reviewing files that changed from the base of the PR and between 5476cba and 9268c83.

📒 Files selected for processing (2)

• registration-processor/registration-processor-core/src/main/java/io/mosip/registration/processor/core/eventbus/KafkaMosipEventBus.java
• registration-processor/registration-processor-core/src/test/java/io/mosip/registration/processor/core/eventbus/KafkaMosipEventBusTest.java

---

📝 Walkthrough

Walkthrough

Refactors KafkaMosipEventBus to defer consumer offset commit until downstream producer write confirms; centralizes MDC and promise resolution into helpers and moves span closure into branch-specific locations. Adds unit tests covering producer success and failure in single and batch commit modes.

Changes

Cohort / File(s)	Summary
Core Event Bus Refactor registration-processor/registration-processor-core/src/main/java/io/mosip/registration/processor/core/eventbus/KafkaMosipEventBus.java	Added helpers setMDCContextMap(...), completeProcessedRecord(...), failPromise(...). Reworked processRecord() to commit offsets only after successful kafkaProducer.write() callback, explicitly fail promises on producer error, and move span closure into branch-specific points (including inside producer callback).
Event Bus Test Coverage registration-processor/registration-processor-core/src/test/java/io/mosip/registration/processor/core/eventbus/KafkaMosipEventBusTest.java	Added Vert.x/Kafka Mockito stubs and helpers; introduced tests for: successful producer write → commit in single mode; producer write failure → promise fails, no commit, consumer resumes (both single and batch modes).

Sequence Diagram(s)

sequenceDiagram
    participant Consumer
    participant KafkaMosipEventBus
    participant KafkaProducer
    participant KafkaBroker
    participant Promise

    Consumer->>KafkaMosipEventBus: processRecord(record)
    activate KafkaMosipEventBus
    KafkaMosipEventBus->>KafkaMosipEventBus: handle record, extract MDC, start span

    alt processing success
        KafkaMosipEventBus->>KafkaProducer: write(producerRecord, callback)
        activate KafkaProducer
        KafkaProducer->>KafkaBroker: send message (async)
        activate KafkaBroker
        KafkaBroker-->>KafkaProducer: ack / failure
        deactivate KafkaBroker
        KafkaProducer-->>KafkaMosipEventBus: callback(success/failure)
        deactivate KafkaProducer

        alt producer success
            KafkaMosipEventBus->>Promise: completeProcessedRecord() / commitOffset()
            KafkaMosipEventBus->>KafkaMosipEventBus: closeSpan()
        else producer failure
            KafkaMosipEventBus->>Promise: failPromise(cause)
            KafkaMosipEventBus->>KafkaMosipEventBus: closeSpan()
        end
    else processing failure
        KafkaMosipEventBus->>Promise: failPromise(cause)
        KafkaMosipEventBus->>KafkaMosipEventBus: closeSpan()
    end

    Promise-->>Consumer: resolved / failed
    deactivate KafkaMosipEventBus

Loading

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~22 minutes

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 5.26% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The PR title 'Fix Kafka commit ordering after producer writes' directly describes the main change: deferring consumer offset commits until after async producer writes succeed.
Linked Issues check	✅ Passed	The PR fully addresses the requirements in issue #2305: moves commit logic inside producer callback, propagates write failures to promise, handles both single and batch modes, and adds comprehensive tests.
Out of Scope Changes check	✅ Passed	All code changes directly relate to the linked issue #2305: fixing commit ordering (main code), adding MDC/span closure helpers, and adding control-flow tests. No out-of-scope changes detected.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with 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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

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

[coderabbitai]

Actionable comments posted: 2

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In
`@registration-processor/registration-processor-core/src/main/java/io/mosip/registration/processor/core/eventbus/KafkaMosipEventBus.java`:
- Around line 374-382: The partition is only resumed on successful
producer.write; when handler.failed() you must also resume the paused partition
so the consumer can continue/retry; update the failure branch in
KafkaMosipEventBus (the handler.failed() branch that calls failPromise(promise,
handler.cause(), "Failed kafkaProducer.write")) to invoke resumePartition(...)
for the current record/partition before/after failing the promise (mirroring the
success path where completeProcessedRecord(...) and resumePartition are called),
and ensure any tracing cleanup (eventTracingHandler.closeSpan(span) and
MDC.clear()) still runs.

In
`@registration-processor/registration-processor-core/src/test/java/io/mosip/registration/processor/core/eventbus/KafkaMosipEventBusTest.java`:
- Around line 363-405: The tests
testProcessRecordCommitsAfterProducerWriteSuccess,
testProcessRecordFailsAndDoesNotCommitWhenProducerWriteFails, and
testProcessRecordFailsBatchFutureWhenProducerWriteFails call
kafkaMosipEventBus.processRecord(...) and assert result.succeeded()/failed()
synchronously which relies on synchronous mocks
(mockKafkaProducerWriteResult/mockCommitResult); change each test to assert
completion inside the Future's asynchronous callback (e.g., result.onComplete or
using TestContext.async) so assertions (result.isSucceeded()/isFailed(),
result.cause(), and the verify/inOrder checks for kafkaProducer.write and
kafkaConsumer.commit) run after the Future actually completes, ensuring the
tests do not race if producer/commit handlers become async.

🪄 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: Repository UI

Review profile: ASSERTIVE

Plan: Pro

Run ID: 0a18e653-d660-4209-a056-9e8cb9a70d0f

📥 Commits

Reviewing files that changed from the base of the PR and between f86a411 and 5476cba.

📒 Files selected for processing (2)

• registration-processor/registration-processor-core/src/main/java/io/mosip/registration/processor/core/eventbus/KafkaMosipEventBus.java
• registration-processor/registration-processor-core/src/test/java/io/mosip/registration/processor/core/eventbus/KafkaMosipEventBusTest.java