Boundary and Persister Service changes - #1377
Conversation
…ditional batch configuration
…letter queue re-processor
…ter + boundary-service) 05f6889 also carried ON CONFLICT edits to persister YAMLs in unrelated modules (audit-service, egov-pg-service, egov-workflow-v2, mdms-v2, service-request). These are out of scope for the persister RCA reliability fix and the boundary-relationship enhancement, so they are reverted to their prior state. This branch now touches only egov-persister and boundary-service.
BoundaryRelationshipMappingShapeTest was a local verification harness used to prove the dedicated-bulk-topic fix (single-object vs array extraction). Its purpose is served, and a boundary-relationship-specific test does not belong in the generic egov-persister module. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughBoundary Service adds bulk boundary-relationship creation with synchronous validation and Kafka publication. egov-persister adds batch processing, failure classification, retry/dead-letter routing, database health monitoring, aggregated persistence, and conflict-safe inserts. Tracer adds Kafka MDC propagation. ChangesBoundary bulk creation
Persister reliability and batching
Kafka tracing
Estimated code review effort: 5 (Critical) | ~90 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant BoundaryService
participant Kafka
participant Persister
participant Database
Client->>BoundaryService: Submit bulk boundary relationships
BoundaryService->>BoundaryService: Validate and enrich records
BoundaryService->>Kafka: Publish batch relationship message
Kafka->>Persister: Deliver batch message
Persister->>Database: Aggregate and persist rows
Database-->>Persister: Success or classified failure
Persister-->>Kafka: Retry, dead-letter, or park failed records
Suggested reviewers: 🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (3 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 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: 10
🧹 Nitpick comments (3)
core-services/boundary-service/src/main/java/digit/web/models/BulkBoundaryRelationshipRequest.java (1)
40-40: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove the hardcoded
@Sizeconstraint to avoid conflicts with programmatic configuration.The
@Size(min = 1, max = 100)constraint hardcodes the maximum batch size to 100. If Bean Validation is ever enabled in the deployment, this will conflict with the configurableboundary.bulk.max.sizeproperty (which defaults to 100 but can be changed). Since the service explicitly enforces the configurable size limit programmatically, consider removing the@Sizeconstraint to prevent potential conflicts and ensure the configuration property remains the single source of truth.♻️ Proposed refactor
Remove the annotation:
- `@Size`(min = 1, max = 100)You can optionally remove the unused import at the top of the file as well:
import jakarta.validation.constraints.Size;🤖 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 `@core-services/boundary-service/src/main/java/digit/web/models/BulkBoundaryRelationshipRequest.java` at line 40, Remove the hardcoded `@Size` constraint from BulkBoundaryRelationshipRequest and delete its now-unused Size import. Keep batch-size validation in the existing programmatic enforcement that uses the configurable boundary.bulk.max.size property.core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/PersisterMessageListenerTest.java (1)
167-189: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winExercise permanent-before-transient ordering.
The current
slow, badorder throws before any DLQ side effect. Addbad, slowand verify that a later transient failure does not leave an already-published DLQ entry before the original message is retried.🤖 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 `@core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/PersisterMessageListenerTest.java` around lines 167 - 189, Extend transientFailureDuringIsolationRethrowsAndNothingIsDeadLettered to exercise the bad, slow ordering, ensuring the first permanent record is processed before the later transient failure. Assert that the listener rethrows TransientPersistException and kafkaTemplate.send is never invoked, so no DLQ entry is published before the original message is retried.core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/PersisterBatchListnerRecordIsolationTest.java (1)
123-145: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winAdd the reverse isolation order.
Test
[bad, slow]as well as[slow, bad]; the former exposes whether a permanent record is dead-lettered before the later transient failure forces the whole poll to replay.🤖 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 `@core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/PersisterBatchListnerRecordIsolationTest.java` around lines 123 - 145, Extend transientFailureDuringIsolationRethrowsForInPlaceRetry to also invoke listener.onMessage with the records ordered as [bad, slow]. Assert that it throws TransientPersistException and verify kafkaTemplate never sends to DLQ for this reverse isolation order, preserving the existing [slow, bad] assertions.
🤖 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
`@core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/DbExceptionClassifier.java`:
- Around line 30-49: Remove SQLState 57014 from the transient classification in
DbExceptionClassifier, while preserving the other transient SQLState checks. In
DbExceptionClassifierTest, remove the assertion that expects 57014 to be
transient; no direct change is needed for the remaining test cases.
In
`@core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterBatchConsumerConfig.java`:
- Around line 112-115: The independently started batchContainer is not
controlled by DbHealthMonitor during database outages. Add idempotent pause and
resume operations for batchContainer in PersisterBatchConsumerConfig, then
update the monitor integration to invoke those operations alongside
PersisterConsumerConfig so both listener containers stop and restart polling
together.
In
`@core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterBatchListner.java`:
- Around line 195-216: Update the duplicate-handling branches in the batch and
single-record paths around DbExceptionClassifier.Kind.BENIGN to track
already-existing records separately from newly persisted records instead of
adding them to persisted. Keep persisted limited to records that were actually
committed so the sendAudit call only emits events for new writes.
- Around line 172-211: Immediate DLQ publication in the record-isolation flow
can duplicate permanent failures when a later transient error retries the batch.
In
core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterBatchListner.java:172-211,
stage isolated DLQ records until the entire sweep completes, or publish them
idempotently using source offsets. Add permanent-before-transient ordering
coverage in
core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/PersisterMessageListenerTest.java:167-189
and
core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/PersisterBatchListnerRecordIsolationTest.java:123-145.
In
`@core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterConsumerConfig.java`:
- Around line 163-166: Update the customExecutorEnabled branch in
PersisterConsumerConfig so the fixed-thread-pool ExecutorService is
lifecycle-managed: either expose it as a managed bean or register shutdown when
the KafkaMessageListenerContainer stops. Do not create an unmanaged local
executor, and preserve the existing listener task-executor configuration.
In
`@core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterMessageListener.java`:
- Around line 149-169: The record-isolation loop in PersisterMessageListener
must defer DLQ and parking publication until the entire sweep completes without
a transient failure. Stage each permanent failure and its retry metadata during
the loop, preserve benign/transient handling, then publish the staged entries
only after the loop succeeds, updating deadLettered and parked counts at
publication time.
- Around line 70-73: Update the attempts parsing in PersisterMessageListener so
only integral numeric values within the valid retry-counter range are accepted;
treat negative, overflowing, non-integral, and non-numeric values as maxRetries.
Avoid Number.intValue() silently narrowing invalid values, while preserving the
existing ceiling behavior for foreign or raw DLQ records.
In
`@core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterProducerConfig.java`:
- Around line 62-71: Update the parking-envelope construction in the producer
error-handling flow to populate parked.body with the original raw bytes from the
deserialization exception header when record.value() is null, while preserving
the existing value for successfully deserialized records. Use the available
exception/header metadata associated with the failed record so poison messages
remain replayable.
In
`@core-services/egov-persister/src/main/java/org/egov/infra/persist/service/PersistService.java`:
- Around line 118-129: Remove the try-catch around persistRepository.getRows in
the document-processing loop so extraction exceptions propagate to the upstream
listener for record-level isolation and DLQ routing. Keep the existing
skippedDocuments increment for empty results, and do not increment it or
log-and-continue when getRows fails.
- Around line 74-88: Replace the documentToMappings LinkedHashMap in the persist
flow with a list of document-to-mappings entries so identical parsed JSON
documents remain separate batch items. Update the subsequent iteration to
process each entry in insertion order while preserving every message’s
applicable mappings and exact multiplicity.
---
Nitpick comments:
In
`@core-services/boundary-service/src/main/java/digit/web/models/BulkBoundaryRelationshipRequest.java`:
- Line 40: Remove the hardcoded `@Size` constraint from
BulkBoundaryRelationshipRequest and delete its now-unused Size import. Keep
batch-size validation in the existing programmatic enforcement that uses the
configurable boundary.bulk.max.size property.
In
`@core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/PersisterBatchListnerRecordIsolationTest.java`:
- Around line 123-145: Extend
transientFailureDuringIsolationRethrowsForInPlaceRetry to also invoke
listener.onMessage with the records ordered as [bad, slow]. Assert that it
throws TransientPersistException and verify kafkaTemplate never sends to DLQ for
this reverse isolation order, preserving the existing [slow, bad] assertions.
In
`@core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/PersisterMessageListenerTest.java`:
- Around line 167-189: Extend
transientFailureDuringIsolationRethrowsAndNothingIsDeadLettered to exercise the
bad, slow ordering, ensuring the first permanent record is processed before the
later transient failure. Assert that the listener rethrows
TransientPersistException and kafkaTemplate.send is never invoked, so no DLQ
entry is published before the original message is retried.
🪄 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: defaults
Review profile: CHILL
Plan: Pro
Run ID: aa854349-5b40-43e0-b5a7-677c9df1b889
📒 Files selected for processing (48)
core-services/boundary-service/README.mdcore-services/boundary-service/src/main/java/digit/config/ApplicationProperties.javacore-services/boundary-service/src/main/java/digit/errors/ErrorCodes.javacore-services/boundary-service/src/main/java/digit/kafka/Producer.javacore-services/boundary-service/src/main/java/digit/repository/BoundaryRelationshipRepository.javacore-services/boundary-service/src/main/java/digit/repository/impl/BoundaryRelationshipRepositoryImpl.javacore-services/boundary-service/src/main/java/digit/repository/querybuilder/BoundaryEntityQueryBuilder.javacore-services/boundary-service/src/main/java/digit/repository/querybuilder/BoundaryRelationshipQueryBuilder.javacore-services/boundary-service/src/main/java/digit/service/BoundaryRelationshipService.javacore-services/boundary-service/src/main/java/digit/service/validator/BoundaryRelationshipValidator.javacore-services/boundary-service/src/main/java/digit/util/HierarchyUtil.javacore-services/boundary-service/src/main/java/digit/web/controllers/BoundaryRelationshipController.javacore-services/boundary-service/src/main/java/digit/web/models/BulkBoundaryRelationshipRequest.javacore-services/boundary-service/src/main/java/digit/web/models/BulkBoundaryRelationshipRequestDTO.javacore-services/boundary-service/src/main/java/digit/web/models/BulkBoundaryRelationshipResponse.javacore-services/boundary-service/src/main/java/digit/web/models/FailedBoundaryRelationship.javacore-services/boundary-service/src/main/resources/application.propertiescore-services/boundary-service/src/main/resources/boundary-persister.ymlcore-services/boundary-service/src/main/resources/db/migration/main/V20260616120000__boundary_relationship_search_indexes.sqlcore-services/boundary-service/src/test/java/digit/repository/impl/BoundaryRelationshipRepositoryImplBulkTest.javacore-services/egov-persister/src/main/java/org/egov/EgovPersistApplication.javacore-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/DbExceptionClassifier.javacore-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/DbHealthMonitor.javacore-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterBatchConsumerConfig.javacore-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterBatchListner.javacore-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterConsumerConfig.javacore-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterMessageListener.javacore-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterProducerConfig.javacore-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/RecordSplitter.javacore-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/StoppingErrorHandler.javacore-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/TransientPersistException.javacore-services/egov-persister/src/main/java/org/egov/infra/persist/repository/PersistRepository.javacore-services/egov-persister/src/main/java/org/egov/infra/persist/service/PersistService.javacore-services/egov-persister/src/main/resources/application.propertiescore-services/egov-persister/src/main/resources/egov-pg-service-persister.ymlcore-services/egov-persister/src/main/resources/egov-user-event-persister.ymlcore-services/egov-persister/src/main/resources/hrms-employee-persister.ymlcore-services/egov-persister/src/main/resources/persister.ymlcore-services/egov-persister/src/main/resources/pgr.v3.ymlcore-services/egov-persister/src/main/resources/property-services.ymlcore-services/egov-persister/src/main/resources/pt-drafts.ymlcore-services/egov-persister/src/main/resources/tl-billing-slab-persister.ymlcore-services/egov-persister/src/main/resources/user-service-persist.ymlcore-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/DbExceptionClassifierTest.javacore-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/PersisterBatchConsumerConfigTest.javacore-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/PersisterBatchListnerRecordIsolationTest.javacore-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/PersisterMessageListenerTest.javacore-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/RecordSplitterTest.java
💤 Files with no reviewable changes (2)
- core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/StoppingErrorHandler.java
- core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/PersisterBatchConsumerConfigTest.java
| batchContainer = new KafkaMessageListenerContainer<>(createConsumerFactory(), properties); | ||
| batchContainer.setCommonErrorHandler(persisterErrorHandler); | ||
| batchContainer.setBeanName("batchContainer"); | ||
| batchContainer.start(); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Include the batch container in database-health pause/resume.
DbHealthMonitor only controls PersisterConsumerConfig, so this independently started batch container continues polling during a database outage. Expose idempotent pause/resume operations here and have the monitor control both containers, or introduce a shared container registry.
🤖 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
`@core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterBatchConsumerConfig.java`
around lines 112 - 115, The independently started batchContainer is not
controlled by DbHealthMonitor during database outages. Add idempotent pause and
resume operations for batchContainer in PersisterBatchConsumerConfig, then
update the monitor integration to invoke those operations alongside
PersisterConsumerConfig so both listener containers stop and restart polling
together.
| for (String message : messages) { | ||
| try { | ||
| persistService.persist(topic, Collections.singletonList(message)); | ||
| persisted.add(message); | ||
| } catch (Exception ex) { | ||
| DbExceptionClassifier.Kind kind = DbExceptionClassifier.classify(ex); | ||
| if (kind == DbExceptionClassifier.Kind.TRANSIENT) { | ||
| // DB went down partway through isolation: abort and retry the whole batch rather | ||
| // than parking the remaining good records as if they were poison. | ||
| throw new TransientPersistException("Transient DB failure isolating record for topic " + topic, ex); | ||
| } | ||
| else { | ||
| topicTorcvDataList.get(data.topic()).add(objectMapper.writeValueAsString(data.value())); | ||
| // A bulk producer publishes a whole list as ONE message, so isolate WITHIN the | ||
| // message (R1 at record granularity): good sibling rows must not share a poison | ||
| // row's dead-letter, and a BENIGN duplicate must not absorb not-yet-persisted | ||
| // siblings (the array insert aborts on the duplicate before reaching them). | ||
| List<String> records = RecordSplitter.split(message); | ||
| if (records != null) { | ||
| for (String record : records) { | ||
| try { | ||
| persistService.persist(topic, Collections.singletonList(record)); | ||
| persisted.add(record); | ||
| } catch (Exception rex) { | ||
| DbExceptionClassifier.Kind recordKind = DbExceptionClassifier.classify(rex); | ||
| if (recordKind == DbExceptionClassifier.Kind.BENIGN) { | ||
| persisted.add(record); // already present -> idempotent success | ||
| } else if (recordKind == DbExceptionClassifier.Kind.TRANSIENT) { | ||
| throw new TransientPersistException("Transient DB failure isolating record for topic " + topic, rex); | ||
| } else { | ||
| sendToDlq(topic, record, 0, rex); // only the offending record | ||
| failed++; | ||
| } | ||
| } | ||
| } | ||
| } else if (kind == DbExceptionClassifier.Kind.BENIGN) { | ||
| persisted.add(message); // single already-present record -> idempotent success | ||
| } else { | ||
| sendToDlq(topic, message, 0, ex); // unsplittable payload: message-level isolation | ||
| failed++; | ||
| } | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Immediate DLQ side effects are unsafe before a potentially transient sweep completes. A later transient failure replays the original poll, duplicating any permanent records already published to the DLQ.
core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterBatchListner.java#L172-L211: stage DLQ entries until isolation completes, or publish them idempotently using source offsets.core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/PersisterMessageListenerTest.java#L167-L189: add permanent-before-transient ordering.core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/PersisterBatchListnerRecordIsolationTest.java#L123-L145: add the same reverse-order regression case.
📍 Affects 3 files
core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterBatchListner.java#L172-L211(this comment)core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/PersisterMessageListenerTest.java#L167-L189core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/PersisterBatchListnerRecordIsolationTest.java#L123-L145
🤖 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
`@core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterBatchListner.java`
around lines 172 - 211, Immediate DLQ publication in the record-isolation flow
can duplicate permanent failures when a later transient error retries the batch.
In
core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterBatchListner.java:172-211,
stage isolated DLQ records until the entire sweep completes, or publish them
idempotently using source offsets. Add permanent-before-transient ordering
coverage in
core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/PersisterMessageListenerTest.java:167-189
and
core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/PersisterBatchListnerRecordIsolationTest.java:123-145.
| if (recordKind == DbExceptionClassifier.Kind.BENIGN) { | ||
| persisted.add(record); // already present -> idempotent success | ||
| } else if (recordKind == DbExceptionClassifier.Kind.TRANSIENT) { | ||
| throw new TransientPersistException("Transient DB failure isolating record for topic " + topic, rex); | ||
| } else { | ||
| sendToDlq(topic, record, 0, rex); // only the offending record | ||
| failed++; | ||
| } | ||
| } | ||
| } | ||
| } else if (kind == DbExceptionClassifier.Kind.BENIGN) { | ||
| persisted.add(message); // single already-present record -> idempotent success | ||
| } else { | ||
| sendToDlq(topic, message, 0, ex); // unsplittable payload: message-level isolation | ||
| failed++; | ||
| } | ||
| } | ||
| catch (JsonProcessingException e) { | ||
| log.error("Failed to serialize incoming message", e); | ||
| } | ||
| }); | ||
|
|
||
| for(Map.Entry<String,List<String>> entry : topicTorcvDataList.entrySet()){ | ||
| persistService.persist(entry.getKey(),entry.getValue()); | ||
| if(!entry.getKey().equalsIgnoreCase(persistAuditKafkaTopic)){ | ||
| Map<String, Object> producerRecord = new HashMap<>(); | ||
| producerRecord.put("topic", entry.getKey()); | ||
| producerRecord.put("value", entry.getValue()); | ||
| kafkaTemplate.send(auditGenerateKafkaTopic, producerRecord); | ||
| } | ||
| } | ||
|
|
||
| if (!persisted.isEmpty()) { | ||
| sendAudit(topic, persisted); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
Exclude benign duplicates from the post-commit audit.
Lines 195 and 205 add already-existing rows to persisted, so Line 216 emits audit events even though no write occurred. Track handled duplicates separately and audit only newly committed rows.
🤖 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
`@core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterBatchListner.java`
around lines 195 - 216, Update the duplicate-handling branches in the batch and
single-record paths around DbExceptionClassifier.Kind.BENIGN to track
already-existing records separately from newly persisted records instead of
adding them to persisted. Keep persisted limited to records that were actually
committed so the sendAudit call only emits events for new writes.
| if (customExecutorEnabled) { | ||
| ExecutorService executorService = Executors.newFixedThreadPool(maxPoolSize); | ||
| AsyncTaskExecutor taskExecutor = new ConcurrentTaskExecutor(executorService); | ||
| properties.setListenerTaskExecutor(taskExecutor); |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
file='core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterConsumerConfig.java'
echo "== outline =="
ast-grep outline "$file" --view expanded || true
echo
echo "== relevant lines =="
sed -n '130,220p' "$file" | cat -n
echo
echo "== search for container type / concurrency / executor lifecycle =="
rg -n "KafkaMessageListenerContainer|ConcurrentMessageListenerContainer|setListenerTaskExecutor|setConcurrency|customExecutorEnabled|maxPoolSize|destroy|shutdown|ExecutorService" core-services/egov-persister/src/main/javaRepository: egovernments/Digit-Core
Length of output: 8570
🌐 Web query:
Spring Kafka ContainerProperties setListenerTaskExecutor polling thread lifecycle shutdown executor ownership
💡 Result:
In Spring Kafka, ContainerProperties.setListenerTaskExecutor is used to configure the AsyncTaskExecutor responsible for running the container's consumer threads [1][2][3]. Ownership and Lifecycle Management: 1. Default Executor: If no listenerTaskExecutor is provided, the container automatically creates a SimpleAsyncTaskExecutor on the fly [4]. This default executor is managed by the container instance itself [4]. 2. Custom Executor: If you provide a custom executor, ownership remains with the caller (e.g., an application-defined bean). The listener container does not shut down or close a custom-provided executor when the container stops [5]. The container simply stops submitting tasks to it; you are responsible for managing the lifecycle of your custom thread pool (e.g., shutting it down as part of the application context's closing process) [5]. Shutdown Behavior: - The listener container's stop process is governed by the ContainerProperties.shutdownTimeout (default 10,000 ms) [6][1][3]. - When the container stops, it signals the consumer thread to stop polling [5]. The container will wait up to the shutdownTimeout for the consumer thread to finish its current processing (e.g., finishing the current record/batch) before completing the stop operation [6][1][5]. - Because the default SimpleAsyncTaskExecutor (when created by the container) does not inherently wait for task termination by default, the container's stop logic relies on its own internal coordination to ensure a graceful exit within the configured timeout [5]. If using a custom executor, ensure it is configured to handle graceful shutdowns if task completion is required after the container stops [5].
Citations:
- 1: https://docs.spring.io/spring-kafka/api/org/springframework/kafka/listener/ContainerProperties.html
- 2: https://docs.spring.io/spring-kafka/docs/3.3.x/api/org/springframework/kafka/listener/ContainerProperties.html
- 3: https://github.com/spring-projects/spring-kafka/blob/master/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java
- 4: https://github.com/spring-projects/spring-kafka/blob/master/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
- 5: ConcurrentMessageListenerContainer isChildRunning API is returning false even though active MessageListenerContainer instances are processing messages. spring-projects/spring-kafka#3338
- 6: https://docs.spring.io/spring-kafka/reference/kafka/container-props.html
🌐 Web query:
Spring Kafka ContainerProperties setListenerTaskExecutor javadoc consumer thread task executor
💡 Result:
In Spring for Apache Kafka, the setListenerTaskExecutor method on the ContainerProperties class is used to define a custom TaskExecutor that manages the threads responsible for polling the Kafka consumer [1][2]. Key points regarding the listener task executor: Purpose: It specifies the executor used to run the threads that perform the Kafka polling loop [1][2]. A dedicated task executor is essential for managing thread allocation, especially in concurrent scenarios [3][4]. Default Behavior: If no custom executor is provided, the container uses a default SimpleAsyncTaskExecutor [3][5]. This default executor generates thread names based on the container bean name (typically following the pattern -C-n, or -m-C-n for concurrent containers) [3][4]. Configuration: When using a ConcurrentMessageListenerContainer, the provided executor is utilized to spawn a thread for each consumer instance (defined by the concurrency setting) [4][6]. If you use a pooled task executor, you must ensure it has sufficient threads available to handle the total concurrency across all containers using it [4]. Virtual Threads: You can set a listener task executor configured for virtual threads to improve performance and resource utilization [7]. Modern Spring Boot versions can often handle this automatically via the spring.threads.virtual.enabled property, but you can also configure it programmatically using setListenerTaskExecutor [7]. Note on Property Names: In some older versions or documentation references, you may see this referred to as the consumer executor [4][8]. The method setListenerTaskExecutor was introduced (notably since version 2.8.9) to provide a clear, explicit way to manage these polling threads [2].
Citations:
- 1: https://docs.spring.io/spring-kafka/api/org/springframework/kafka/listener/ContainerProperties.html
- 2: https://github.com/spring-projects/spring-kafka/blob/master/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java
- 3: https://docs.spring.io/spring-kafka/reference/kafka/container-props.html
- 4: https://docs.spring.io/spring-kafka/reference/kafka/receiving-messages/container-thread-naming.html
- 5: https://docs.spring.io/spring-kafka/docs/3.1.x/reference/kafka/container-props.html
- 6: https://docs.spring.io/spring-kafka/docs/3.1.x/reference/kafka/receiving-messages/container-thread-naming.html
- 7: Concurrency and Virtual Threads spring-projects/spring-kafka#2915
- 8: https://stackoverflow.com/questions/47862458/how-to-set-a-custom-consumer-task-executor-for-spring-kafka-when-using-auto-con
Avoid creating an unmanaged fixed pool here. KafkaMessageListenerContainer still uses a single consumer loop, so maxPoolSize won’t add record-processing parallelism. The local ExecutorService is never shut down; make it a managed bean or close it when the container stops.
🤖 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
`@core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterConsumerConfig.java`
around lines 163 - 166, Update the customExecutorEnabled branch in
PersisterConsumerConfig so the fixed-thread-pool ExecutorService is
lifecycle-managed: either expose it as a managed bean or register shutdown when
the KafkaMessageListenerContainer stops. Do not create an unmanaged local
executor, and preserve the existing listener task-executor configuration.
| Object a = message.get("attempts"); | ||
| // A dead-letter envelope with no numeric attempts (a foreign/raw record) is treated as | ||
| // already at the ceiling, so it parks instead of being granted a fresh retry budget. | ||
| attempts = (a instanceof Number) ? ((Number) a).intValue() : maxRetries; |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Reject invalid DLQ attempt counters.
Number.intValue() can turn negative or overflowing envelope values into counters below maxRetries, producing an effectively unbounded requeue loop. Treat any non-integral or out-of-range value as already at the retry ceiling.
Proposed fix
Object a = message.get("attempts");
-attempts = (a instanceof Number) ? ((Number) a).intValue() : maxRetries;
+if (a instanceof Number number
+ && number.longValue() >= 0
+ && number.longValue() <= maxRetries
+ && number.doubleValue() == number.longValue()) {
+ attempts = Math.toIntExact(number.longValue());
+} else {
+ attempts = maxRetries;
+}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| Object a = message.get("attempts"); | |
| // A dead-letter envelope with no numeric attempts (a foreign/raw record) is treated as | |
| // already at the ceiling, so it parks instead of being granted a fresh retry budget. | |
| attempts = (a instanceof Number) ? ((Number) a).intValue() : maxRetries; | |
| Object a = message.get("attempts"); | |
| // A dead-letter envelope with no numeric attempts (a foreign/raw record) is treated as | |
| // already at the ceiling, so it parks instead of being granted a fresh retry budget. | |
| if (a instanceof Number number | |
| && number.longValue() >= 0 | |
| && number.longValue() <= maxRetries | |
| && number.doubleValue() == number.longValue()) { | |
| attempts = Math.toIntExact(number.longValue()); | |
| } else { | |
| attempts = maxRetries; | |
| } |
🤖 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
`@core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterMessageListener.java`
around lines 70 - 73, Update the attempts parsing in PersisterMessageListener so
only integral numeric values within the valid retry-counter range are accepted;
treat negative, overflowing, non-integral, and non-numeric values as maxRetries.
Avoid Number.intValue() silently narrowing invalid values, while preserving the
existing ceiling behavior for foreign or raw DLQ records.
| for (String record : records) { | ||
| try { | ||
| persistService.persist(topic, record); | ||
| persisted++; | ||
| } catch (Exception e) { | ||
| DbExceptionClassifier.Kind kind = DbExceptionClassifier.classify(e); | ||
| if (kind == DbExceptionClassifier.Kind.BENIGN) { | ||
| duplicates++; | ||
| } else if (kind == DbExceptionClassifier.Kind.TRANSIENT) { | ||
| log.warn("Transient failure during record-level isolation for topic {} - retrying whole message in place", topic, e); | ||
| throw new TransientPersistException("Transient DB failure isolating record for topic " + topic, e); | ||
| } else if (!fromDlq) { | ||
| sendToDlq(topic, record, 1, e); | ||
| deadLettered++; | ||
| } else if (attempts < maxRetries) { | ||
| sendToDlq(topic, record, attempts + 1, e); | ||
| deadLettered++; | ||
| } else { | ||
| sendToParking(topic, record, attempts, e); | ||
| parked++; | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Defer DLQ publication until the isolation sweep succeeds without transient errors.
If an early record is dead-lettered and a later record fails transiently, Line 159 retries the original message, causing the early record to be published to the DLQ again on every retry. Stage permanent failures and publish them only after the complete sweep encounters no transient failure.
🤖 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
`@core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterMessageListener.java`
around lines 149 - 169, The record-isolation loop in PersisterMessageListener
must defer DLQ and parking publication until the entire sweep completes without
a transient failure. Stage each permanent failure and its retry metadata during
the loop, preserve benign/transient handling, then publish the staged entries
only after the loop succeeds, updating deadLettered and parked counts at
publication time.
| // Permanent / poison (incl. undeserialisable records): park a self-describing envelope (source | ||
| // topic + error reason), not the bare value, so a terminally-parked record can be triaged/replayed. | ||
| // Blocking send -> throws on failure, so a failed park is not committed (DefaultErrorHandler | ||
| // re-seeks instead of dropping the record). | ||
| Map<String, Object> parked = new HashMap<>(); | ||
| parked.put("source", record.topic()); | ||
| parked.put("body", record.value()); | ||
| parked.put("error", ex == null ? null : ex.toString()); | ||
| parked.put("ts", System.currentTimeMillis()); | ||
| customKafkaTemplate.send(parkingTopic, parked); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map the target file and nearby symbols first
ast-grep outline core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterProducerConfig.java --view expanded || true
# Read the relevant section with line numbers
sed -n '1,220p' core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterProducerConfig.java | cat -n
# Find ErrorHandlingDeserializer and recoverer usage in the module
rg -n "ErrorHandlingDeserializer|DefaultErrorHandler|DeadLetterPublishingRecoverer|parkingTopic|customKafkaTemplate|Deserializer" core-services/egov-persister/src/main/java core-services/egov-persister/src/main/resourcesRepository: egovernments/Digit-Core
Length of output: 10238
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '1,240p' core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterConsumerConfig.java | cat -n
printf '\n---\n'
sed -n '1,240p' core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterBatchConsumerConfig.java | cat -n
printf '\n---\n'
sed -n '1,260p' core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterMessageListener.java | cat -nRepository: egovernments/Digit-Core
Length of output: 30822
Preserve raw bytes in the parking envelope
PersisterProducerConfig.java:66-71 parks record.value(), which is null for deserialization failures under ErrorHandlingDeserializer. Copy the raw bytes from the exception header into body (or otherwise restore them) so poison records can be replayed.
🤖 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
`@core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/PersisterProducerConfig.java`
around lines 62 - 71, Update the parking-envelope construction in the producer
error-handling flow to populate parked.body with the original raw bytes from the
deserialization exception header when record.value() is null, while preserving
the existing value for successfully deserialized records. Use the available
exception/header metadata associated with the failed record so poison messages
remain replayable.
| // Step 1: Parse all documents and pair with their applicable mappings | ||
| // Using LinkedHashMap to preserve message order | ||
| Map<Object, List<Mapping>> documentToMappings = new LinkedHashMap<>(); | ||
|
|
||
| for (String json : jsons) { | ||
| Object document = Configuration.defaultConfiguration().jsonProvider().parse(json); | ||
| applicableMappings.put(document, filterMappings(map.get(topic), document)); | ||
| documentToMappings.put(document, filterMappings(map.get(topic), document)); | ||
| } | ||
|
|
||
| applicableMappings.forEach((jsonObj, mappings) -> { | ||
| // Step 2: Group documents by mapping (using mapping name as key for identity) | ||
| // This handles the case where different messages might have different versions | ||
| Map<String, List<Object>> mappingNameToDocuments = new LinkedHashMap<>(); | ||
| Map<String, Mapping> mappingNameToMapping = new LinkedHashMap<>(); | ||
|
|
||
| for (Map.Entry<Object, List<Mapping>> entry : documentToMappings.entrySet()) { |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔴 Critical | ⚡ Quick win
Prevent silent deduplication of identical JSON messages.
Using the parsed Object document as a key in a LinkedHashMap relies on its content-based equals() method. If a batch contains multiple identical JSON messages, subsequent messages will overwrite the first one, silently dropping them from the batch. Use a List<Map.Entry> instead to preserve all documents and maintain the exact multiplicity of the batch.
🐛 Proposed fix to prevent deduplication
- // Step 1: Parse all documents and pair with their applicable mappings
- // Using LinkedHashMap to preserve message order
- Map<Object, List<Mapping>> documentToMappings = new LinkedHashMap<>();
-
- for (String json : jsons) {
- Object document = Configuration.defaultConfiguration().jsonProvider().parse(json);
- documentToMappings.put(document, filterMappings(map.get(topic), document));
- }
-
- // Step 2: Group documents by mapping (using mapping name as key for identity)
- // This handles the case where different messages might have different versions
- Map<String, List<Object>> mappingNameToDocuments = new LinkedHashMap<>();
- Map<String, Mapping> mappingNameToMapping = new LinkedHashMap<>();
-
- for (Map.Entry<Object, List<Mapping>> entry : documentToMappings.entrySet()) {
+ // Step 1: Parse all documents and pair with their applicable mappings
+ // Using a List to preserve message order and avoid deduplicating identical JSON payloads
+ List<java.util.Map.Entry<Object, List<Mapping>>> documentToMappings = new ArrayList<>();
+
+ for (String json : jsons) {
+ Object document = Configuration.defaultConfiguration().jsonProvider().parse(json);
+ documentToMappings.add(new java.util.AbstractMap.SimpleEntry<>(document, filterMappings(map.get(topic), document)));
+ }
+
+ // Step 2: Group documents by mapping (using mapping name as key for identity)
+ // This handles the case where different messages might have different versions
+ Map<String, List<Object>> mappingNameToDocuments = new LinkedHashMap<>();
+ Map<String, Mapping> mappingNameToMapping = new LinkedHashMap<>();
+
+ for (java.util.Map.Entry<Object, List<Mapping>> entry : documentToMappings) {📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| // Step 1: Parse all documents and pair with their applicable mappings | |
| // Using LinkedHashMap to preserve message order | |
| Map<Object, List<Mapping>> documentToMappings = new LinkedHashMap<>(); | |
| for (String json : jsons) { | |
| Object document = Configuration.defaultConfiguration().jsonProvider().parse(json); | |
| applicableMappings.put(document, filterMappings(map.get(topic), document)); | |
| documentToMappings.put(document, filterMappings(map.get(topic), document)); | |
| } | |
| applicableMappings.forEach((jsonObj, mappings) -> { | |
| // Step 2: Group documents by mapping (using mapping name as key for identity) | |
| // This handles the case where different messages might have different versions | |
| Map<String, List<Object>> mappingNameToDocuments = new LinkedHashMap<>(); | |
| Map<String, Mapping> mappingNameToMapping = new LinkedHashMap<>(); | |
| for (Map.Entry<Object, List<Mapping>> entry : documentToMappings.entrySet()) { | |
| // Step 1: Parse all documents and pair with their applicable mappings | |
| // Using a List to preserve message order and avoid deduplicating identical JSON payloads | |
| List<java.util.Map.Entry<Object, List<Mapping>>> documentToMappings = new ArrayList<>(); | |
| for (String json : jsons) { | |
| Object document = Configuration.defaultConfiguration().jsonProvider().parse(json); | |
| documentToMappings.add(new java.util.AbstractMap.SimpleEntry<>(document, filterMappings(map.get(topic), document))); | |
| } | |
| // Step 2: Group documents by mapping (using mapping name as key for identity) | |
| // This handles the case where different messages might have different versions | |
| Map<String, List<Object>> mappingNameToDocuments = new LinkedHashMap<>(); | |
| Map<String, Mapping> mappingNameToMapping = new LinkedHashMap<>(); | |
| for (java.util.Map.Entry<Object, List<Mapping>> entry : documentToMappings) { |
🤖 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
`@core-services/egov-persister/src/main/java/org/egov/infra/persist/service/PersistService.java`
around lines 74 - 88, Replace the documentToMappings LinkedHashMap in the
persist flow with a list of document-to-mappings entries so identical parsed
JSON documents remain separate batch items. Update the subsequent iteration to
process each entry in insertion order while preserving every message’s
applicable mappings and exact multiplicity.
| for (Object document : documents) { | ||
| try { | ||
| List<Object[]> rows = persistRepository.getRows(jsonMaps, document, basePath); | ||
| if (rows.isEmpty()) { | ||
| skippedDocuments++; | ||
| } | ||
| aggregatedRows.addAll(rows); | ||
| } catch (Exception e) { | ||
| skippedDocuments++; | ||
| log.warn("Failed to extract rows for basePath '{}': {}", basePath, e.getMessage()); | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win
Remove the try-catch block to ensure malformed records are routed to the DLQ.
Swallowing exceptions from getRows prevents extraction failures (such as parsing errors or unexpected structural mismatches) from propagating. If a document fails extraction due to bad data, the batch will succeed without it, and the listener will commit the offset without ever sending the bad document to the dead-letter queue.
Remove the try-catch block so extraction errors bubble up, triggering the upstream listener's record-level isolation and DLQ routing.
🐛 Proposed fix to propagate extraction failures
for (Object document : documents) {
- try {
- List<Object[]> rows = persistRepository.getRows(jsonMaps, document, basePath);
- if (rows.isEmpty()) {
- skippedDocuments++;
- }
- aggregatedRows.addAll(rows);
- } catch (Exception e) {
- skippedDocuments++;
- log.warn("Failed to extract rows for basePath '{}': {}", basePath, e.getMessage());
- }
+ List<Object[]> rows = persistRepository.getRows(jsonMaps, document, basePath);
+ if (rows.isEmpty()) {
+ skippedDocuments++;
+ }
+ aggregatedRows.addAll(rows);
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| for (Object document : documents) { | |
| try { | |
| List<Object[]> rows = persistRepository.getRows(jsonMaps, document, basePath); | |
| if (rows.isEmpty()) { | |
| skippedDocuments++; | |
| } | |
| aggregatedRows.addAll(rows); | |
| } catch (Exception e) { | |
| skippedDocuments++; | |
| log.warn("Failed to extract rows for basePath '{}': {}", basePath, e.getMessage()); | |
| } | |
| } | |
| for (Object document : documents) { | |
| List<Object[]> rows = persistRepository.getRows(jsonMaps, document, basePath); | |
| if (rows.isEmpty()) { | |
| skippedDocuments++; | |
| } | |
| aggregatedRows.addAll(rows); | |
| } |
🤖 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
`@core-services/egov-persister/src/main/java/org/egov/infra/persist/service/PersistService.java`
around lines 118 - 129, Remove the try-catch around persistRepository.getRows in
the document-processing loop so extraction exceptions propagate to the upstream
listener for record-level isolation and DLQ routing. Keep the existing
skippedDocuments increment for empty results, and do not increment it or
log-and-continue when getRows fails.
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 limitations.
⚠️ Outside diff range comments (2)
core-services/boundary-service/README.md (1)
56-65: 🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick winClarify the parent-ordering contract.
Lines 56-60 require batches to use already-persisted parents and recommend top-down creation, but line 65 says ordering across levels is not required. Clarify that Kafka ordering is unnecessary only when callers retry
PARENT_NOT_FOUND; parent-before-child persistence is still required for successful validation.Proposed wording
- so at-least-once redelivery and caller re-submission are safe no-ops. Bulk messages are keyed by the batch's shared parent code, so sibling batches under one parent stay ordered on the same partition. + so at-least-once redelivery and caller resubmission are safe no-ops. Parent-before-child submission is still required; Kafka partition ordering across hierarchy levels is not relied upon because callers retry `PARENT_NOT_FOUND`.🤖 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 `@core-services/boundary-service/README.md` around lines 56 - 65, Clarify the README’s parent-ordering contract by reconciling the top-down batching guidance with the statement that ordering across levels is not required: state that Kafka or global submission ordering is unnecessary only when callers retry PARENT_NOT_FOUND, while each child batch still requires its parent to be persisted before validation succeeds.core-services/libraries/tracer/src/main/java/org/egov/tracer/kafka/KafkaTemplateLoggingInterceptors.java (1)
42-58: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winClear MDC after processing each record to prevent context leakage.
In the
ConsumerInterceptor.onConsumemethod,TracerKafkaMdcUtil.applyMdcFromRecordsets the MDC context for the current thread (the Kafka consumer poll thread). Because the context is not cleared at the end of the loop, thecorrelationIdandtenantIdof the last record in the batch will leak and remain on the poll thread. This can result in incorrect correlation IDs appearing in internal Kafka logs or other interceptors executing on this thread.Add a call to
TracerKafkaMdcUtil.clearMdc()at the end of the loop to ensure the thread's MDC is left clean.♻️ Proposed fix
} else { log.info(RECEIVED_MESSAGE, consumerRecord.topic(), consumerRecord.topic(), consumerRecord.key()); } + TracerKafkaMdcUtil.clearMdc(); } return consumerRecords; }🤖 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 `@core-services/libraries/tracer/src/main/java/org/egov/tracer/kafka/KafkaTemplateLoggingInterceptors.java` around lines 42 - 58, Update the record-processing loop in ConsumerInterceptor.onConsume to call TracerKafkaMdcUtil.clearMdc() after each record’s logging completes, ensuring correlationId and tenantId do not remain on the Kafka poll thread. Keep the existing MDC application and logging behavior unchanged.
🧹 Nitpick comments (1)
core-services/libraries/tracer/src/main/java/org/egov/tracer/kafka/TracerKafkaMdcUtil.java (1)
67-81: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winSupport raw JSON parsing in
correlationIdFromBody.If
consumerRecord.value()is a raw JSONStringorbyte[](e.g., when the listener is configured with aStringDeserializerorByteArrayDeserializer),objectMapper.convertValuewill throw anIllegalArgumentExceptionrather than parsing the JSON. This results in the correlation ID extraction silently failing and returningnull.To safely handle both parsed POJOs and raw JSON payload formats, add explicit parsing paths for strings and byte arrays before falling back to
convertValue.♻️ Proposed refactor
`@SuppressWarnings`("unchecked") private static String correlationIdFromBody(Object value) { try { - Map<String, Object> requestMap = objectMapper.convertValue(value, Map.class); + Map<String, Object> requestMap; + if (value instanceof String) { + requestMap = objectMapper.readValue((String) value, Map.class); + } else if (value instanceof byte[]) { + requestMap = objectMapper.readValue((byte[]) value, Map.class); + } else { + requestMap = objectMapper.convertValue(value, Map.class); + } Object requestInfo = requestMap.containsKey(REQUEST_INFO_FIELD_NAME_IN_JAVA_CLASS_CASE) ? requestMap.get(REQUEST_INFO_FIELD_NAME_IN_JAVA_CLASS_CASE)🤖 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 `@core-services/libraries/tracer/src/main/java/org/egov/tracer/kafka/TracerKafkaMdcUtil.java` around lines 67 - 81, Update correlationIdFromBody to explicitly parse String and byte[] values as JSON before the existing objectMapper.convertValue fallback, while preserving the current request-info and correlation-ID extraction behavior for parsed objects. Ensure both raw payload formats are handled without silently returning null due to convertValue rejecting them.
🤖 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
`@core-services/libraries/tracer/src/main/java/org/egov/tracer/kafka/TracerKafkaMdcUtil.java`:
- Around line 13-16: Replace the removed static
org.springframework.util.StringUtils.isEmpty import and update all affected
checks in TracerKafkaMdcUtil to use the Spring 6-compatible StringUtils.hasText
or hasLength semantics, preserving each existing empty-value behavior across the
referenced methods.
---
Outside diff comments:
In `@core-services/boundary-service/README.md`:
- Around line 56-65: Clarify the README’s parent-ordering contract by
reconciling the top-down batching guidance with the statement that ordering
across levels is not required: state that Kafka or global submission ordering is
unnecessary only when callers retry PARENT_NOT_FOUND, while each child batch
still requires its parent to be persisted before validation succeeds.
In
`@core-services/libraries/tracer/src/main/java/org/egov/tracer/kafka/KafkaTemplateLoggingInterceptors.java`:
- Around line 42-58: Update the record-processing loop in
ConsumerInterceptor.onConsume to call TracerKafkaMdcUtil.clearMdc() after each
record’s logging completes, ensuring correlationId and tenantId do not remain on
the Kafka poll thread. Keep the existing MDC application and logging behavior
unchanged.
---
Nitpick comments:
In
`@core-services/libraries/tracer/src/main/java/org/egov/tracer/kafka/TracerKafkaMdcUtil.java`:
- Around line 67-81: Update correlationIdFromBody to explicitly parse String and
byte[] values as JSON before the existing objectMapper.convertValue fallback,
while preserving the current request-info and correlation-ID extraction behavior
for parsed objects. Ensure both raw payload formats are handled without silently
returning null due to convertValue rejecting them.
🪄 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: defaults
Review profile: CHILL
Plan: Pro
Run ID: 5d3348a3-3a92-46fb-bafa-25347d780652
📒 Files selected for processing (14)
core-services/boundary-service/CHANGELOG.mdcore-services/boundary-service/README.mdcore-services/boundary-service/pom.xmlcore-services/egov-persister/CHANGELOG.mdcore-services/egov-persister/README.mdcore-services/egov-persister/pom.xmlcore-services/libraries/tracer/CHANGELOG.mdcore-services/libraries/tracer/pom.xmlcore-services/libraries/tracer/readme.mdcore-services/libraries/tracer/src/main/java/org/egov/tracer/config/TracerConfiguration.javacore-services/libraries/tracer/src/main/java/org/egov/tracer/kafka/KafkaTemplateLoggingInterceptors.javacore-services/libraries/tracer/src/main/java/org/egov/tracer/kafka/MdcRecordInterceptor.javacore-services/libraries/tracer/src/main/java/org/egov/tracer/kafka/TracerKafkaMdcUtil.javacore-services/libraries/tracer/src/main/resources/tracer.properties
| import static java.util.Objects.isNull; | ||
| import static org.egov.tracer.constants.TracerConstants.*; | ||
| import static org.springframework.util.StringUtils.isEmpty; | ||
|
|
There was a problem hiding this comment.
🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win
Fix compilation error: StringUtils.isEmpty is removed in Spring 6.
The project uses Spring Boot 3.4.5 (Spring Framework 6+), where org.springframework.util.StringUtils.isEmpty has been completely removed after being deprecated in Spring 5.3. This will cause a compilation failure.
Replace isEmpty with !StringUtils.hasText or !StringUtils.hasLength (or ObjectUtils.isEmpty).
🛠 Proposed fix
import static java.util.Objects.isNull;
import static org.egov.tracer.constants.TracerConstants.*;
-import static org.springframework.util.StringUtils.isEmpty;
+import org.springframework.util.StringUtils;
/** Carries correlationId + tenantId across Kafka via record headers, and rebuilds MDC on consume. */Update the method calls:
public static void applyMdcFromRecord(ConsumerRecord<?, ?> consumerRecord) {
String correlationId = headerValue(consumerRecord, CORRELATION_ID_HEADER);
- if (isEmpty(correlationId))
+ if (!StringUtils.hasLength(correlationId))
correlationId = correlationIdFromBody(consumerRecord.value());
setOrRemove(CORRELATION_ID_MDC, correlationId); private static void addHeaderIfAbsent(ProducerRecord<?, ?> producerRecord, String headerName, String value) {
- if (!isEmpty(value) && producerRecord.headers().lastHeader(headerName) == null)
+ if (StringUtils.hasLength(value) && producerRecord.headers().lastHeader(headerName) == null)
producerRecord.headers().add(headerName, value.getBytes(StandardCharsets.UTF_8));
} private static void setOrRemove(String key, String value) {
- if (isEmpty(value))
+ if (!StringUtils.hasLength(value))
MDC.remove(key);
else
MDC.put(key, value);
}Also applies to: 32-40, 48-51, 60-65
🤖 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
`@core-services/libraries/tracer/src/main/java/org/egov/tracer/kafka/TracerKafkaMdcUtil.java`
around lines 13 - 16, Replace the removed static
org.springframework.util.StringUtils.isEmpty import and update all affected
checks in TracerKafkaMdcUtil to use the Spring 6-compatible StringUtils.hasText
or hasLength semantics, preserving each existing empty-value behavior across the
referenced methods.
# Conflicts: # core-services/egov-persister/CHANGELOG.md
…rrFix Hcmpre 2023 tracer kafka corr fix
5 participants
Summary by CodeRabbit
New Features
POST /boundary-relationships/bulk/_createwith per-record success/failure reporting and bulk limits.Bug Fixes
Reliability
ON CONFLICT DO NOTHING).