One message per record (rather than one message carrying the whole batch) preserves per-record
- * isolation on either listener: a single un-insertable record fails/dead-letters on its own without
- * affecting the rest. Because the insert is idempotent, at-least-once redelivery is a safe no-op.
+ *
The message is keyed by the batch's parent code (callers batch siblings under one already-persisted
+ * parent), so batches for the same parent stay ordered on the same partition. A null/mixed parent falls
+ * back to the keyless behaviour of the single path.
*
* @param boundaryRelationships validated and enriched relationships to persist
- * @param requestInfo request info propagated onto each published message
+ * @param requestInfo request info propagated onto the published message
*/
@Override
public void createBulk(List boundaryRelationships, RequestInfo requestInfo) {
if (CollectionUtils.isEmpty(boundaryRelationships))
return;
+ // Convert each validated+enriched contract POJO to the DTO that exposes ancestralMaterializedPath
+ // on the wire (it is @JsonIgnore on BoundaryRelation but @JsonProperty on the DTO), mirroring the
+ // single-create serialization so both paths persist identical rows.
+ List boundaryRelationshipDTOs = new ArrayList<>(boundaryRelationships.size());
for (BoundaryRelation boundaryRelationship : boundaryRelationships) {
- create(BoundaryRelationshipRequest.builder()
- .requestInfo(requestInfo)
- .boundaryRelationship(boundaryRelationship)
- .build());
+ boundaryRelationshipDTOs.add(convertRelationPOJOToDTO(boundaryRelationship));
}
+
+ BulkBoundaryRelationshipRequestDTO batchMessage = BulkBoundaryRelationshipRequestDTO.builder()
+ .requestInfo(requestInfo)
+ .boundaryRelationship(boundaryRelationshipDTOs)
+ .build();
+
+ // Publish the whole validated list as ONE message to the unchanged topic.
+ producer.push(applicationProperties.getCreateBoundaryRelationshipTopic(), resolveBatchKey(boundaryRelationships), batchMessage);
+ }
+
+ /**
+ * Kafka key for a batch: the shared parent code when every record in the batch has the same
+ * (non-null) parent, else null (keyless, i.e. the single-create default-partitioner behaviour).
+ * Keying by parent keeps sibling batches under one parent ordered on the same partition; a mixed or
+ * root batch must not be forced onto one partition, so it falls back to keyless.
+ */
+ private String resolveBatchKey(List boundaryRelationships) {
+ String firstParent = boundaryRelationships.get(0).getParent();
+ if (firstParent == null)
+ return null;
+ for (BoundaryRelation boundaryRelationship : boundaryRelationships) {
+ if (!firstParent.equals(boundaryRelationship.getParent()))
+ return null;
+ }
+ return firstParent;
+ }
+
+ /**
+ * Copies a validated+enriched {@link BoundaryRelation} into a {@link BoundaryRelationshipDTO},
+ * carrying over the enriched {@code ancestralMaterializedPath} explicitly (it is not copied by
+ * BeanUtils onto the wire because it is {@code @JsonIgnore} on the source). Mirrors the field copy
+ * that {@link #convertContractPOJOToDTO} performs for the single-create path.
+ */
+ private BoundaryRelationshipDTO convertRelationPOJOToDTO(BoundaryRelation boundaryRelationship) {
+ BoundaryRelationshipDTO boundaryRelationshipDTO = new BoundaryRelationshipDTO();
+ BeanUtils.copyProperties(boundaryRelationship, boundaryRelationshipDTO);
+ boundaryRelationshipDTO.setAncestralMaterializedPath(boundaryRelationship.getAncestralMaterializedPath());
+ return boundaryRelationshipDTO;
}
/**
diff --git a/core-services/boundary-service/src/main/java/digit/service/BoundaryRelationshipService.java b/core-services/boundary-service/src/main/java/digit/service/BoundaryRelationshipService.java
index 666ca629647..3b382831eb0 100644
--- a/core-services/boundary-service/src/main/java/digit/service/BoundaryRelationshipService.java
+++ b/core-services/boundary-service/src/main/java/digit/service/BoundaryRelationshipService.java
@@ -158,14 +158,14 @@ public BulkBoundaryRelationshipResponse createBulkBoundaryRelationship(BulkBound
}
}
- // Persistence is delegated to egov-persister (no direct DB write from this service): each
- // validated + enriched record is published, one message per record, to the save-boundary-relationship
- // topic, which egov-persister writes via an idempotent INSERT ... ON CONFLICT DO NOTHING. Publishing
- // is blocking (CustomKafkaTemplate.send().get()); a publish failure (broker unreachable within
- // max.block.ms, serialization) is reported transient so the caller retries the affected records —
- // re-publishing is a safe no-op because the insert is idempotent. One message per record preserves
- // per-record isolation on the persister side regardless of whether it runs the normal or the
- // (optional) batch listener.
+ // Persistence is delegated to egov-persister (no direct DB write from this service): the whole
+ // validated + enriched list is published as ONE message (relationships as an array) to the
+ // save-boundary-relationship topic, which egov-persister maps to N rows and writes in a single
+ // batchUpdate via the idempotent INSERT ... ON CONFLICT DO NOTHING. Publishing is blocking
+ // (CustomKafkaTemplate.send().get()); a publish failure (broker unreachable within max.block.ms,
+ // serialization) is reported transient so the caller retries the batch — re-publishing is a safe
+ // no-op because the insert is idempotent and duplicates are skipped by ON CONFLICT without
+ // aborting the batch.
List successfulRelationships = validatedRelationships;
if (!CollectionUtils.isEmpty(validatedRelationships)) {
try {
diff --git a/core-services/boundary-service/src/main/java/digit/web/models/BulkBoundaryRelationshipRequestDTO.java b/core-services/boundary-service/src/main/java/digit/web/models/BulkBoundaryRelationshipRequestDTO.java
new file mode 100644
index 00000000000..d44a56d640b
--- /dev/null
+++ b/core-services/boundary-service/src/main/java/digit/web/models/BulkBoundaryRelationshipRequestDTO.java
@@ -0,0 +1,42 @@
+package digit.web.models;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.egov.common.contract.request.RequestInfo;
+
+import java.util.List;
+
+/**
+ * Kafka payload for persisting a validated batch of boundary relationships in ONE message.
+ *
+ *
The whole validated list is published as a single message under the {@code BoundaryRelationship}
+ * key (an ARRAY here, whereas the single-create {@link BoundaryRelationshipRequestDTO} publishes the
+ * same key as a single object). The persister's {@code save-boundary-relationship} mapping reads this
+ * with an array base path ({@code $.BoundaryRelationship.*}) so {@code PersistRepository.getRows}
+ * emits one row per element and the listener does ONE {@code jdbcTemplate.batchUpdate} for the whole
+ * message.
+ *
+ *
{@code RequestInfo} is retained at the top level so the persister's version filter
+ * ({@code $.RequestInfo.ver}) still selects the mapping exactly as it does for the single message.
+ *
+ *
The elements are {@link BoundaryRelationshipDTO} (not the contract {@link BoundaryRelation})
+ * because {@code BoundaryRelation.ancestralMaterializedPath} is {@code @JsonIgnore}; the DTO exposes it
+ * as {@code ancestralMaterializedPath}, so the enriched path is carried on the wire and persisted —
+ * identical to the single-create path, which serializes the same DTO field.
The batch change must publish the WHOLE validated list as EXACTLY ONE Kafka message (previously it
+ * looped and published one message per record). These tests pin that contract: one push per bulk call,
+ * carrying an array whose size equals the input size, on the unchanged topic, with the enriched
+ * ancestralMaterializedPath preserved and the batch keyed by the shared parent.
+ */
+@ExtendWith(MockitoExtension.class)
+class BoundaryRelationshipRepositoryImplBulkTest {
+
+ private static final String TOPIC = "save-boundary-relationship";
+
+ @Mock
+ private Producer producer;
+
+ @Mock
+ private ApplicationProperties applicationProperties;
+
+ @InjectMocks
+ private BoundaryRelationshipRepositoryImpl repository;
+
+ private RequestInfo requestInfo() {
+ return RequestInfo.builder().apiId("boundary").ver("1.0").build();
+ }
+
+ private BoundaryRelation relation(String code, String parent, String amp) {
+ return BoundaryRelation.builder()
+ .id("id-" + code)
+ .code(code)
+ .tenantId("mz")
+ .hierarchyType("ADMIN")
+ .boundaryType("Village")
+ .parent(parent)
+ .ancestralMaterializedPath(amp)
+ .auditDetails(AuditDetails.builder()
+ .createdBy("u1").createdTime(1L).lastModifiedBy("u1").lastModifiedTime(1L).build())
+ .build();
+ }
+
+ @Test
+ void createBulk_publishesExactlyOneMessageCarryingTheWholeList() {
+ lenient().when(applicationProperties.getCreateBoundaryRelationshipTopic()).thenReturn(TOPIC);
+
+ List input = new ArrayList<>();
+ for (int i = 0; i < 5; i++) {
+ input.add(relation("V" + i, "P1", "R1|P1"));
+ }
+
+ repository.createBulk(input, requestInfo());
+
+ ArgumentCaptor topicCaptor = ArgumentCaptor.forClass(String.class);
+ ArgumentCaptor keyCaptor = ArgumentCaptor.forClass(String.class);
+ ArgumentCaptor valueCaptor = ArgumentCaptor.forClass(Object.class);
+
+ // EXACTLY ONE push (was N=5 with the old per-record loop).
+ verify(producer, times(1)).push(topicCaptor.capture(), keyCaptor.capture(), valueCaptor.capture());
+ verify(producer, never()).push(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.any());
+
+ assertEquals(TOPIC, topicCaptor.getValue(), "topic must be unchanged save-boundary-relationship");
+ assertEquals("P1", keyCaptor.getValue(), "batch keyed by the shared parent code");
+
+ Object value = valueCaptor.getValue();
+ assertNotNull(value);
+ assertEquals(BulkBoundaryRelationshipRequestDTO.class, value.getClass(), "payload must be the batch DTO");
+ BulkBoundaryRelationshipRequestDTO payload = (BulkBoundaryRelationshipRequestDTO) value;
+
+ assertEquals(5, payload.getBoundaryRelationship().size(), "array size must equal input size");
+ assertNotNull(payload.getRequestInfo(), "RequestInfo retained for persister version filter");
+
+ // ancestralMaterializedPath is @JsonIgnore on BoundaryRelation but must survive on the DTO.
+ assertEquals("R1|P1", payload.getBoundaryRelationship().get(0).getAncestralMaterializedPath());
+ assertEquals("V0", payload.getBoundaryRelationship().get(0).getCode());
+ assertEquals("id-V0", payload.getBoundaryRelationship().get(0).getId());
+ assertNotNull(payload.getBoundaryRelationship().get(0).getAuditDetails());
+ }
+
+ @Test
+ void createBulk_mixedParents_fallsBackToKeylessPush() {
+ lenient().when(applicationProperties.getCreateBoundaryRelationshipTopic()).thenReturn(TOPIC);
+
+ List input = new ArrayList<>();
+ input.add(relation("V0", "P1", "R1|P1"));
+ input.add(relation("V1", "P2", "R1|P2"));
+
+ repository.createBulk(input, requestInfo());
+
+ ArgumentCaptor keyCaptor = ArgumentCaptor.forClass(String.class);
+ verify(producer, times(1)).push(org.mockito.ArgumentMatchers.eq(TOPIC), keyCaptor.capture(), org.mockito.ArgumentMatchers.any());
+ assertEquals(null, keyCaptor.getValue(), "mixed parents must fall back to keyless (null key)");
+ }
+
+ @Test
+ void createBulk_emptyList_publishesNothing() {
+ repository.createBulk(new ArrayList<>(), requestInfo());
+ verify(producer, never()).push(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.any());
+ verify(producer, never()).push(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.any());
+ }
+}
From 0aeb68a10d9e2604c55e96ed9f3f4521afd39651 Mon Sep 17 00:00:00 2001
From: bhaswatmandal-egov
Date: Mon, 13 Jul 2026 16:10:49 +0530
Subject: [PATCH 13/24] Bulk kafka handling
---
.../digit/config/ApplicationProperties.java | 6 +
.../BoundaryRelationshipRepository.java | 18 ++-
.../BoundaryRelationshipRepositoryImpl.java | 37 ++---
.../service/BoundaryRelationshipService.java | 6 +-
.../BulkBoundaryRelationshipRequestDTO.java | 10 +-
.../src/main/resources/application.properties | 4 +
.../src/main/resources/boundary-persister.yml | 31 +++-
...aryRelationshipRepositoryImplBulkTest.java | 11 +-
.../BoundaryRelationshipMappingShapeTest.java | 142 ++++++++++++++++++
9 files changed, 229 insertions(+), 36 deletions(-)
create mode 100644 core-services/egov-persister/src/test/java/org/egov/infra/persist/repository/BoundaryRelationshipMappingShapeTest.java
diff --git a/core-services/boundary-service/src/main/java/digit/config/ApplicationProperties.java b/core-services/boundary-service/src/main/java/digit/config/ApplicationProperties.java
index fbc569aeb7e..c22ed759cd6 100644
--- a/core-services/boundary-service/src/main/java/digit/config/ApplicationProperties.java
+++ b/core-services/boundary-service/src/main/java/digit/config/ApplicationProperties.java
@@ -99,6 +99,12 @@ public class ApplicationProperties {
@Value("${kafka.topics.update.boundary.relationship}")
private String updateBoundaryRelationshipTopic;
+ // Dedicated topic for the batched bulk-create path: createBulk publishes ONE array message here, mapped by
+ // the persister with an array base path ($.BoundaryRelationship.*). Kept separate from the single-create
+ // topic because a single object and an array cannot share one persister queryMap.
+ @Value("${kafka.topics.bulk.create.boundary.relationship.job}")
+ private String bulkCreateBoundaryRelationshipJobTopic;
+
// Upper bound on records accepted by POST /boundary-relationships/bulk/_create. Enforced in the
// service (bean validation is not active in this deployment). Keep in sync with the caller's chunk size.
@Value("${boundary.bulk.max.size:100}")
diff --git a/core-services/boundary-service/src/main/java/digit/repository/BoundaryRelationshipRepository.java b/core-services/boundary-service/src/main/java/digit/repository/BoundaryRelationshipRepository.java
index bfd656cdf9d..21017f86ede 100644
--- a/core-services/boundary-service/src/main/java/digit/repository/BoundaryRelationshipRepository.java
+++ b/core-services/boundary-service/src/main/java/digit/repository/BoundaryRelationshipRepository.java
@@ -14,13 +14,17 @@ public interface BoundaryRelationshipRepository {
/**
* Persists the given validated and enriched boundary relationships through egov-persister (no direct
- * DB write): the WHOLE list is published as ONE message to the same save-boundary-relationship topic
- * the single {@link #create} uses, carrying the relationships as an array under the same
- * {@code BoundaryRelationship} key. The publish is blocking (it returns once the broker has accepted
- * the message); egov-persister reads the array (array base path) and writes it as a single
- * batchUpdate through the idempotent INSERT ... ON CONFLICT (tenantId, code, hierarchyType) DO NOTHING,
- * so redelivery is a safe no-op and duplicates are skipped without aborting the batch. Batching is
- * WITHIN the one message, so it needs no persister.batch.topics / persister.bulk.enabled configuration.
+ * DB write): the WHOLE list is published as ONE message to the DEDICATED bulk topic
+ * {@code boundary-relationship-bulk-create-job} (NOT the single {@link #create} topic
+ * {@code save-boundary-relationship}), carrying the relationships as an array under the
+ * {@code BoundaryRelationship} key. A dedicated topic is required because the persister maps this one
+ * with an array base path ({@code $.BoundaryRelationship.*}) while single-create stays single-object
+ * ({@code $.BoundaryRelationship}) — one queryMap has one base path, so the two shapes cannot share a
+ * topic. The publish is blocking (it returns once the broker has accepted the message); egov-persister
+ * reads the array and writes it as a single batchUpdate through the idempotent
+ * INSERT ... ON CONFLICT (tenantId, code, hierarchyType) DO NOTHING, so redelivery is a safe no-op and
+ * duplicates are skipped without aborting the batch. Batching is WITHIN the one message, so it needs no
+ * persister.batch.topics / persister.bulk.enabled configuration.
*
* @param boundaryRelationships validated and enriched relationships to persist
* @param requestInfo request info propagated onto the published message
diff --git a/core-services/boundary-service/src/main/java/digit/repository/impl/BoundaryRelationshipRepositoryImpl.java b/core-services/boundary-service/src/main/java/digit/repository/impl/BoundaryRelationshipRepositoryImpl.java
index 777436f4475..ee846cb7db3 100644
--- a/core-services/boundary-service/src/main/java/digit/repository/impl/BoundaryRelationshipRepositoryImpl.java
+++ b/core-services/boundary-service/src/main/java/digit/repository/impl/BoundaryRelationshipRepositoryImpl.java
@@ -53,22 +53,25 @@ public void create(BoundaryRelationshipRequest boundaryRelationshipRequest) {
/**
* Persists the given validated and enriched boundary relationships through egov-persister rather than
- * a direct JDBC write. The WHOLE validated list is published as ONE message to the SAME topic the
- * single create uses ({@code save-boundary-relationship}), under the same {@code BoundaryRelationship}
- * key but carrying an ARRAY (whereas single-create carries one object). The persister's mapping reads
- * it with an array base path ({@code $.BoundaryRelationship.*}), so {@code PersistRepository.getRows}
- * emits one row per element and the listener performs ONE {@code jdbcTemplate.batchUpdate} for the
- * whole message through the same idempotent
- * {@code INSERT ... ON CONFLICT (tenantId, code, hierarchyType) DO NOTHING} query. One message per
- * batch (instead of one message per record) is what restores batched throughput while keeping the
- * write owned by the persister.
+ * a direct JDBC write. The WHOLE validated list is published as ONE message — an ARRAY under the
+ * {@code BoundaryRelationship} key — to the DEDICATED bulk topic
+ * ({@code boundary-relationship-bulk-create-job}). The persister maps that topic with an array base
+ * path ({@code $.BoundaryRelationship.*}), so {@code PersistRepository.getRows} emits one row per
+ * element and the listener performs ONE {@code jdbcTemplate.batchUpdate} for the whole message through
+ * the idempotent {@code INSERT ... ON CONFLICT (tenantId, code, hierarchyType) DO NOTHING} query. One
+ * message per batch (instead of one message per record) is what restores batched throughput while
+ * keeping the write owned by the persister.
*
- *
Reusing that topic (instead of a dedicated {@code -batch} topic) is deliberate for
- * deployment-safety: {@code save-boundary-relationship} is always consumed by the persister's normal
- * single-record listener. Batching here is WITHIN a single message (the array), so it does not depend
- * on {@code persister.bulk.enabled}: the normal listener maps the array to N rows and batch-inserts
- * them in one transaction. Because the insert is idempotent, at-least-once redelivery is a safe no-op;
- * duplicates within/across messages are silently skipped by ON CONFLICT and never abort the batch.
+ *
A DEDICATED topic (NOT the single-create {@code save-boundary-relationship}) is required for
+ * correctness: single-create publishes {@code BoundaryRelationship} as a single OBJECT, bulk publishes
+ * it as an ARRAY. A persister queryMap has exactly one base path, so the two shapes cannot share a
+ * topic — an array mapping ({@code .*}) mis-reads a single object (JsonPath {@code .*} over an object
+ * yields its property values, not one row) and a single mapping mis-reads an array. Each shape
+ * therefore gets its own topic + queryMap. Batching is WITHIN a single message (the array), so it does
+ * not depend on {@code persister.bulk.enabled}: the normal listener maps the array to N rows and
+ * batch-inserts them in one transaction. Because the insert is idempotent, at-least-once redelivery is
+ * a safe no-op; duplicates within/across messages are silently skipped by ON CONFLICT and never abort
+ * the batch.
*
*
The message is keyed by the batch's parent code (callers batch siblings under one already-persisted
* parent), so batches for the same parent stay ordered on the same partition. A null/mixed parent falls
@@ -95,8 +98,8 @@ public void createBulk(List boundaryRelationships, RequestInfo
.boundaryRelationship(boundaryRelationshipDTOs)
.build();
- // Publish the whole validated list as ONE message to the unchanged topic.
- producer.push(applicationProperties.getCreateBoundaryRelationshipTopic(), resolveBatchKey(boundaryRelationships), batchMessage);
+ // Publish the whole validated list as ONE message to the dedicated bulk topic.
+ producer.push(applicationProperties.getBulkCreateBoundaryRelationshipJobTopic(), resolveBatchKey(boundaryRelationships), batchMessage);
}
/**
diff --git a/core-services/boundary-service/src/main/java/digit/service/BoundaryRelationshipService.java b/core-services/boundary-service/src/main/java/digit/service/BoundaryRelationshipService.java
index 3b382831eb0..88e1847bb70 100644
--- a/core-services/boundary-service/src/main/java/digit/service/BoundaryRelationshipService.java
+++ b/core-services/boundary-service/src/main/java/digit/service/BoundaryRelationshipService.java
@@ -160,8 +160,10 @@ public BulkBoundaryRelationshipResponse createBulkBoundaryRelationship(BulkBound
// Persistence is delegated to egov-persister (no direct DB write from this service): the whole
// validated + enriched list is published as ONE message (relationships as an array) to the
- // save-boundary-relationship topic, which egov-persister maps to N rows and writes in a single
- // batchUpdate via the idempotent INSERT ... ON CONFLICT DO NOTHING. Publishing is blocking
+ // dedicated boundary-relationship-bulk-create-job topic, which egov-persister maps ($.BoundaryRelationship.*)
+ // to N rows and writes in a single batchUpdate via the idempotent INSERT ... ON CONFLICT DO NOTHING.
+ // (Single-create keeps its own save-boundary-relationship topic + single-object mapping; the two
+ // message shapes cannot share a topic.) Publishing is blocking
// (CustomKafkaTemplate.send().get()); a publish failure (broker unreachable within max.block.ms,
// serialization) is reported transient so the caller retries the batch — re-publishing is a safe
// no-op because the insert is idempotent and duplicates are skipped by ON CONFLICT without
diff --git a/core-services/boundary-service/src/main/java/digit/web/models/BulkBoundaryRelationshipRequestDTO.java b/core-services/boundary-service/src/main/java/digit/web/models/BulkBoundaryRelationshipRequestDTO.java
index d44a56d640b..8dba1c79b8d 100644
--- a/core-services/boundary-service/src/main/java/digit/web/models/BulkBoundaryRelationshipRequestDTO.java
+++ b/core-services/boundary-service/src/main/java/digit/web/models/BulkBoundaryRelationshipRequestDTO.java
@@ -14,10 +14,12 @@
*
*
The whole validated list is published as a single message under the {@code BoundaryRelationship}
* key (an ARRAY here, whereas the single-create {@link BoundaryRelationshipRequestDTO} publishes the
- * same key as a single object). The persister's {@code save-boundary-relationship} mapping reads this
- * with an array base path ({@code $.BoundaryRelationship.*}) so {@code PersistRepository.getRows}
- * emits one row per element and the listener does ONE {@code jdbcTemplate.batchUpdate} for the whole
- * message.
+ * same key as a single object). It goes to the DEDICATED bulk topic
+ * ({@code boundary-relationship-bulk-create-job}), whose persister mapping reads it with an array base
+ * path ({@code $.BoundaryRelationship.*}) so {@code PersistRepository.getRows} emits one row per element
+ * and the listener does ONE {@code jdbcTemplate.batchUpdate} for the whole message. It must NOT share the
+ * single-create topic {@code save-boundary-relationship}: a persister queryMap has one base path, and an
+ * array shape and a single-object shape cannot both be read by the same mapping.
*
*
{@code RequestInfo} is retained at the top level so the persister's version filter
* ({@code $.RequestInfo.ver}) still selects the mapping exactly as it does for the single message.
diff --git a/core-services/boundary-service/src/main/resources/application.properties b/core-services/boundary-service/src/main/resources/application.properties
index 010a62b7291..d9deeedf80e 100644
--- a/core-services/boundary-service/src/main/resources/application.properties
+++ b/core-services/boundary-service/src/main/resources/application.properties
@@ -95,6 +95,10 @@ kafka.topics.update.boundary.hierarchy = update-boundary-hierarchy-definition
# optional throughput optimization that lets the persister aggregate a poll into one multi-row insert.
kafka.topics.create.boundary.relationship = save-boundary-relationship
kafka.topics.update.boundary.relationship = update-boundary-relationship
+# Dedicated topic for the batched bulk-create path (one ARRAY message per batch). Kept separate from the
+# single-create topic above because the persister maps this one with an array base path ($.BoundaryRelationship.*)
+# while save-boundary-relationship stays single-object ($.BoundaryRelationship) — the two shapes cannot share a topic.
+kafka.topics.bulk.create.boundary.relationship.job = boundary-relationship-bulk-create-job
boundary.default.offset=0
boundary.default.limit=50
boundary.max.default.limit=300
diff --git a/core-services/boundary-service/src/main/resources/boundary-persister.yml b/core-services/boundary-service/src/main/resources/boundary-persister.yml
index a070352bb8d..7674cc101b5 100644
--- a/core-services/boundary-service/src/main/resources/boundary-persister.yml
+++ b/core-services/boundary-service/src/main/resources/boundary-persister.yml
@@ -75,9 +75,38 @@ serviceMaps:
- jsonPath: $.BoundaryHierarchy.auditDetails.lastModifiedBy
- version: 1.0
- description: Persists the boundary relationship data
+ description: Persists a SINGLE boundary relationship (POST /boundary-relationships/_create -> object payload)
fromTopic: save-boundary-relationship
isTransaction: true
+ queryMaps:
+ - query: INSERT INTO boundary_relationship (id, tenantId, code, hierarchyType, boundaryType, parent, ancestralMaterializedPath, createdTime, createdBy, lastModifiedTime, lastModifiedBy) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (tenantId, code, hierarchyType) DO NOTHING;
+ basePath: $.BoundaryRelationship
+ jsonMaps:
+ - jsonPath: $.BoundaryRelationship.id
+
+ - jsonPath: $.BoundaryRelationship.tenantId
+
+ - jsonPath: $.BoundaryRelationship.code
+
+ - jsonPath: $.BoundaryRelationship.hierarchyType
+
+ - jsonPath: $.BoundaryRelationship.boundaryType
+
+ - jsonPath: $.BoundaryRelationship.parent
+
+ - jsonPath: $.BoundaryRelationship.ancestralMaterializedPath
+
+ - jsonPath: $.BoundaryRelationship.auditDetails.createdTime
+
+ - jsonPath: $.BoundaryRelationship.auditDetails.createdBy
+
+ - jsonPath: $.BoundaryRelationship.auditDetails.lastModifiedTime
+
+ - jsonPath: $.BoundaryRelationship.auditDetails.lastModifiedBy
+ - version: 1.0
+ description: Persists a BATCH of boundary relationships published as ONE array message (POST /boundary-relationships/bulk/_create -> array payload)
+ fromTopic: boundary-relationship-bulk-create-job
+ isTransaction: true
queryMaps:
- query: INSERT INTO boundary_relationship (id, tenantId, code, hierarchyType, boundaryType, parent, ancestralMaterializedPath, createdTime, createdBy, lastModifiedTime, lastModifiedBy) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (tenantId, code, hierarchyType) DO NOTHING;
basePath: $.BoundaryRelationship.*
diff --git a/core-services/boundary-service/src/test/java/digit/repository/impl/BoundaryRelationshipRepositoryImplBulkTest.java b/core-services/boundary-service/src/test/java/digit/repository/impl/BoundaryRelationshipRepositoryImplBulkTest.java
index db5cccd9765..308dd4cfa1a 100644
--- a/core-services/boundary-service/src/test/java/digit/repository/impl/BoundaryRelationshipRepositoryImplBulkTest.java
+++ b/core-services/boundary-service/src/test/java/digit/repository/impl/BoundaryRelationshipRepositoryImplBulkTest.java
@@ -29,13 +29,14 @@
*
*
The batch change must publish the WHOLE validated list as EXACTLY ONE Kafka message (previously it
* looped and published one message per record). These tests pin that contract: one push per bulk call,
- * carrying an array whose size equals the input size, on the unchanged topic, with the enriched
+ * carrying an array whose size equals the input size, on the dedicated bulk topic
+ * (boundary-relationship-bulk-create-job, separate from the single-create topic), with the enriched
* ancestralMaterializedPath preserved and the batch keyed by the shared parent.
*/
@ExtendWith(MockitoExtension.class)
class BoundaryRelationshipRepositoryImplBulkTest {
- private static final String TOPIC = "save-boundary-relationship";
+ private static final String TOPIC = "boundary-relationship-bulk-create-job";
@Mock
private Producer producer;
@@ -66,7 +67,7 @@ private BoundaryRelation relation(String code, String parent, String amp) {
@Test
void createBulk_publishesExactlyOneMessageCarryingTheWholeList() {
- lenient().when(applicationProperties.getCreateBoundaryRelationshipTopic()).thenReturn(TOPIC);
+ lenient().when(applicationProperties.getBulkCreateBoundaryRelationshipJobTopic()).thenReturn(TOPIC);
List input = new ArrayList<>();
for (int i = 0; i < 5; i++) {
@@ -83,7 +84,7 @@ void createBulk_publishesExactlyOneMessageCarryingTheWholeList() {
verify(producer, times(1)).push(topicCaptor.capture(), keyCaptor.capture(), valueCaptor.capture());
verify(producer, never()).push(org.mockito.ArgumentMatchers.anyString(), org.mockito.ArgumentMatchers.any());
- assertEquals(TOPIC, topicCaptor.getValue(), "topic must be unchanged save-boundary-relationship");
+ assertEquals(TOPIC, topicCaptor.getValue(), "bulk batch must go to the dedicated boundary-relationship-bulk-create-job topic, NOT save-boundary-relationship");
assertEquals("P1", keyCaptor.getValue(), "batch keyed by the shared parent code");
Object value = valueCaptor.getValue();
@@ -103,7 +104,7 @@ void createBulk_publishesExactlyOneMessageCarryingTheWholeList() {
@Test
void createBulk_mixedParents_fallsBackToKeylessPush() {
- lenient().when(applicationProperties.getCreateBoundaryRelationshipTopic()).thenReturn(TOPIC);
+ lenient().when(applicationProperties.getBulkCreateBoundaryRelationshipJobTopic()).thenReturn(TOPIC);
List input = new ArrayList<>();
input.add(relation("V0", "P1", "R1|P1"));
diff --git a/core-services/egov-persister/src/test/java/org/egov/infra/persist/repository/BoundaryRelationshipMappingShapeTest.java b/core-services/egov-persister/src/test/java/org/egov/infra/persist/repository/BoundaryRelationshipMappingShapeTest.java
new file mode 100644
index 00000000000..aa4cb994011
--- /dev/null
+++ b/core-services/egov-persister/src/test/java/org/egov/infra/persist/repository/BoundaryRelationshipMappingShapeTest.java
@@ -0,0 +1,142 @@
+package org.egov.infra.persist.repository;
+
+import com.jayway.jsonpath.Configuration;
+import org.egov.infra.persist.web.contract.JsonMap;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * Proves the boundary-relationship "dedicated bulk topic" fix at the persister layer, using the REAL
+ * {@link PersistRepository#getRows} extraction (documents parsed exactly as
+ * {@code PersistService} does: {@code Configuration.defaultConfiguration().jsonProvider().parse(json)}).
+ *
+ *
Two message shapes exist on two different topics:
+ *
+ *
single-create publishes {@code {"BoundaryRelationship": {..one object..}}} to
+ * {@code save-boundary-relationship} — mapped with base path {@code $.BoundaryRelationship} (no wildcard).
+ *
bulk-create publishes {@code {"BoundaryRelationship": [..array..]}} to
+ * {@code boundary-relationship-bulk-create-job} — mapped with base path {@code $.BoundaryRelationship.*}.
+ *
+ * The positive tests prove each shape+mapping pairing extracts the correct number of rows with correct
+ * field values. The negative tests prove WHY the topics must stay separate: crossing a shape with the
+ * wrong base path fails (the original single-topic blocker).
+ */
+class BoundaryRelationshipMappingShapeTest {
+
+ // Column order of the INSERT in boundary-persister.yml
+ private static final String[] FIELDS = {
+ "id", "tenantId", "code", "hierarchyType", "boundaryType", "parent",
+ "ancestralMaterializedPath",
+ "auditDetails.createdTime", "auditDetails.createdBy",
+ "auditDetails.lastModifiedTime", "auditDetails.lastModifiedBy"
+ };
+
+ private List jsonMaps(String prefix) {
+ List maps = new ArrayList<>();
+ for (String f : FIELDS) {
+ JsonMap m = new JsonMap();
+ m.setJsonPath(prefix + f); // e.g. "$.BoundaryRelationship." + "code" or "$.BoundaryRelationship.*." + "code"
+ maps.add(m);
+ }
+ return maps;
+ }
+
+ private String relJson(String id, String code, String parent, String amp) {
+ return "{"
+ + "\"id\":\"" + id + "\","
+ + "\"tenantId\":\"mz\","
+ + "\"code\":\"" + code + "\","
+ + "\"hierarchyType\":\"ADMIN\","
+ + "\"boundaryType\":\"Village\","
+ + "\"parent\":" + (parent == null ? "null" : "\"" + parent + "\"") + ","
+ + "\"ancestralMaterializedPath\":\"" + amp + "\","
+ + "\"auditDetails\":{\"createdTime\":100,\"createdBy\":\"u1\",\"lastModifiedTime\":200,\"lastModifiedBy\":\"u2\"}"
+ + "}";
+ }
+
+ private Object parse(String json) {
+ // identical to PersistService.persist(...)
+ return Configuration.defaultConfiguration().jsonProvider().parse(json);
+ }
+
+ // -------------------- POSITIVE: correct shape + correct mapping --------------------
+
+ @Test
+ void singleObject_onSingleMapping_extractsExactlyOneRowWithCorrectValues() {
+ String json = "{\"RequestInfo\":{\"ver\":\"1.0\"},\"BoundaryRelationship\":"
+ + relJson("id-C1", "C1", null, "C1") + "}";
+
+ List rows = new PersistRepository()
+ .getRows(jsonMaps("$.BoundaryRelationship."), parse(json), "$.BoundaryRelationship");
+
+ assertEquals(1, rows.size(), "single-object message must yield exactly ONE row");
+ Object[] r = rows.get(0);
+ assertEquals(11, r.length, "row must have all 11 columns");
+ assertEquals("id-C1", r[0]);
+ assertEquals("mz", r[1]);
+ assertEquals("C1", r[2]);
+ assertEquals("ADMIN", r[3]);
+ assertEquals("Village", r[4]);
+ assertEquals(null, r[5], "root parent is null");
+ assertEquals("C1", r[6]);
+ assertEquals("u1", r[8]);
+ assertEquals("u2", r[10]);
+ }
+
+ @Test
+ void array_onBulkMapping_extractsOneRowPerElementWithCorrectValues() {
+ StringBuilder arr = new StringBuilder();
+ int n = 5;
+ for (int i = 0; i < n; i++) {
+ if (i > 0) arr.append(",");
+ arr.append(relJson("id-V" + i, "V" + i, "P1", "R1|P1|V" + i));
+ }
+ String json = "{\"RequestInfo\":{\"ver\":\"1.0\"},\"BoundaryRelationship\":[" + arr + "]}";
+
+ List rows = new PersistRepository()
+ .getRows(jsonMaps("$.BoundaryRelationship.*."), parse(json), "$.BoundaryRelationship.*");
+
+ assertEquals(n, rows.size(), "array message must yield ONE row per element");
+ // first element
+ assertEquals("id-V0", rows.get(0)[0]);
+ assertEquals("V0", rows.get(0)[2]);
+ assertEquals("P1", rows.get(0)[5]);
+ assertEquals("R1|P1|V0", rows.get(0)[6]);
+ // last element
+ assertEquals("id-V4", rows.get(4)[0]);
+ assertEquals("V4", rows.get(4)[2]);
+ assertEquals("R1|P1|V4", rows.get(4)[6]);
+ // every row fully populated
+ for (Object[] r : rows) {
+ assertEquals(11, r.length);
+ assertEquals("mz", r[1]);
+ assertEquals("ADMIN", r[3]);
+ }
+ }
+
+ // -------------------- NEGATIVE: why the topics must stay separate (the original blocker) --------------------
+
+ @Test
+ void singleObject_onBulkMapping_breaks() {
+ // This is exactly what happened when both shapes shared save-boundary-relationship with a .* mapping.
+ String json = "{\"RequestInfo\":{\"ver\":\"1.0\"},\"BoundaryRelationship\":"
+ + relJson("id-C1", "C1", null, "C1") + "}";
+ assertThrows(Exception.class, () -> new PersistRepository()
+ .getRows(jsonMaps("$.BoundaryRelationship.*."), parse(json), "$.BoundaryRelationship.*"),
+ "a single OBJECT under a .* (array) base path must fail — this is why bulk needs its own topic");
+ }
+
+ @Test
+ void array_onSingleMapping_breaks() {
+ String json = "{\"RequestInfo\":{\"ver\":\"1.0\"},\"BoundaryRelationship\":["
+ + relJson("id-V0", "V0", "P1", "R1|P1|V0") + "]}";
+ assertThrows(Exception.class, () -> new PersistRepository()
+ .getRows(jsonMaps("$.BoundaryRelationship."), parse(json), "$.BoundaryRelationship"),
+ "an ARRAY under a single (non-wildcard) base path must fail — confirms single-create keeps its object topic");
+ }
+}
From 7c8d3ea104e93e601d950d0e20320d1fc09a42f1 Mon Sep 17 00:00:00 2001
From: bhaswatmandal-egov
Date: Mon, 13 Jul 2026 16:18:37 +0530
Subject: [PATCH 14/24] Remove boundary-relationship persister shape test
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)
---
.../BoundaryRelationshipMappingShapeTest.java | 142 ------------------
1 file changed, 142 deletions(-)
delete mode 100644 core-services/egov-persister/src/test/java/org/egov/infra/persist/repository/BoundaryRelationshipMappingShapeTest.java
diff --git a/core-services/egov-persister/src/test/java/org/egov/infra/persist/repository/BoundaryRelationshipMappingShapeTest.java b/core-services/egov-persister/src/test/java/org/egov/infra/persist/repository/BoundaryRelationshipMappingShapeTest.java
deleted file mode 100644
index aa4cb994011..00000000000
--- a/core-services/egov-persister/src/test/java/org/egov/infra/persist/repository/BoundaryRelationshipMappingShapeTest.java
+++ /dev/null
@@ -1,142 +0,0 @@
-package org.egov.infra.persist.repository;
-
-import com.jayway.jsonpath.Configuration;
-import org.egov.infra.persist.web.contract.JsonMap;
-import org.junit.jupiter.api.Test;
-
-import java.util.ArrayList;
-import java.util.List;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-
-/**
- * Proves the boundary-relationship "dedicated bulk topic" fix at the persister layer, using the REAL
- * {@link PersistRepository#getRows} extraction (documents parsed exactly as
- * {@code PersistService} does: {@code Configuration.defaultConfiguration().jsonProvider().parse(json)}).
- *
- *
Two message shapes exist on two different topics:
- *
- *
single-create publishes {@code {"BoundaryRelationship": {..one object..}}} to
- * {@code save-boundary-relationship} — mapped with base path {@code $.BoundaryRelationship} (no wildcard).
- *
bulk-create publishes {@code {"BoundaryRelationship": [..array..]}} to
- * {@code boundary-relationship-bulk-create-job} — mapped with base path {@code $.BoundaryRelationship.*}.
- *
- * The positive tests prove each shape+mapping pairing extracts the correct number of rows with correct
- * field values. The negative tests prove WHY the topics must stay separate: crossing a shape with the
- * wrong base path fails (the original single-topic blocker).
- */
-class BoundaryRelationshipMappingShapeTest {
-
- // Column order of the INSERT in boundary-persister.yml
- private static final String[] FIELDS = {
- "id", "tenantId", "code", "hierarchyType", "boundaryType", "parent",
- "ancestralMaterializedPath",
- "auditDetails.createdTime", "auditDetails.createdBy",
- "auditDetails.lastModifiedTime", "auditDetails.lastModifiedBy"
- };
-
- private List jsonMaps(String prefix) {
- List maps = new ArrayList<>();
- for (String f : FIELDS) {
- JsonMap m = new JsonMap();
- m.setJsonPath(prefix + f); // e.g. "$.BoundaryRelationship." + "code" or "$.BoundaryRelationship.*." + "code"
- maps.add(m);
- }
- return maps;
- }
-
- private String relJson(String id, String code, String parent, String amp) {
- return "{"
- + "\"id\":\"" + id + "\","
- + "\"tenantId\":\"mz\","
- + "\"code\":\"" + code + "\","
- + "\"hierarchyType\":\"ADMIN\","
- + "\"boundaryType\":\"Village\","
- + "\"parent\":" + (parent == null ? "null" : "\"" + parent + "\"") + ","
- + "\"ancestralMaterializedPath\":\"" + amp + "\","
- + "\"auditDetails\":{\"createdTime\":100,\"createdBy\":\"u1\",\"lastModifiedTime\":200,\"lastModifiedBy\":\"u2\"}"
- + "}";
- }
-
- private Object parse(String json) {
- // identical to PersistService.persist(...)
- return Configuration.defaultConfiguration().jsonProvider().parse(json);
- }
-
- // -------------------- POSITIVE: correct shape + correct mapping --------------------
-
- @Test
- void singleObject_onSingleMapping_extractsExactlyOneRowWithCorrectValues() {
- String json = "{\"RequestInfo\":{\"ver\":\"1.0\"},\"BoundaryRelationship\":"
- + relJson("id-C1", "C1", null, "C1") + "}";
-
- List rows = new PersistRepository()
- .getRows(jsonMaps("$.BoundaryRelationship."), parse(json), "$.BoundaryRelationship");
-
- assertEquals(1, rows.size(), "single-object message must yield exactly ONE row");
- Object[] r = rows.get(0);
- assertEquals(11, r.length, "row must have all 11 columns");
- assertEquals("id-C1", r[0]);
- assertEquals("mz", r[1]);
- assertEquals("C1", r[2]);
- assertEquals("ADMIN", r[3]);
- assertEquals("Village", r[4]);
- assertEquals(null, r[5], "root parent is null");
- assertEquals("C1", r[6]);
- assertEquals("u1", r[8]);
- assertEquals("u2", r[10]);
- }
-
- @Test
- void array_onBulkMapping_extractsOneRowPerElementWithCorrectValues() {
- StringBuilder arr = new StringBuilder();
- int n = 5;
- for (int i = 0; i < n; i++) {
- if (i > 0) arr.append(",");
- arr.append(relJson("id-V" + i, "V" + i, "P1", "R1|P1|V" + i));
- }
- String json = "{\"RequestInfo\":{\"ver\":\"1.0\"},\"BoundaryRelationship\":[" + arr + "]}";
-
- List rows = new PersistRepository()
- .getRows(jsonMaps("$.BoundaryRelationship.*."), parse(json), "$.BoundaryRelationship.*");
-
- assertEquals(n, rows.size(), "array message must yield ONE row per element");
- // first element
- assertEquals("id-V0", rows.get(0)[0]);
- assertEquals("V0", rows.get(0)[2]);
- assertEquals("P1", rows.get(0)[5]);
- assertEquals("R1|P1|V0", rows.get(0)[6]);
- // last element
- assertEquals("id-V4", rows.get(4)[0]);
- assertEquals("V4", rows.get(4)[2]);
- assertEquals("R1|P1|V4", rows.get(4)[6]);
- // every row fully populated
- for (Object[] r : rows) {
- assertEquals(11, r.length);
- assertEquals("mz", r[1]);
- assertEquals("ADMIN", r[3]);
- }
- }
-
- // -------------------- NEGATIVE: why the topics must stay separate (the original blocker) --------------------
-
- @Test
- void singleObject_onBulkMapping_breaks() {
- // This is exactly what happened when both shapes shared save-boundary-relationship with a .* mapping.
- String json = "{\"RequestInfo\":{\"ver\":\"1.0\"},\"BoundaryRelationship\":"
- + relJson("id-C1", "C1", null, "C1") + "}";
- assertThrows(Exception.class, () -> new PersistRepository()
- .getRows(jsonMaps("$.BoundaryRelationship.*."), parse(json), "$.BoundaryRelationship.*"),
- "a single OBJECT under a .* (array) base path must fail — this is why bulk needs its own topic");
- }
-
- @Test
- void array_onSingleMapping_breaks() {
- String json = "{\"RequestInfo\":{\"ver\":\"1.0\"},\"BoundaryRelationship\":["
- + relJson("id-V0", "V0", "P1", "R1|P1|V0") + "]}";
- assertThrows(Exception.class, () -> new PersistRepository()
- .getRows(jsonMaps("$.BoundaryRelationship."), parse(json), "$.BoundaryRelationship"),
- "an ARRAY under a single (non-wildcard) base path must fail — confirms single-create keeps its object topic");
- }
-}
From 80308c38f3bd66c407fc00f8e676f61f909c57e6 Mon Sep 17 00:00:00 2001
From: hruthvikl-egov
Date: Fri, 17 Jul 2026 12:17:47 +0530
Subject: [PATCH 15/24] tracer changes 2.9.3 version CHANGELOG added
---
core-services/libraries/tracer/CHANGELOG.md | 8 ++++++++
1 file changed, 8 insertions(+)
diff --git a/core-services/libraries/tracer/CHANGELOG.md b/core-services/libraries/tracer/CHANGELOG.md
index 94b19eb2aba..956e8bf578f 100644
--- a/core-services/libraries/tracer/CHANGELOG.md
+++ b/core-services/libraries/tracer/CHANGELOG.md
@@ -1,6 +1,14 @@
# Changelog
All notable changes to this library will be documented in this file.
+## 2.9.3 - 2026-07-17
+- Propagated correlationId and tenantId across Kafka — stamp both onto message headers on produce and rebuild the MDC from them on consume, so consumer/async logs carry the same IDs
+- Added MdcRecordInterceptor (auto-attached to the default Kafka listener container factory), KafkaTemplateLoggingInterceptors, and TracerKafkaMdcUtil helper
+- Added kill-switch `tracer.kafka.mdc.enabled` (default true)
+
+## 2.9.2 - 2026-03-10
+- Addition of Data Access Exception Handling in ExceptionAdvice.
+
## 2.9.0 - 2024-02-29
- Upgraded spring boot version from 2.2.6.RELEASE to 3.2.2
- Upgraded java version from 1.8 to 17
From b242c9b6c1af46d47d0d4ed9a185686b1acac387 Mon Sep 17 00:00:00 2001
From: bhaswatmandal-egov
Date: Mon, 20 Jul 2026 17:19:44 +0530
Subject: [PATCH 16/24] Version bumps for 2.1 release
---
core-services/boundary-service/CHANGELOG.md | 11 +++++
core-services/boundary-service/README.md | 25 ++++++-----
core-services/boundary-service/pom.xml | 2 +-
core-services/egov-persister/CHANGELOG.md | 10 +++++
core-services/egov-persister/README.md | 47 ++++++++++++++++++++-
core-services/egov-persister/pom.xml | 2 +-
core-services/libraries/tracer/readme.md | 42 +++++++++++++++---
7 files changed, 119 insertions(+), 20 deletions(-)
create mode 100644 core-services/boundary-service/CHANGELOG.md
diff --git a/core-services/boundary-service/CHANGELOG.md b/core-services/boundary-service/CHANGELOG.md
new file mode 100644
index 00000000000..6cd144c381f
--- /dev/null
+++ b/core-services/boundary-service/CHANGELOG.md
@@ -0,0 +1,11 @@
+# Changelog
+
+All notable changes to this module will be documented in this file.
+
+## 1.0.2 - 2026-07-20
+- Bulk relationship create API (`POST /boundary-relationships/bulk/_create`): relationships are validated + enriched synchronously and published together as **one array message keyed by the shared parent code** to a dedicated bulk topic, instead of one message per record — new models `BulkBoundaryRelationshipRequest` / `BulkBoundaryRelationshipRequestDTO` / `BulkBoundaryRelationshipResponse` / `FailedBoundaryRelationship`
+- Persister config (`boundary-persister.yml`) now carries **two separate relationship queryMaps** — a single-object map for `save-boundary-relationship` (single create) and an array map for the dedicated bulk topic
+- Boundary relationship creation performance enhancement: query-builder and repository bulk paths reworked; new search-index migration `V20260616120000__boundary_relationship_search_indexes.sql`
+- `correlationId` + `tenantId` now propagated across Kafka on publish (via tracer `2.9.3-SNAPSHOT`)
+- New error codes and configurable properties added (`ApplicationProperties`, `ErrorCodes`, `application.properties`)
+- Code-review (CodeRabbit) fixes applied
diff --git a/core-services/boundary-service/README.md b/core-services/boundary-service/README.md
index 614a4952835..07aa5dc64da 100644
--- a/core-services/boundary-service/README.md
+++ b/core-services/boundary-service/README.md
@@ -28,12 +28,12 @@ All endpoints are `POST`. Base URL: `http://:8081/boundary-service`.
### Single vs. bulk relationship create
-Both paths validate + enrich a relationship (assign `id`, audit details, and the ancestral materialized path) and then **publish it to the `save-boundary-relationship` topic**, which `egov-persister` writes with an idempotent `INSERT … ON CONFLICT (tenantId, code, hierarchyType) DO NOTHING`. Neither path writes to the database directly.
+Both paths validate + enrich a relationship (assign `id`, audit details, and the ancestral materialized path) before anything is placed on Kafka, and `egov-persister` writes every record with an idempotent `INSERT … ON CONFLICT (tenantId, code, hierarchyType) DO NOTHING`. Neither path writes to the database directly. The two paths use **different topics** because the message shapes differ:
-- The **single** `_create` validates one record and publishes it, returning `202 Accepted` (acceptance, not a committed write).
-- The **bulk** `_create` validates + enriches every record in the request **synchronously**, publishes the valid ones (one message per record), and returns `200 OK` with a **per-record outcome** so the caller learns immediately which records were accepted and which failed validation, and with what reason.
+- The **single** `_create` validates one record and publishes it as a single object to the **`save-boundary-relationship`** topic, returning `202 Accepted` (acceptance, not a committed write).
+- The **bulk** `_create` validates + enriches every record in the request **synchronously**, then publishes the *whole validated batch as ONE array message* (the `BoundaryRelationship` field is a JSON array) to the dedicated **`boundary-relationship-bulk-create-job`** topic — keyed by the batch's shared parent code (keyless when parents differ or the batch is root) — and returns `200 OK` with a **per-record outcome** so the caller learns immediately which records were accepted and which failed validation, and with what reason.
-See [`docs/Bulk-Boundary-Relationship-Creation-Design.docx`](docs/Bulk-Boundary-Relationship-Creation-Design.docx) for the design rationale and [`docs/Bulk-Boundary-Relationship-Flow.docx`](docs/Bulk-Boundary-Relationship-Flow.docx) for the end-to-end flow.
+The design rationale and end-to-end flow for bulk creation are documented in the sections below.
---
@@ -41,13 +41,13 @@ See [`docs/Bulk-Boundary-Relationship-Creation-Design.docx`](docs/Bulk-Boundary-
`POST /boundary-relationships/bulk/_create`
-Creates many relationships in one request with a **per-record outcome**. Validation and enrichment happen synchronously on the request thread **before anything is placed on Kafka**; each valid record is then published to `save-boundary-relationship` for `egov-persister` to write. Individual record failures do **not** fail the whole request.
+Creates many relationships in one request with a **per-record outcome**. Validation and enrichment happen synchronously on the request thread **before anything is placed on Kafka**; the valid records are then published together as one array message to the dedicated `boundary-relationship-bulk-create-job` topic for `egov-persister` to write. Individual record failures do **not** fail the whole request.
### Semantics
- **Request guards.** `RequestInfo.userInfo` must be present, and the request must carry `1 … boundary.bulk.max.size` (default `100`) relationships. These are enforced in-service (bean validation is not active in this deployment) and return a structured `400` (`BULK_REQUEST_INFO_MISSING` / `BULK_REQUEST_EMPTY` / `BULK_REQUEST_SIZE_EXCEEDED`).
- **Per-record validation.** Each record runs the same business rules as the single create (boundary entity exists, no duplicate, parent exists, correct hierarchy level; `code`/`tenantId`/`hierarchyType` must not contain the reserved `|` path delimiter). A record that fails validation is reported in `failedBoundaryRelationships`; the rest continue.
-- **Persistence via egov-persister.** Validated + enriched records are published, **one message per record**, to `save-boundary-relationship`. The publish is blocking (it returns once the broker has accepted each record). `egov-persister` writes each with `INSERT … ON CONFLICT (tenantId, code, hierarchyType) DO NOTHING`, so redelivery / re-submission is a safe no-op and one un-insertable record never fails the others. A record accepted here is reported in `successfulBoundaryRelationships` ("accepted for persistence", not "already committed").
+- **Persistence via egov-persister.** The validated + enriched records are published as **ONE array message** (`BoundaryRelationship` is a JSON array, keyed by the batch's shared parent code) to the dedicated `boundary-relationship-bulk-create-job` topic. The publish is blocking (it returns once the broker has accepted the batch message). `egov-persister` maps that array (`$.BoundaryRelationship.*`) to N rows and writes them in a single `jdbcTemplate.batchUpdate` via `INSERT … ON CONFLICT (tenantId, code, hierarchyType) DO NOTHING`, so redelivery / re-submission is a safe no-op and an already-present (conflicting) row is skipped without failing the others. A record whose batch message is accepted here is reported in `successfulBoundaryRelationships` ("accepted for persistence", not "already committed").
- **Transient failures are retryable, not fatal.** A parent/entity not yet persisted, a transient DB error, or a publish failure (broker unreachable within `max.block.ms`) is reported with a retryable code (`PARENT_NOT_FOUND`, `BOUNDARY_ENTITY_DOES_NOT_EXIST`, `BULK_RELATIONSHIP_PERSIST_TRANSIENT`) rather than aborting; the caller retries only those records.
- **Intra-request de-duplication.** Two records with the same `(tenantId, hierarchyType, code)` in one request are rejected (`DUPLICATE_RECORD_IN_REQUEST`).
@@ -63,7 +63,7 @@ The endpoint expects each batch to be a set of **siblings whose parent is alread
- **Horizontal scale** is by running more boundary-service replicas behind the load balancer (a stateless HTTP path) and by the caller submitting chunks concurrently — there is no Kafka consumer group to size and no per-partition head-of-line blocking.
- **Idempotency** comes from `INSERT … ON CONFLICT (tenantid, code, hierarchytype) DO NOTHING`, so a parent that is briefly late simply causes its children to be retried by the caller until it lands; ordering across levels is *not* required.
-- **DB-write reliability** (per-record isolation of an un-insertable row, transient-DB retry, dead-lettering) is owned by `egov-persister`. Because it is one message per record, a bad record is isolated on its own regardless of whether the persister runs its normal listener or the optional batch listener.
+- **DB-write reliability** (transient-DB retry, dead-lettering) is owned by `egov-persister`. The batch is one array message written in a single `batchUpdate`; a row that already exists is isolated via `ON CONFLICT … DO NOTHING` — it is skipped without aborting the rest of the batch, and no per-record Kafka message or optional batch listener is involved.
### Request
@@ -142,7 +142,7 @@ mvn clean install
# Run (after configuring datasource & kafka in application.properties)
mvn spring-boot:run
# or
-java -jar target/boundary-service-1.0.1.jar
+java -jar target/boundary-service-1.0.2.jar
```
### Key configuration (`src/main/resources/application.properties`)
@@ -154,17 +154,20 @@ java -jar target/boundary-service-1.0.1.jar
| `spring.flyway.enabled` | `false` | Enable to run DB migrations on startup |
| `spring.kafka.bootstrap-servers` | `localhost:9092` | Kafka brokers used by the producer. In DIGIT deployments the common Helm chart injects `SPRING_KAFKA_BOOTSTRAP_SERVERS`. |
| `spring.kafka.producer.properties.max.block.ms` | `15000` | Bounds how long a synchronous publish blocks if the broker is unreachable, so an outage fails fast (as a transient error the caller retries) instead of tying up request threads. |
-| `kafka.topics.create.boundary.relationship` | `save-boundary-relationship` | Topic for **both** single and bulk relationship create, consumed by `egov-persister`. |
+| `kafka.topics.create.boundary.relationship` | `save-boundary-relationship` | Topic for **single** relationship create only (single-object payload), consumed by `egov-persister`. |
+| `kafka.topics.bulk.create.boundary.relationship.job` | `boundary-relationship-bulk-create-job` | **Dedicated** topic for `/bulk/_create`; carries one **array** message per batch (`$.BoundaryRelationship.*`), kept separate from the single-create topic because the array and single-object shapes cannot share a topic. |
| `boundary.bulk.max.size` | `100` | Max records accepted by `/bulk/_create` (enforced in-service; keep ≥ the caller's chunk size). |
| `boundary.default.limit` / `boundary.max.default.limit` | `50` / `300` | Search paging defaults |
-> Bulk creation writes through `egov-persister`, not directly to PostgreSQL, so the request path is not gated by the DB connection pool for writes. To make the persister aggregate high-volume creates into batched multi-row inserts, add `save-boundary-relationship` to the persister's `persister.batch.topics` (with `persister.bulk.enabled=true`); this is an **optional throughput optimization** — the topic is otherwise consumed one record at a time by the persister's normal listener, so bulk creation works on any persister deployment with no extra configuration.
+> Bulk creation writes through `egov-persister`, not directly to PostgreSQL, so the request path is not gated by the DB connection pool for writes. Batching is **within the one array message** — the persister maps `$.BoundaryRelationship.*` to N rows and writes them in a single `jdbcTemplate.batchUpdate` on its **normal** listener. It therefore needs **no** `persister.batch.topics` / `persister.bulk.enabled` configuration; the dedicated `boundary-relationship-bulk-create-job` topic plus the array queryMap does the batched multi-row insert on any persister deployment with no extra tuning.
---
## Database
-Relationships are stored in `boundary_relationship` (`tenantId, code, hierarchyType` primary key; `ancestralMaterializedPath` holds the `|`-delimited ancestor chain used for subtree search). All writes go through the `egov-persister` mapping in `src/main/resources/boundary-persister.yml`, which uses `INSERT … ON CONFLICT (tenantid, code, hierarchytype) DO NOTHING` so at-least-once redelivery and caller re-submission are idempotent.
+Relationships are stored in `boundary_relationship` (`tenantId, code, hierarchyType` primary key; `ancestralMaterializedPath` holds the `|`-delimited ancestor chain used for subtree search). All writes go through the `egov-persister` mapping in `src/main/resources/boundary-persister.yml`.
+
+That mapping has **two separate relationship queryMaps**: one for the single-create topic `save-boundary-relationship` (single-object base path `$.BoundaryRelationship`) and a dedicated one for the bulk topic `boundary-relationship-bulk-create-job` (array base path `$.BoundaryRelationship.*`, written as a single `batchUpdate`). Both, together with the boundary-entity and boundary-hierarchy inserts, are now idempotent — every insert uses `ON CONFLICT … DO NOTHING` (`(code, tenantId)` for `boundary`, `(tenantId, hierarchyType)` for `boundary_hierarchy`, `(tenantId, code, hierarchyType)` for `boundary_relationship`), 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.
### Search indexes
diff --git a/core-services/boundary-service/pom.xml b/core-services/boundary-service/pom.xml
index 33276f04667..c0818c0220b 100644
--- a/core-services/boundary-service/pom.xml
+++ b/core-services/boundary-service/pom.xml
@@ -4,7 +4,7 @@
boundary-servicejarboundary-service
- 1.0.1
+ 1.0.217${java.version}
diff --git a/core-services/egov-persister/CHANGELOG.md b/core-services/egov-persister/CHANGELOG.md
index 9cc3eaf0a5f..5042c180203 100644
--- a/core-services/egov-persister/CHANGELOG.md
+++ b/core-services/egov-persister/CHANGELOG.md
@@ -2,6 +2,16 @@
# Changelog
All notable changes to this module will be documented in this file.
+## 2.9.4 - 2026-07-20
+- At-least-once delivery: manual offset commit (`spring.kafka.consumer.enable-auto-commit=false`), per-record (RECORD) ack on the single container and per-batch (BATCH) ack on the batch container — offsets commit only after durable handling
+- SQLSTATE-based failure classification: benign duplicate (`unique_violation` 23505) treated as idempotent success, transient failures (connection/deadlock/serialization) retried in place, permanent/bad-data records routed to the dead-letter topic
+- Dead-letter topic (`tracer.errorsTopic`, default `egov-persister-deadletter`) with a bounded reprocessor and a terminal parking topic (`egov-persister-deadletter-processed`); durable DLQ/parking publishes (`acks=all`, idempotent producer)
+- DB-health pause/resume monitor: the single container is paused while the datasource is unreachable and resumed on recovery
+- Per-record poison isolation: a failing bulk (bare JSON array) message is split so only the offending record is dead-lettered while its siblings commit
+- Batch persist optimization: rows aggregated per QueryMap across all messages into a single order-preserving `batchUpdate`
+- Idempotent service configs: added `ON CONFLICT (uuid) DO NOTHING` to inserts to support safe redelivery / DLQ replay
+- New config keys: `persister.batch.topics`, `persister.dead-letter.*`, `persister.db-health.check-interval-ms`, `persister.custom.executor.*`, `persister.batch.parallel-topic-processing.thread-pool-size`, and the live-read `persister.kafka.*` consumer tuning knobs
+
## 2.9.3 - 2026-03-16
- Upgraded Spring Boot version from 3.2.2 to 3.4.5 to fix HIGH/CRITICAL CVEs
- Upgraded tracer, services-common, mdms-client, enc-client library versions
diff --git a/core-services/egov-persister/README.md b/core-services/egov-persister/README.md
index 381a3316323..5efc73b65f9 100644
--- a/core-services/egov-persister/README.md
+++ b/core-services/egov-persister/README.md
@@ -75,8 +75,39 @@ The bulk persister have the following two settings:
|-------------------------|---------------|-------------------------------------------------|
| `persister.bulk.enabled`| false | Switch to turn on or off the bulk kafka consumer|
| `persister.batch.size` | 100 | The batch size for bulk update |
+| `persister.batch.topics`| (empty) | Comma-separated topics to force through the batch consumer, in addition to any topic whose name contains '-batch' |
-Any kafka topic containing data which has to be bulk persisted should have '-batch' appended at the end of topic name example: save-pt-assessment-batch
+Any kafka topic containing data which has to be bulk persisted should have '-batch' appended at the end of topic name example: save-pt-assessment-batch. Alternatively, topics can be routed through the batch consumer without renaming them by listing them (comma-separated) in `persister.batch.topics`. Batch topics (both '-batch' named and those in `persister.batch.topics`) are only excluded from the single (record) consumer when `persister.bulk.enabled=true`.
+
+### Reliability: at-least-once, DB-health pause, dead-letter & parking
+
+The persister is an at-least-once, poison-tolerant writer. Key operational behaviour:
+
+1. **At-least-once delivery.** Auto-commit is turned off (`spring.kafka.consumer.enable-auto-commit=false`). Both listeners commit manually — the single (record) consumer per record (`RECORD` ack), the batch consumer per batch (`BATCH` ack). An offset is committed only after the record has been persisted or durably dead-lettered, so a crash mid-processing results in redelivery, never silent loss.
+2. **Failure classification by PostgreSQL SQLSTATE.** Database failures are classified into three buckets:
+ - **BENIGN** — `unique_violation` (23505) is treated as an idempotent success (the row is already there from a prior at-least-once delivery).
+ - **TRANSIENT** — connection/deadlock/serialization failures (e.g. `08*`, `57*`, `40001`, `40P01`, `53300`, `55P03`, connection-acquisition) are retried in place with back-off and are never dead-lettered.
+ - **PERMANENT** — bad-data failures are routed to the dead-letter topic.
+3. **DB-health pause/resume.** While the datasource is unreachable the single container is paused, and resumed on recovery, so transient outages retry in place instead of hammering a dead DB. The poll interval is `persister.db-health.check-interval-ms` (default `5000` ms).
+4. **Per-record poison isolation.** When a bulk (bare JSON array) message fails, it is split into single-record inserts so only the offending record is dead-lettered while its good siblings still commit.
+5. **Idempotent writes.** Service persister configs use `ON CONFLICT (uuid) DO NOTHING` on inserts so that redelivery / dead-letter replay is safe.
+
+**Reliability configuration**
+
+| variable name | Default value | Description |
+|------------------------------------------------------------|--------------------------------------|--------------------------------------------------------------------------------------|
+| `spring.kafka.consumer.enable-auto-commit` | false | Manual offset commit; offsets commit only after durable handling (at-least-once) |
+| `persister.db-health.check-interval-ms` | 5000 | Interval for the DB-health pause/resume monitor |
+| `persister.dead-letter.reprocess.enabled` | true | Re-consume the dead-letter topic on the single listener and retry failed records |
+| `persister.dead-letter.reprocess.error-topic` | egov-persister-deadletter-processed | Terminal parking topic for records that exhaust retries |
+| `persister.dead-letter.max-retries` | 5 | Bounded DLQ retries before a record is parked |
+| `persister.custom.executor.enabled` | false | Optional listener task executor for the single container (off by default) |
+| `persister.custom.executor.max-pool-size` | 10 | Max pool size for the optional listener task executor |
+| `persister.batch.parallel-topic-processing.thread-pool-size` | 1 | Thread pool for parallel per-topic processing inside a batch |
+| `spring.kafka.producer.acks` | all | Durable, no-loss dead-letter / parking publishes |
+| `spring.kafka.producer.properties.enable.idempotence` | true | Idempotent producer for dead-letter / parking publishes |
+
+The following consumer tuning keys are read live and are not written in `application.properties`: `persister.kafka.partition.assignment.strategy` (default `CooperativeStickyAssignor,RangeAssignor`), `persister.kafka.group.instance.id` (Kafka static membership), and `persister.kafka.session.timeout.ms`.
### Persister Config Versioning
@@ -89,7 +120,19 @@ Any kafka topic containing data which has to be bulk persisted should have '-bat
### Kafka Consumers
- From the Kafka topic which are mentioned in the persister config, persister service get message/data and push the data into the particular tables of the database.
+- When `persister.dead-letter.reprocess.enabled=true`, the single listener additionally re-consumes the dead-letter topic (`tracer.errorsTopic`, default `egov-persister-deadletter`) to retry previously failed records.
### Kafka Producers
-- NA
+- The persister produces to a **dead-letter topic** (`tracer.errorsTopic`, default `egov-persister-deadletter`) when a record cannot be persisted (permanent/bad-data failure, or a split poison record from a bulk message).
+- When `persister.dead-letter.reprocess.enabled=true`, the single listener re-consumes the dead-letter topic and retries each record up to `persister.dead-letter.max-retries` (default `5`). Records that exhaust their retries are produced to a **terminal parking topic** (`persister.dead-letter.reprocess.error-topic`, default `egov-persister-deadletter-processed`).
+- Dead-letter and parking publishes are durable (`spring.kafka.producer.acks=all`, `spring.kafka.producer.properties.enable.idempotence=true`).
+
+### Dead-letter & parking topics
+
+| topic | property | Purpose |
+|----------------------------------------|-------------------------------------------------|----------------------------------------------------------------|
+| `egov-persister-deadletter` | `tracer.errorsTopic` | Dead-letter topic for failed records; re-consumed on retry |
+| `egov-persister-deadletter-processed` | `persister.dead-letter.reprocess.error-topic` | Terminal parking topic for records that exhaust their retries |
+
+> Parking-topic growth is the terminal-poison signal — monitor it (and dead-letter lag) in ops.
diff --git a/core-services/egov-persister/pom.xml b/core-services/egov-persister/pom.xml
index cd599e83aae..002f2fc2dcd 100644
--- a/core-services/egov-persister/pom.xml
+++ b/core-services/egov-persister/pom.xml
@@ -9,7 +9,7 @@
org.egovegov-persister
- 2.9.3-SNAPSHOT
+ 2.9.4-SNAPSHOTegov-persisteregov persister framework
diff --git a/core-services/libraries/tracer/readme.md b/core-services/libraries/tracer/readme.md
index d03d20bf86e..0f763b4fc7b 100644
--- a/core-services/libraries/tracer/readme.md
+++ b/core-services/libraries/tracer/readme.md
@@ -41,20 +41,48 @@ Map map = new HashMap<>();
The logging of the http request/response body and Kakfa message body can be toggled on/off using
"tracer.detailed.tracing.enabled" application property.
+The per-record Kafka MDC rebuild + cleanup (see "Setting the correlation id in the MDC") can be
+toggled on/off using the "tracer.kafka.mdc.enabled" application property. It defaults to true; set it
+to false to disable registration of the record interceptor.
+
###### Correlation id retrieval and forwarding -
The library takes care of retrieving the correlation id from -
- Incoming http request body or header
-- Kafka message payload
+- Kafka message headers (with a fallback to the message payload)
For an outgoing http request the correlation id is sent as a custom request header "x-correlation-id".
+As of version 2.9.3 both the correlation id and the tenant id are also propagated across Kafka via
+message headers -
+
+- On produce, the correlation id and tenant id are stamped from the MDC onto the outgoing message
+ headers "x-correlation-id" (CORRELATION_ID_HEADER) and "tenantId" (TENANT_ID_HEADER), only when
+ present in the MDC and not already set on the record.
+- On consume, the MDC is rebuilt from these headers. The correlation id falls back to the message
+ payload (RequestInfo.correlationId / requestInfo.correlationId) when the header is absent.
+
+This means consumer and downstream async logs carry the same correlation id and tenant id as the
+originating request.
+
###### Setting the correlation id in the MDC -
Given the library takes care of placing the correlation id into the MDC, any custom logging done in the
application would seamlessly include the correlation id in the log message.
+For Kafka consumers, as of version 2.9.3 the MDC is managed per record by a RecordInterceptor
+(MdcRecordInterceptor). Before each record is processed the correlation id and tenant id are rebuilt
+into the MDC from the record headers, and after the record completes (on both success and failure)
+both MDC keys are cleared. This per-record cleanup prevents the ids from leaking across records when
+listener threads are reused.
+
+Note - The mdcRecordInterceptor bean is registered by TracerConfiguration, but Spring Boot only
+auto-attaches it to its autoconfigured record-listener container factory. Services that define a
+custom ConcurrentKafkaListenerContainerFactory or use batch listeners must wire it manually by calling
+setRecordInterceptor(mdcRecordInterceptor()) on their factory; otherwise their consumer logs will not
+carry the propagated correlation id and tenant id.
+
Note - See the "logging.pattern" mentioned in the "Tracer integration" section.
#### Steps to integrate Tracer to your Spring application -
@@ -114,10 +142,14 @@ Note - See the "logging.pattern" mentioned in the "Tracer integration" section.
logging.
- The LogAwareKafkaTemplate is a wrapper Spring bean class for Spring Kafka's KafkaTemplate that performs the logging
of messages sent to Kafka.
-- For a Kafka consumer implemented using Spring Kafka's KafkaListener annotation an AspectJ's aspect is used to log and
- retrieve the correlation id from the received payload.
-- The correlation id retrieved via the filter or aspect is then stored in a thread local variable to
- forward as necessary.
+- For Kafka, a producer/consumer interceptor (KafkaTemplateLoggingInterceptors) performs the logging of messages
+ sent and received. As of version 2.9.3, on produce it stamps the correlation id and tenant id from the MDC onto the
+ message headers ("x-correlation-id" and "tenantId"), and on consume it rebuilds the MDC from those headers (with a
+ fallback to the payload for the correlation id) via the TracerKafkaMdcUtil helper.
+- For Kafka consumers, a RecordInterceptor (MdcRecordInterceptor) rebuilds the MDC (correlation id + tenant id) before
+ each record and clears both keys afterwards, so ids do not leak across records on reused listener threads.
+- The correlation id retrieved via the filter (for http) or the Kafka headers (for consumers) is placed into the MDC
+ to forward as necessary.
#### Change log -
From fe334d2c304d494cb427f2dbdd43aa6f4102ca21 Mon Sep 17 00:00:00 2001
From: bhaswatmandal-egov
Date: Wed, 22 Jul 2026 18:57:35 +0530
Subject: [PATCH 17/24] Record Splitting updated for non bare array
---
.../persist/consumer/RecordSplitter.java | 96 ++++++++++++++++---
.../persist/consumer/RecordSplitterTest.java | 44 ++++++++-
2 files changed, 121 insertions(+), 19 deletions(-)
diff --git a/core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/RecordSplitter.java b/core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/RecordSplitter.java
index 461998b2e5e..afdece6d274 100644
--- a/core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/RecordSplitter.java
+++ b/core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/RecordSplitter.java
@@ -3,20 +3,36 @@
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
/**
* Splits a multi-record bulk message into standalone single-record messages so a permanent failure
* can be isolated to the offending record(s) instead of failing every record that happened to share
* the same Kafka message (R1 at record granularity, not just message granularity).
*
- *
Bulk producers publish a whole validated list as ONE message whose payload is a bare JSON
- * array, and the persister maps such topics with array base paths ({@code $.*}), so a
- * single-element array is processed identically to the full array. Only bare arrays with more than
- * one element are split; object-shaped payloads (whose base paths may not be per-element) keep
- * message-level handling.
+ *
Two payload shapes are split:
+ *
+ *
bare array — {@code [ {..}, {..} ]} (base path {@code $.*}): re-emitted as one
+ * single-element array per record.
+ *
object with one record array — {@code { "Boundary": [ {..}, {..} ] }} (base path
+ * {@code $.Boundary.*}): the message is deep-copied and only that array is replaced with a
+ * single element, so sibling keys (e.g. {@code RequestInfo}) are preserved on the re-persist.
+ *
+ *
+ *
The object case only fires when EXACTLY ONE top-level field is a multi-element array AND every
+ * element of that array is a JSON object. The single-array rule makes the record list unambiguous;
+ * the objects-only rule keeps the splitter off list-valued COLUMNS of a single whole-message record
+ * (e.g. {@code {"entityIds":["e1","e2"],...}} persisted with base path {@code $}), whose truncation
+ * would corrupt the record instead of isolating it. Anything else — not JSON, a single element, zero
+ * or several top-level array fields, non-object elements, or the array nested deeper — is left
+ * unsplit and the caller keeps message-level handling. Residual blind spot: an object-element array
+ * that is itself a column of a whole-message-mapped record is structurally indistinguishable from a
+ * record list; no current persister mapping has that shape (a mapping-driven splitter is required if
+ * one ever does).
*/
final class RecordSplitter {
@@ -27,9 +43,9 @@ private RecordSplitter() {
}
/**
- * @return one single-element-array JSON string per record, or null when the payload is not a
- * multi-record bare array (not JSON / an object / an array of 0..1 elements) and the
- * caller must keep message-level handling.
+ * @return one single-record JSON string per element, or null when the payload is not a
+ * multi-record array (bare or under a single object key) and the caller must keep
+ * message-level handling.
*/
static List split(String json) {
if (json == null) {
@@ -37,18 +53,68 @@ static List split(String json) {
}
try {
JsonNode root = MAPPER.readTree(json);
- if (root == null || !root.isArray() || root.size() <= 1) {
+ if (root == null) {
return null;
}
- List records = new ArrayList<>(root.size());
- for (JsonNode element : root) {
- ArrayNode single = MAPPER.createArrayNode();
- single.add(element);
- records.add(MAPPER.writeValueAsString(single));
+ if (root.isArray()) {
+ return root.size() <= 1 ? null : splitArray(root, null, null);
+ }
+ if (root.isObject()) {
+ String field = singleMultiElementArrayField((ObjectNode) root);
+ if (field != null) {
+ return splitArray((ArrayNode) root.get(field), (ObjectNode) root, field);
+ }
}
- return records;
+ return null;
} catch (Exception e) {
return null;
}
}
+
+ /**
+ * Emit one message per element. When {@code wrapper} is null the record is a bare single-element
+ * array; otherwise the wrapper object is deep-copied and its {@code field} set to a single-element
+ * array, preserving every sibling key.
+ */
+ private static List splitArray(JsonNode array, ObjectNode wrapper, String field) throws Exception {
+ List records = new ArrayList<>(array.size());
+ for (JsonNode element : array) {
+ ArrayNode single = MAPPER.createArrayNode().add(element);
+ if (wrapper == null) {
+ records.add(MAPPER.writeValueAsString(single));
+ } else {
+ ObjectNode clone = wrapper.deepCopy();
+ clone.set(field, single);
+ records.add(MAPPER.writeValueAsString(clone));
+ }
+ }
+ return records;
+ }
+
+ /**
+ * The single top-level field holding a multi-element array whose elements are all JSON objects,
+ * or null when no field is unambiguously the record list (zero or several top-level array fields,
+ * an array of 0..1 elements, or any non-object element — a scalar list is a column of one record,
+ * not a record list).
+ */
+ private static String singleMultiElementArrayField(ObjectNode root) {
+ String found = null;
+ for (Map.Entry field : root.properties()) {
+ if (field.getValue().isArray()) {
+ if (found != null) {
+ return null; // more than one array field -> ambiguous, do not split
+ }
+ found = field.getKey();
+ }
+ }
+ if (found == null || root.get(found).size() <= 1) {
+ return null;
+ }
+ for (JsonNode element : root.get(found)) {
+ if (!element.isObject()) {
+ return null; // scalar/mixed elements -> a list-valued column, not a record list
+ }
+ }
+ return found;
+ }
}
diff --git a/core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/RecordSplitterTest.java b/core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/RecordSplitterTest.java
index 361570a3cea..17b5dc46ba4 100644
--- a/core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/RecordSplitterTest.java
+++ b/core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/RecordSplitterTest.java
@@ -30,10 +30,46 @@ void emptyArrayIsNotSplit() {
}
@Test
- void objectPayloadIsNotSplit() {
- // Object-shaped payloads (e.g. {"RequestInfo":...,"Entity":{...}}) may not map per-element -
- // they must keep message-level handling.
- assertNull(RecordSplitter.split("{\"RequestInfo\":{},\"records\":[{\"id\":\"a\"},{\"id\":\"b\"}]}"));
+ void objectWithSingleRecordArraySplitsPreservingSiblingKeys() {
+ // {"RequestInfo":{...},"Boundary":[a,b]} -> one message per element, RequestInfo preserved.
+ List records =
+ RecordSplitter.split("{\"RequestInfo\":{\"ver\":\"1.0\"},\"Boundary\":[{\"id\":\"a\"},{\"id\":\"b\"}]}");
+
+ assertEquals(2, records.size());
+ assertEquals("{\"RequestInfo\":{\"ver\":\"1.0\"},\"Boundary\":[{\"id\":\"a\"}]}", records.get(0));
+ assertEquals("{\"RequestInfo\":{\"ver\":\"1.0\"},\"Boundary\":[{\"id\":\"b\"}]}", records.get(1));
+ }
+
+ @Test
+ void objectWithSingleElementArrayIsNotSplit() {
+ assertNull(RecordSplitter.split("{\"Boundary\":[{\"id\":\"a\"}]}"));
+ }
+
+ @Test
+ void objectWithMultipleArrayFieldsIsNotSplit() {
+ // Ambiguous which array is the record list -> keep message-level handling.
+ assertNull(RecordSplitter.split("{\"Boundary\":[{\"id\":\"a\"},{\"id\":\"b\"}],\"extra\":[{\"x\":1}]}"));
+ }
+
+ @Test
+ void objectWithScalarArrayIsNotSplit() {
+ // A scalar list is a list-valued COLUMN of one whole-message record (e.g. privacy-audit
+ // enc-user-audit-info entityIds, base path $), not a record list - splitting would truncate
+ // the record's column and duplicate its key. Must keep message-level handling.
+ assertNull(RecordSplitter.split(
+ "{\"id\":\"u1\",\"userId\":\"9\",\"entityIds\":[\"e1\",\"e2\",\"e3\"],\"purpose\":{\"code\":\"kyc\"}}"));
+ }
+
+ @Test
+ void objectWithMixedElementArrayIsNotSplit() {
+ // Every element must be a JSON object for the array to count as a record list.
+ assertNull(RecordSplitter.split("{\"records\":[{\"id\":\"a\"},\"stray\",{\"id\":\"b\"}]}"));
+ }
+
+ @Test
+ void objectWithNoTopLevelArrayIsNotSplit() {
+ // Array nested deeper (e.g. $.bill.billDetails.*) is not reached -> message-level handling.
+ assertNull(RecordSplitter.split("{\"bill\":{\"billDetails\":[{\"id\":\"a\"},{\"id\":\"b\"}]}}"));
}
@Test
From 0a4c80ac7e3baa164aa0eb075ea16dbd4d2ff9b9 Mon Sep 17 00:00:00 2001
From: hruthvikl-egov
Date: Thu, 23 Jul 2026 15:26:05 +0530
Subject: [PATCH 18/24] tracer changes 2.9.3 version updated for
boundary-service and persister
---
core-services/egov-persister/pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/core-services/egov-persister/pom.xml b/core-services/egov-persister/pom.xml
index 002f2fc2dcd..dd142e84bad 100644
--- a/core-services/egov-persister/pom.xml
+++ b/core-services/egov-persister/pom.xml
@@ -89,7 +89,7 @@
org.egov.servicestracer
- 2.9.2-SNAPSHOT
+ 2.9.3-SNAPSHOTorg.flywaydb
From ad568fb50a46800db02143453fea3ffd1891b9f6 Mon Sep 17 00:00:00 2001
From: hruthvikl-egov
Date: Thu, 23 Jul 2026 15:27:13 +0530
Subject: [PATCH 19/24] tracer changes 2.9.3 version updated for
boundary-service
---
core-services/boundary-service/pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/core-services/boundary-service/pom.xml b/core-services/boundary-service/pom.xml
index c0818c0220b..5c6a65a318d 100644
--- a/core-services/boundary-service/pom.xml
+++ b/core-services/boundary-service/pom.xml
@@ -90,7 +90,7 @@
org.egov.servicestracer
- 2.9.2-SNAPSHOT
+ 2.9.3-SNAPSHOTorg.egov.services
From c287a5d15b930c20da2db7354f5f39d68e104848 Mon Sep 17 00:00:00 2001
From: hruthvikl-egov
Date: Mon, 27 Jul 2026 12:07:59 +0530
Subject: [PATCH 20/24] tracer version update from 2.9.2 to 2.9.3 in core
services.
---
core-services/egov-accesscontrol/CHANGELOG.md | 3 +++
core-services/egov-accesscontrol/pom.xml | 4 ++--
core-services/egov-enc-service/CHANGELOG.md | 3 +++
core-services/egov-enc-service/pom.xml | 4 ++--
core-services/egov-filestore/CHANGELOG.md | 3 +++
core-services/egov-filestore/pom.xml | 4 ++--
core-services/egov-indexer/CHANGELOG.md | 3 +++
core-services/egov-indexer/pom.xml | 4 ++--
core-services/egov-localization/CHANGELOG.md | 3 +++
core-services/egov-localization/pom.xml | 4 ++--
core-services/egov-mdms-service/CHANGELOG.md | 3 +++
core-services/egov-mdms-service/pom.xml | 4 ++--
core-services/egov-persister/CHANGELOG.md | 3 +++
core-services/egov-persister/pom.xml | 4 ++--
core-services/egov-url-shortening/CHANGELOG.md | 3 +++
core-services/egov-url-shortening/pom.xml | 4 ++--
core-services/egov-workflow-v2/CHANGELOG.md | 3 +++
core-services/egov-workflow-v2/pom.xml | 4 ++--
core-services/mdms-v2/CHANGELOG.md | 3 +++
core-services/mdms-v2/pom.xml | 4 ++--
20 files changed, 50 insertions(+), 20 deletions(-)
diff --git a/core-services/egov-accesscontrol/CHANGELOG.md b/core-services/egov-accesscontrol/CHANGELOG.md
index 0c0e46d240b..4f936754f59 100644
--- a/core-services/egov-accesscontrol/CHANGELOG.md
+++ b/core-services/egov-accesscontrol/CHANGELOG.md
@@ -3,6 +3,9 @@
# Changelog
All notable changes to this module will be documented in this file.
+## 2.9.4 - 2026-07-27
+- Bumped tracer to 2.9.3-SNAPSHOT for end-to-end correlationId + tenantId propagation across HTTP↔Kafka (HCM flow tracing)
+
## 2.9.3 - 2026-03-16
- Upgraded Spring Boot version from 3.2.2 to 3.4.5 to fix HIGH/CRITICAL CVEs
- Upgraded tracer, services-common, mdms-client, enc-client library versions
diff --git a/core-services/egov-accesscontrol/pom.xml b/core-services/egov-accesscontrol/pom.xml
index 7d5bbaad3f2..f98dfae9cb4 100644
--- a/core-services/egov-accesscontrol/pom.xml
+++ b/core-services/egov-accesscontrol/pom.xml
@@ -5,7 +5,7 @@
org.egovegov-accesscontrol
- 2.9.3-SNAPSHOT
+ 2.9.4-SNAPSHOTjaregov-accesscontrol
@@ -57,7 +57,7 @@
org.egov.servicestracer
- 2.9.2-SNAPSHOT
+ 2.9.3-SNAPSHOT
diff --git a/core-services/egov-enc-service/CHANGELOG.md b/core-services/egov-enc-service/CHANGELOG.md
index 5bc6b351043..1e35728c2ba 100644
--- a/core-services/egov-enc-service/CHANGELOG.md
+++ b/core-services/egov-enc-service/CHANGELOG.md
@@ -3,6 +3,9 @@
# Changelog
All notable changes to this module will be documented in this file.
+## 2.9.4 - 2026-07-27
+- Bumped tracer to 2.9.3-SNAPSHOT for end-to-end correlationId + tenantId propagation across HTTP↔Kafka (HCM flow tracing)
+
## 2.9.3 - 2026-03-16
- Upgraded Spring Boot version from 3.2.2 to 3.4.5 to fix HIGH/CRITICAL CVEs
- Upgraded tracer, services-common, mdms-client, enc-client library versions
diff --git a/core-services/egov-enc-service/pom.xml b/core-services/egov-enc-service/pom.xml
index 4d094a46ab4..8f7629e894b 100644
--- a/core-services/egov-enc-service/pom.xml
+++ b/core-services/egov-enc-service/pom.xml
@@ -9,7 +9,7 @@
org.egovegov-enc-service
- 2.9.3-SNAPSHOT
+ 2.9.4-SNAPSHOTegov-enc-service17
@@ -71,7 +71,7 @@
org.egov.servicestracer
- 2.9.2-SNAPSHOT
+ 2.9.3-SNAPSHOTorg.projectlombok
diff --git a/core-services/egov-filestore/CHANGELOG.md b/core-services/egov-filestore/CHANGELOG.md
index d725d85a0db..3d4f8c8517f 100644
--- a/core-services/egov-filestore/CHANGELOG.md
+++ b/core-services/egov-filestore/CHANGELOG.md
@@ -1,6 +1,9 @@
# Changelog
All notable changes to this module will be documented in this file.
+## 2.9.4 - 2026-07-27
+- Bumped tracer to 2.9.3-SNAPSHOT for end-to-end correlationId + tenantId propagation across HTTP↔Kafka (HCM flow tracing)
+
## 2.9.3 - 2026-03-16
- Upgraded Spring Boot version from 3.2.2 to 3.4.5 to fix HIGH/CRITICAL CVEs
- Upgraded tracer, services-common, mdms-client, enc-client library versions
diff --git a/core-services/egov-filestore/pom.xml b/core-services/egov-filestore/pom.xml
index c938b5c9c6f..93f6ddb2ebf 100644
--- a/core-services/egov-filestore/pom.xml
+++ b/core-services/egov-filestore/pom.xml
@@ -12,7 +12,7 @@
org.egovegov-filestore
- 2.9.3-SNAPSHOT
+ 2.9.4-SNAPSHOTegov-filestoreeGov File store project for eGov services
@@ -53,7 +53,7 @@
org.egov.servicestracer
- 2.9.2-SNAPSHOT
+ 2.9.3-SNAPSHOTorg.springframework.boot
diff --git a/core-services/egov-indexer/CHANGELOG.md b/core-services/egov-indexer/CHANGELOG.md
index 94968009bea..c11f55e746b 100644
--- a/core-services/egov-indexer/CHANGELOG.md
+++ b/core-services/egov-indexer/CHANGELOG.md
@@ -3,6 +3,9 @@
# Changelog
All notable changes to this module will be documented in this file.
+## 2.9.4 - 2026-07-27
+- Bumped tracer to 2.9.3-SNAPSHOT for end-to-end correlationId + tenantId propagation across HTTP↔Kafka (HCM flow tracing)
+
## 2.9.3 - 2026-03-16
- Upgraded Spring Boot version from 3.2.2 to 3.4.5 to fix HIGH/CRITICAL CVEs
- Upgraded tracer, services-common, mdms-client, enc-client library versions
diff --git a/core-services/egov-indexer/pom.xml b/core-services/egov-indexer/pom.xml
index f19aeb18d57..70ca035bad1 100644
--- a/core-services/egov-indexer/pom.xml
+++ b/core-services/egov-indexer/pom.xml
@@ -10,7 +10,7 @@
org.egovegov-indexer
- 2.9.3-SNAPSHOT
+ 2.9.4-SNAPSHOTegov-indexeregov indexer frameworkhttp://maven.apache.org
@@ -54,7 +54,7 @@
org.egov.servicestracer
- 2.9.2-SNAPSHOT
+ 2.9.3-SNAPSHOTorg.springframework.boot
diff --git a/core-services/egov-localization/CHANGELOG.md b/core-services/egov-localization/CHANGELOG.md
index cef8652ac11..110e26339ef 100644
--- a/core-services/egov-localization/CHANGELOG.md
+++ b/core-services/egov-localization/CHANGELOG.md
@@ -2,6 +2,9 @@
# Changelog
All notable changes to this module will be documented in this file.
+## 2.9.4 - 2026-07-27
+- Bumped tracer to 2.9.3-SNAPSHOT for end-to-end correlationId + tenantId propagation across HTTP↔Kafka (HCM flow tracing)
+
## 2.9.3 - 2026-03-16
- Upgraded Spring Boot version from 3.2.2 to 3.4.5 to fix HIGH/CRITICAL CVEs
- Upgraded tracer, services-common, mdms-client, enc-client library versions
diff --git a/core-services/egov-localization/pom.xml b/core-services/egov-localization/pom.xml
index 5e26260ba0b..ed292989fec 100644
--- a/core-services/egov-localization/pom.xml
+++ b/core-services/egov-localization/pom.xml
@@ -9,7 +9,7 @@
org.egovegov-localization
- 2.9.3-SNAPSHOT
+ 2.9.4-SNAPSHOTegov-localizationLocalization for messages
@@ -91,7 +91,7 @@
org.egov.servicestracer
- 2.9.2-SNAPSHOT
+ 2.9.3-SNAPSHOTorg.springframework.boot
diff --git a/core-services/egov-mdms-service/CHANGELOG.md b/core-services/egov-mdms-service/CHANGELOG.md
index 3d6e96fec74..33b43e20526 100644
--- a/core-services/egov-mdms-service/CHANGELOG.md
+++ b/core-services/egov-mdms-service/CHANGELOG.md
@@ -3,6 +3,9 @@
# Changelog
All notable changes to this module will be documented in this file.
+## 2.9.2 - 2026-07-27
+- Bumped tracer to 2.9.3-SNAPSHOT for end-to-end correlationId + tenantId propagation across HTTP↔Kafka (HCM flow tracing)
+
## 2.9.1 - 2025-05-21
- Upgraded tracer version from 2.9.0 to 2.9.1
- added variables in application.properties required for opentelemetry
diff --git a/core-services/egov-mdms-service/pom.xml b/core-services/egov-mdms-service/pom.xml
index 1eee10a5383..0981ed0e01d 100644
--- a/core-services/egov-mdms-service/pom.xml
+++ b/core-services/egov-mdms-service/pom.xml
@@ -9,7 +9,7 @@
org.egov.mdmsegov-mdms-service-test
- 2.9.1-SNAPSHOT
+ 2.9.2-SNAPSHOTegov-infra-mdms-servicehttp://maven.apache.org
@@ -74,7 +74,7 @@
org.egov.servicestracer
- 2.9.2-SNAPSHOT
+ 2.9.3-SNAPSHOTorg.egov
diff --git a/core-services/egov-persister/CHANGELOG.md b/core-services/egov-persister/CHANGELOG.md
index 9cc3eaf0a5f..d0e332c7984 100644
--- a/core-services/egov-persister/CHANGELOG.md
+++ b/core-services/egov-persister/CHANGELOG.md
@@ -2,6 +2,9 @@
# Changelog
All notable changes to this module will be documented in this file.
+## 2.9.4 - 2026-07-27
+- Bumped tracer to 2.9.3-SNAPSHOT for end-to-end correlationId + tenantId propagation across HTTP↔Kafka (HCM flow tracing)
+
## 2.9.3 - 2026-03-16
- Upgraded Spring Boot version from 3.2.2 to 3.4.5 to fix HIGH/CRITICAL CVEs
- Upgraded tracer, services-common, mdms-client, enc-client library versions
diff --git a/core-services/egov-persister/pom.xml b/core-services/egov-persister/pom.xml
index cd599e83aae..dd142e84bad 100644
--- a/core-services/egov-persister/pom.xml
+++ b/core-services/egov-persister/pom.xml
@@ -9,7 +9,7 @@
org.egovegov-persister
- 2.9.3-SNAPSHOT
+ 2.9.4-SNAPSHOTegov-persisteregov persister framework
@@ -89,7 +89,7 @@
org.egov.servicestracer
- 2.9.2-SNAPSHOT
+ 2.9.3-SNAPSHOTorg.flywaydb
diff --git a/core-services/egov-url-shortening/CHANGELOG.md b/core-services/egov-url-shortening/CHANGELOG.md
index 85ec23be0f1..d46914c88a0 100644
--- a/core-services/egov-url-shortening/CHANGELOG.md
+++ b/core-services/egov-url-shortening/CHANGELOG.md
@@ -3,6 +3,9 @@
# Changelog
All notable changes to this module will be documented in this file.
+## 2.9.4 - 2026-07-27
+- Bumped tracer to 2.9.3-SNAPSHOT for end-to-end correlationId + tenantId propagation across HTTP↔Kafka (HCM flow tracing)
+
## 2.9.3 - 2026-03-16
- Upgraded Spring Boot version from 3.2.2 to 3.4.5 to fix HIGH/CRITICAL CVEs
- Upgraded tracer, services-common, mdms-client, enc-client library versions
diff --git a/core-services/egov-url-shortening/pom.xml b/core-services/egov-url-shortening/pom.xml
index 018d0aaa6af..ef575798263 100644
--- a/core-services/egov-url-shortening/pom.xml
+++ b/core-services/egov-url-shortening/pom.xml
@@ -2,7 +2,7 @@
4.0.0org.egovegov-url-shortening
- 2.9.3-SNAPSHOT
+ 2.9.4-SNAPSHOTorg.springframework.bootspring-boot-starter-parent
@@ -72,7 +72,7 @@
org.egov.servicestracer
- 2.9.2-SNAPSHOT
+ 2.9.3-SNAPSHOT
diff --git a/core-services/egov-workflow-v2/CHANGELOG.md b/core-services/egov-workflow-v2/CHANGELOG.md
index 88d3362ecfe..38099089bae 100644
--- a/core-services/egov-workflow-v2/CHANGELOG.md
+++ b/core-services/egov-workflow-v2/CHANGELOG.md
@@ -3,6 +3,9 @@
# Changelog
All notable changes to this module will be documented in this file.
+## 2.9.4 - 2026-07-27
+- Bumped tracer to 2.9.3-SNAPSHOT for end-to-end correlationId + tenantId propagation across HTTP↔Kafka (HCM flow tracing)
+
## 2.9.3 - 2026-03-16
- Upgraded Spring Boot version from 3.2.2 to 3.4.5 to fix HIGH/CRITICAL CVEs
- Upgraded tracer, services-common, mdms-client, enc-client library versions
diff --git a/core-services/egov-workflow-v2/pom.xml b/core-services/egov-workflow-v2/pom.xml
index f3ccbd1f7d0..6a9b971c5b1 100644
--- a/core-services/egov-workflow-v2/pom.xml
+++ b/core-services/egov-workflow-v2/pom.xml
@@ -4,7 +4,7 @@
egov-workflow-v2jaregov-workflow-v2
- 2.9.3-SNAPSHOT
+ 2.9.4-SNAPSHOT3.1.117
@@ -69,7 +69,7 @@
org.egov.servicestracer
- 2.9.2-SNAPSHOT
+ 2.9.3-SNAPSHOT
diff --git a/core-services/mdms-v2/CHANGELOG.md b/core-services/mdms-v2/CHANGELOG.md
index 9a9cc2d77ea..daac3bad473 100644
--- a/core-services/mdms-v2/CHANGELOG.md
+++ b/core-services/mdms-v2/CHANGELOG.md
@@ -3,6 +3,9 @@
# Changelog
All notable changes to this module will be documented in this file.
+## 1.3.6 - 2026-07-27
+- Bumped tracer to 2.9.3-SNAPSHOT for end-to-end correlationId + tenantId propagation across HTTP↔Kafka (HCM flow tracing)
+
## 1.3.5 - 2026-03-16
- Upgraded Spring Boot version from 3.2.2 to 3.4.5 to fix HIGH/CRITICAL CVEs
- Upgraded tracer, services-common library versions
diff --git a/core-services/mdms-v2/pom.xml b/core-services/mdms-v2/pom.xml
index 5e33c67ee43..7b881579560 100644
--- a/core-services/mdms-v2/pom.xml
+++ b/core-services/mdms-v2/pom.xml
@@ -11,7 +11,7 @@
org.egov.mdmsegov-mdms-service-test
- 1.3.5-SNAPSHOT
+ 1.3.6-SNAPSHOTegov-infra-mdms-servicehttp://maven.apache.org
@@ -62,7 +62,7 @@
org.egov.servicestracer
- 2.9.2-SNAPSHOT
+ 2.9.3-SNAPSHOT
From 879ac5397b7835e4c21443346553d26712bac737 Mon Sep 17 00:00:00 2001
From: bhaswatmandal-egov
Date: Wed, 29 Jul 2026 17:57:08 +0530
Subject: [PATCH 21/24] Restricting characters for hierarchy creation
---
.../main/java/digit/errors/ErrorCodes.java | 4 +
.../validator/BoundaryHierarchyValidator.java | 90 +++++++++++++
.../BoundaryHierarchyTypeSeparatorTest.java | 123 ++++++++++++++++++
3 files changed, 217 insertions(+)
create mode 100644 core-services/boundary-service/src/test/java/digit/service/validator/BoundaryHierarchyTypeSeparatorTest.java
diff --git a/core-services/boundary-service/src/main/java/digit/errors/ErrorCodes.java b/core-services/boundary-service/src/main/java/digit/errors/ErrorCodes.java
index f9efbfa76f0..67377d4cc41 100644
--- a/core-services/boundary-service/src/main/java/digit/errors/ErrorCodes.java
+++ b/core-services/boundary-service/src/main/java/digit/errors/ErrorCodes.java
@@ -44,5 +44,9 @@ public class ErrorCodes {
public static final String BULK_REQUEST_INFO_MISSING_MSG = "Bulk boundary relationship request is missing RequestInfo.userInfo.";
public static final String INVALID_BOUNDARY_CODE_CODE = "INVALID_BOUNDARY_CODE";
public static final String INVALID_BOUNDARY_CODE_MSG = "code, tenantId and hierarchyType must not contain the '|' character, which is reserved as the ancestral materialized-path delimiter.";
+ // Distinct code (not reused from INVALID_HIERARCHY_DEFINITION) so the UI can render a specific,
+ // actionable message for this case rather than a generic hierarchy-definition failure.
+ public static final String INVALID_HIERARCHY_TYPE_SEPARATOR_CODE = "INVALID_HIERARCHY_TYPE_SEPARATOR";
+ public static final String INVALID_HIERARCHY_TYPE_SEPARATOR_MSG = "hierarchyType must not contain separator characters. The characters '.', ':', '-', '/', '_' and any whitespace are all normalized to '_' when deriving the localisation module name and the boundary code prefix, so two hierarchy types differing only by these characters would silently collide. Offending character(s): ";
}
diff --git a/core-services/boundary-service/src/main/java/digit/service/validator/BoundaryHierarchyValidator.java b/core-services/boundary-service/src/main/java/digit/service/validator/BoundaryHierarchyValidator.java
index 460799ddfa0..02985436092 100644
--- a/core-services/boundary-service/src/main/java/digit/service/validator/BoundaryHierarchyValidator.java
+++ b/core-services/boundary-service/src/main/java/digit/service/validator/BoundaryHierarchyValidator.java
@@ -12,12 +12,38 @@
import java.util.Collections;
import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
import java.util.Map;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
import java.util.stream.Collectors;
@Component
public class BoundaryHierarchyValidator {
+ /**
+ * Characters that must not appear in a hierarchyType.
+ *
+ * Downstream (boundary-management getTransformedLocale, project-factory) every one of these is
+ * rewritten to '_' when deriving the localisation module name (hcm-boundary-<type>) and the
+ * boundary code prefix. Two hierarchy types differing only by these characters therefore collapse
+ * onto the SAME module and the SAME code prefix - e.g. "MODFIX-A", "MODFIX_A" and "MODFIX A" all
+ * become "modfix_a" - and silently overwrite each other's data. '_' itself is included because it
+ * is the normalization target, so it collides with all the others.
+ *
+ * The set is written out explicitly rather than using \s: Java's \s is ASCII-only, and
+ * Character.isWhitespace() returns false for NBSP (U+00A0), figure space (U+2007) and narrow NBSP
+ * (U+202F). This class mirrors JavaScript's \s exactly, so server and client agree character for
+ * character.
+ */
+ private static final Pattern FORBIDDEN_HIERARCHY_TYPE_CHARS = Pattern.compile(
+ "[.:\\-/_" +
+ "\\u0009\\u000A\\u000B\\u000C\\u000D\\u0020" + // tab, LF, VT, FF, CR, space
+ "\\u00A0\\u1680\\u2000-\\u200A" + // NBSP, ogham, en/em/thin/hair spaces
+ "\\u2028\\u2029\\u202F\\u205F\\u3000\\uFEFF" + // line/para sep, narrow NBSP, math, ideographic, BOM
+ "]");
+
private BoundaryHierarchyRepository boundaryHierarchyRepository;
@Autowired
@@ -31,6 +57,10 @@ public BoundaryHierarchyValidator(BoundaryHierarchyRepository boundaryHierarchyR
*/
public void validateBoundaryTypeHierarchy(BoundaryTypeHierarchyRequest body) {
+ // Validate that the hierarchy type carries no separator character that would collide
+ // downstream. Runs first so a bad name is rejected before any DB lookup.
+ validateHierarchyTypeSeparators(body);
+
// Validate if only single root node exists
validateIfSingleRootNodeExists(body);
@@ -73,6 +103,66 @@ private void validateIfBoundaryHierarchyFormsDAG(BoundaryTypeHierarchyRequest bo
});
}
+ /**
+ * Rejects a hierarchyType containing any character that is normalized to '_' downstream, plus '_'
+ * itself. Without this, "MODFIX-A" and "MODFIX_A" are accepted as two distinct hierarchies but
+ * share one localisation module and one boundary code prefix, so the second silently overwrites
+ * the first. Verified live: three such hierarchies produced 30/30 identical boundary codes.
+ *
+ * Thrown with its own error code so the UI can show a specific message; the offending characters
+ * are named in the message (with code points, since several are invisible - NBSP pasted from
+ * Word/Excel being the common one).
+ * @param body
+ */
+ private void validateHierarchyTypeSeparators(BoundaryTypeHierarchyRequest body) {
+
+ String hierarchyType = body.getBoundaryHierarchy().getHierarchyType();
+
+ if (ObjectUtils.isEmpty(hierarchyType))
+ return;
+
+ // Collect every distinct offending character so the caller can fix them all in one go
+ // rather than resubmitting once per bad character.
+ Set offenders = new LinkedHashSet<>();
+ Matcher matcher = FORBIDDEN_HIERARCHY_TYPE_CHARS.matcher(hierarchyType);
+ while (matcher.find()) {
+ char offender = matcher.group().charAt(0);
+ offenders.add(String.format("'%s' (U+%04X)", describe(offender), (int) offender));
+ }
+
+ if (!CollectionUtils.isEmpty(offenders)) {
+ throw new CustomException(ErrorCodes.INVALID_HIERARCHY_TYPE_SEPARATOR_CODE,
+ ErrorCodes.INVALID_HIERARCHY_TYPE_SEPARATOR_MSG + String.join(", ", offenders));
+ }
+ }
+
+ /**
+ * Renders a character for an error message. Invisible characters have no useful glyph, so they
+ * are named instead - an operator seeing "'-' (U+002D)" can act, but a bare NBSP looks like a
+ * space and reads as a service bug.
+ */
+ private String describe(char c) {
+ switch (c) {
+ case '\t': return "tab";
+ case '\n': return "line feed";
+ case '\u000B': return "vertical tab";
+ case '\f': return "form feed";
+ case '\r': return "carriage return";
+ case '\u0020': return "space";
+ case '\u00A0': return "non-breaking space";
+ case '\u2007': return "figure space";
+ case '\u202F': return "narrow non-breaking space";
+ case '\u2028': return "line separator";
+ case '\u2029': return "paragraph separator";
+ case '\u3000': return "ideographic space";
+ case '\uFEFF': return "zero-width no-break space";
+ default:
+ // Remaining matches are either visible punctuation or one of the Unicode space
+ // separators, which have no distinct name worth spelling out individually.
+ return Character.isSpaceChar(c) ? "space separator" : String.valueOf(c);
+ }
+ }
+
/**
* This method validates if only a single root node has been defined in hierarchy definition.
* @param body
diff --git a/core-services/boundary-service/src/test/java/digit/service/validator/BoundaryHierarchyTypeSeparatorTest.java b/core-services/boundary-service/src/test/java/digit/service/validator/BoundaryHierarchyTypeSeparatorTest.java
new file mode 100644
index 00000000000..d7617934208
--- /dev/null
+++ b/core-services/boundary-service/src/test/java/digit/service/validator/BoundaryHierarchyTypeSeparatorTest.java
@@ -0,0 +1,123 @@
+package digit.service.validator;
+
+import digit.errors.ErrorCodes;
+import digit.repository.BoundaryHierarchyRepository;
+import digit.web.models.BoundaryTypeHierarchy;
+import digit.web.models.BoundaryTypeHierarchyDefinition;
+import digit.web.models.BoundaryTypeHierarchyRequest;
+import org.egov.tracer.model.CustomException;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Covers the hierarchyType separator guard: every character that getTransformedLocale rewrites to
+ * '_' (plus '_' itself) must be rejected, and nothing else may be.
+ */
+class BoundaryHierarchyTypeSeparatorTest {
+
+ private BoundaryHierarchyValidator validator() {
+ BoundaryHierarchyRepository repo = mock(BoundaryHierarchyRepository.class);
+ // Not reached for the rejection cases; empty so accepted names fall through cleanly.
+ when(repo.search(org.mockito.ArgumentMatchers.any())).thenReturn(Collections.emptyList());
+ return new BoundaryHierarchyValidator(repo);
+ }
+
+ private BoundaryTypeHierarchyRequest requestFor(String hierarchyType) {
+ BoundaryTypeHierarchy node = BoundaryTypeHierarchy.builder()
+ .boundaryType("COUNTRY")
+ .parentBoundaryType(null)
+ .active(Boolean.TRUE)
+ .build();
+ BoundaryTypeHierarchyDefinition definition = BoundaryTypeHierarchyDefinition.builder()
+ .tenantId("dev")
+ .hierarchyType(hierarchyType)
+ .boundaryHierarchy(List.of(node))
+ .build();
+ return BoundaryTypeHierarchyRequest.builder()
+ .boundaryHierarchy(definition)
+ .build();
+ }
+
+ private static final char[] MUST_REJECT = {
+ '.', ':', '-', '/', '_', '\t',
+ '\n', '\u000B', '\f', '\r', '\u0020', '\u00A0',
+ '\u1680', '\u2000', '\u2001', '\u2002', '\u2003', '\u2004',
+ '\u2005', '\u2006', '\u2007', '\u2008', '\u2009', '\u200A',
+ '\u2028', '\u2029', '\u202F', '\u205F', '\u3000', '\uFEFF'
+ };
+
+ @Test
+ void rejectsEveryCharacterThatNormalizesToUnderscore() {
+ BoundaryHierarchyValidator v = validator();
+ assertEquals(30, MUST_REJECT.length, "expected 29 transform chars + '_' itself");
+
+ for (char c : MUST_REJECT) {
+ String hierarchyType = "MODFIX" + c + "A";
+ CustomException ex = assertThrows(CustomException.class,
+ () -> v.validateBoundaryTypeHierarchy(requestFor(hierarchyType)),
+ () -> String.format("U+%04X should have been rejected", (int) c));
+ assertEquals(ErrorCodes.INVALID_HIERARCHY_TYPE_SEPARATOR_CODE, ex.getCode(),
+ () -> String.format("U+%04X rejected with the wrong error code", (int) c));
+ assertTrue(ex.getMessage().contains(String.format("U+%04X", (int) c)),
+ () -> String.format("message should name the offending code point U+%04X", (int) c));
+ }
+ }
+
+ @Test
+ void acceptsCleanHierarchyTypes() {
+ BoundaryHierarchyValidator v = validator();
+ for (String ok : List.of("MODFIXA", "LNPERF115K", "MICROPLAN", "SIERRALEON", "Nigeria2026")) {
+ assertDoesNotThrow(() -> v.validateBoundaryTypeHierarchy(requestFor(ok)),
+ () -> ok + " should be accepted");
+ }
+ }
+
+ @Test
+ void doesNotRejectSemicolonOrOtherPunctuation() {
+ // ';' looks like it falls in a ':'-to-'s' range but the '-' in the source regex is literal.
+ // Guards against someone "simplifying" the class into an actual range.
+ BoundaryHierarchyValidator v = validator();
+ for (String ok : List.of("MODFIX;A", "MODFIX@A", "MODFIX+A", "MODFIX=A", "MODFIX(A)")) {
+ assertDoesNotThrow(() -> v.validateBoundaryTypeHierarchy(requestFor(ok)),
+ () -> ok + " must not be rejected by the separator guard");
+ }
+ }
+
+ @Test
+ void reportsAllDistinctOffendersAtOnce() {
+ BoundaryHierarchyValidator v = validator();
+ CustomException ex = assertThrows(CustomException.class,
+ () -> v.validateBoundaryTypeHierarchy(requestFor("A-B_C D.E")));
+ assertTrue(ex.getMessage().contains("U+002D"), "should name hyphen");
+ assertTrue(ex.getMessage().contains("U+005F"), "should name underscore");
+ assertTrue(ex.getMessage().contains("U+0020"), "should name space");
+ assertTrue(ex.getMessage().contains("U+002E"), "should name period");
+ }
+
+ @Test
+ void namesInvisibleCharactersReadably() {
+ BoundaryHierarchyValidator v = validator();
+ CustomException nbsp = assertThrows(CustomException.class,
+ () -> v.validateBoundaryTypeHierarchy(requestFor("CHAD" + "\u00A0" + "SMC")));
+ assertTrue(nbsp.getMessage().contains("non-breaking space"),
+ "NBSP must be named, not rendered as an invisible glyph: " + nbsp.getMessage());
+ assertTrue(nbsp.getMessage().contains("U+00A0"));
+ }
+
+ @Test
+ void nullHierarchyTypeIsLeftToOtherValidators() {
+ BoundaryHierarchyValidator v = validator();
+ // Null/blank naming is not this validator's concern; it must not throw the separator error.
+ try {
+ v.validateBoundaryTypeHierarchy(requestFor(null));
+ } catch (CustomException ex) {
+ assertNotEquals(ErrorCodes.INVALID_HIERARCHY_TYPE_SEPARATOR_CODE, ex.getCode());
+ }
+ }
+}
From e70d3290177e2d1ce9bf6ba7c018cc61643bca92 Mon Sep 17 00:00:00 2001
From: nikhilmulinti
Date: Tue, 11 Aug 2026 09:57:23 +0530
Subject: [PATCH 22/24] chore: dummy commit to trigger egov-localization build
for ArgoCD/Kargo pipeline test
---
core-services/egov-localization/README.md | 2 ++
1 file changed, 2 insertions(+)
diff --git a/core-services/egov-localization/README.md b/core-services/egov-localization/README.md
index fc6bcef0a19..ed6ad4f7458 100644
--- a/core-services/egov-localization/README.md
+++ b/core-services/egov-localization/README.md
@@ -31,3 +31,5 @@ Localisation can be search using combination of code , module, tenantid and loca
### Kafka Consumers
### Kafka Producers
+
+
From cd6015865e0fd0f1db4b4136c9ecd4d40c7641df Mon Sep 17 00:00:00 2001
From: hruthvikl-egov
Date: Tue, 11 Aug 2026 14:42:46 +0530
Subject: [PATCH 23/24] remove tracer.kafka.mdc.enabled kill-switch (tracer)
---
core-services/libraries/tracer/CHANGELOG.md | 1 -
core-services/libraries/tracer/readme.md | 8 +++++---
.../java/org/egov/tracer/config/TracerConfiguration.java | 3 ++-
.../libraries/tracer/src/main/resources/tracer.properties | 3 ---
4 files changed, 7 insertions(+), 8 deletions(-)
diff --git a/core-services/libraries/tracer/CHANGELOG.md b/core-services/libraries/tracer/CHANGELOG.md
index 956e8bf578f..2d20a29716b 100644
--- a/core-services/libraries/tracer/CHANGELOG.md
+++ b/core-services/libraries/tracer/CHANGELOG.md
@@ -4,7 +4,6 @@ All notable changes to this library will be documented in this file.
## 2.9.3 - 2026-07-17
- Propagated correlationId and tenantId across Kafka — stamp both onto message headers on produce and rebuild the MDC from them on consume, so consumer/async logs carry the same IDs
- Added MdcRecordInterceptor (auto-attached to the default Kafka listener container factory), KafkaTemplateLoggingInterceptors, and TracerKafkaMdcUtil helper
-- Added kill-switch `tracer.kafka.mdc.enabled` (default true)
## 2.9.2 - 2026-03-10
- Addition of Data Access Exception Handling in ExceptionAdvice.
diff --git a/core-services/libraries/tracer/readme.md b/core-services/libraries/tracer/readme.md
index 0f763b4fc7b..71e89ced5e1 100644
--- a/core-services/libraries/tracer/readme.md
+++ b/core-services/libraries/tracer/readme.md
@@ -41,9 +41,11 @@ Map map = new HashMap<>();
The logging of the http request/response body and Kakfa message body can be toggled on/off using
"tracer.detailed.tracing.enabled" application property.
-The per-record Kafka MDC rebuild + cleanup (see "Setting the correlation id in the MDC") can be
-toggled on/off using the "tracer.kafka.mdc.enabled" application property. It defaults to true; set it
-to false to disable registration of the record interceptor.
+The per-record Kafka MDC rebuild + cleanup (see "Setting the correlation id in the MDC") is not
+toggleable. Correlation/tenant propagation over Kafka lives in the client interceptor registered by
+"spring.kafka.properties.interceptor.classes", so disabling the record interceptor would have left
+propagation running while dropping per-record cleanup — mis-attributing every record in a poll to the
+last one. Remove the client interceptor property if the Kafka hop must be switched off entirely.
###### Correlation id retrieval and forwarding -
diff --git a/core-services/libraries/tracer/src/main/java/org/egov/tracer/config/TracerConfiguration.java b/core-services/libraries/tracer/src/main/java/org/egov/tracer/config/TracerConfiguration.java
index 19797784caf..93e39de38e8 100644
--- a/core-services/libraries/tracer/src/main/java/org/egov/tracer/config/TracerConfiguration.java
+++ b/core-services/libraries/tracer/src/main/java/org/egov/tracer/config/TracerConfiguration.java
@@ -42,8 +42,9 @@ public RestTemplate logAwareRestTemplate(RestTemplateBuilder builder, TracerProp
// MDC rebuild + cleanup from Kafka headers. Boot auto-applies to the autoconfig record-listener
// factory; custom-factory or batch-listener services must wire it manually.
+ // Not conditional: propagation itself lives in the Kafka client interceptor, so gating this bean
+ // only dropped per-record cleanup and left MDC values leaking across records in a poll.
@Bean
- @ConditionalOnProperty(name = "tracer.kafka.mdc.enabled", havingValue = "true", matchIfMissing = true)
public RecordInterceptor mdcRecordInterceptor() {
return new MdcRecordInterceptor<>();
}
diff --git a/core-services/libraries/tracer/src/main/resources/tracer.properties b/core-services/libraries/tracer/src/main/resources/tracer.properties
index 8aea2169bb6..84a88389fea 100644
--- a/core-services/libraries/tracer/src/main/resources/tracer.properties
+++ b/core-services/libraries/tracer/src/main/resources/tracer.properties
@@ -15,9 +15,6 @@ tracer.filterSkipPattern=/api-docs.*|/autoconfig|/configprops|/dump|/health|/inf
#Logging config
spring.kafka.properties.interceptor.classes=org.egov.tracer.kafka.KafkaTemplateLoggingInterceptors
-# per-record MDC rebuild + cleanup on Kafka consume (default on)
-tracer.kafka.mdc.enabled=true
-
# Actuator Configs
endpoints.enabled=false
endpoints.health.enabled=true
From 71b89966335aa525ba0cd5f9467e4327f3c4b9fa Mon Sep 17 00:00:00 2001
From: bhaswatmandal-egov
Date: Tue, 25 Aug 2026 17:26:11 +0530
Subject: [PATCH 24/24] Readonly db case added to the db exception classifier
---
.../consumer/DbExceptionClassifier.java | 23 ++++-
.../persist/consumer/DbHealthMonitor.java | 24 ++++-
.../consumer/DbExceptionClassifierTest.java | 39 +++++++-
.../persist/consumer/DbHealthMonitorTest.java | 88 +++++++++++++++++++
4 files changed, 171 insertions(+), 3 deletions(-)
create mode 100644 core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/DbHealthMonitorTest.java
diff --git a/core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/DbExceptionClassifier.java b/core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/DbExceptionClassifier.java
index a9fe95fdc9b..2a203555393 100644
--- a/core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/DbExceptionClassifier.java
+++ b/core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/DbExceptionClassifier.java
@@ -15,6 +15,20 @@
*
PERMANENT — constraint / data / grammar errors: retrying will always fail the same way.
*
+ *
+ *
Note on 25006 (read_only_sql_transaction). SQLSTATE class 25 is "Invalid Transaction
+ * State" and is normally a programming error, so only this one member of the class is treated as
+ * transient — not the class as a whole. A managed Postgres instance can turn the whole server
+ * read-only for a bounded window (failover, maintenance, a storage-full condition), during which
+ * every write fails with 25006 and every one of them succeeds again once the window closes.
+ * Observed on ng-central-dev / mhbase on 2026-08-25: the server flipped read-only repeatedly and
+ * a single ~35s window dead-lettered 4,521 records, logged as 46 consecutive batches of
+ * "100 record(s) -> 0 persisted, 0 duplicate(s), 100 dead-lettered, 0 parked". Whole batches
+ * failing identically is the signature of an environmental fault, not of per-record corruption.
+ * Note this path is reached only when the driver exception carries the SQLSTATE: the same outage's
+ * connection kills arrived as DataAccessResourceFailureException and were already classified
+ * TRANSIENT by the cause-chain scan below, but the read-only writes arrived as
+ * UncategorizedSQLException, which that scan does not match.
*/
public final class DbExceptionClassifier {
@@ -29,7 +43,14 @@ public static Kind classify(Throwable t) {
}
if (sqlState != null && (
sqlState.startsWith("08") // connection exception
- || sqlState.startsWith("57") // operator intervention (e.g. admin shutdown, query cancel)
+ // Operator intervention: admin/crash shutdown, cannot_connect_now, database_dropped.
+ // The server is going away, so the in-place retry path (paired with the pause-on-DB-health
+ // backstop) is right. EXCLUDING 57014 query_canceled, which is a per-statement
+ // cancellation or timeout, not an outage: the next attempt is likely to be cancelled the
+ // same way, so retrying it in place can loop indefinitely. It must fall through to the
+ // bounded DLQ / parking flow instead.
+ || (sqlState.startsWith("57") && !"57014".equals(sqlState))
+ || "25006".equals(sqlState) // read_only_sql_transaction — see note below
|| "40001".equals(sqlState) // serialization_failure
|| "40P01".equals(sqlState) // deadlock_detected
|| "53300".equals(sqlState) // too_many_connections
diff --git a/core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/DbHealthMonitor.java b/core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/DbHealthMonitor.java
index a6f4692e8f2..4af1867580c 100644
--- a/core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/DbHealthMonitor.java
+++ b/core-services/egov-persister/src/main/java/org/egov/infra/persist/consumer/DbHealthMonitor.java
@@ -46,11 +46,33 @@ public void checkDatasource() {
}
}
+ /**
+ * Healthy means "a write would be accepted", not merely "the server answers".
+ *
+ *
A read-only Postgres serves SELECTs perfectly, so a bare {@code SELECT 1} reports healthy
+ * throughout a read-only window while every INSERT/UPDATE fails. Observed on ng-central-dev /
+ * mhbase on 2026-08-25: the server flipped read-only repeatedly, this monitor never paused, and
+ * the consumer kept pulling work it could not persist. Both a hot standby and a server carrying
+ * {@code default_transaction_read_only=on} (how a managed instance signals failover, maintenance
+ * or a storage-full condition) report {@code transaction_read_only = on}, so that setting is the
+ * probe. {@code pg_is_in_recovery()} is checked too as a belt-and-braces signal for standby.
+ * Both are cheap, side-effect free, and need no write.
+ */
private boolean isHealthy() {
try {
- jdbcTemplate.queryForObject("SELECT 1", Integer.class);
+ String readOnly = jdbcTemplate.queryForObject("SHOW transaction_read_only", String.class);
+ if ("on".equalsIgnoreCase(readOnly)) {
+ log.debug("Datasource probe: server is read-only (transaction_read_only=on)");
+ return false;
+ }
+ Boolean inRecovery = jdbcTemplate.queryForObject("SELECT pg_is_in_recovery()", Boolean.class);
+ if (Boolean.TRUE.equals(inRecovery)) {
+ log.debug("Datasource probe: server is in recovery (standby)");
+ return false;
+ }
return true;
} catch (Exception e) {
+ // Covers unreachable/down as well - the probe itself throws.
log.debug("Datasource probe failed: {}", e.getMessage());
return false;
}
diff --git a/core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/DbExceptionClassifierTest.java b/core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/DbExceptionClassifierTest.java
index 057dfe898f5..232185fefce 100644
--- a/core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/DbExceptionClassifierTest.java
+++ b/core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/DbExceptionClassifierTest.java
@@ -3,6 +3,7 @@
import org.junit.jupiter.api.Test;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.jdbc.CannotGetJdbcConnectionException;
+import org.springframework.jdbc.UncategorizedSQLException;
import java.sql.SQLException;
@@ -28,7 +29,7 @@ void connectionAndConcurrencyStatesAreTransient() {
assertEquals(Kind.TRANSIENT, DbExceptionClassifier.classify(sqlState("08001"))); // unable to connect
assertEquals(Kind.TRANSIENT, DbExceptionClassifier.classify(sqlState("40001"))); // serialization_failure
assertEquals(Kind.TRANSIENT, DbExceptionClassifier.classify(sqlState("40P01"))); // deadlock_detected
- assertEquals(Kind.TRANSIENT, DbExceptionClassifier.classify(sqlState("57014"))); // query_canceled
+ assertEquals(Kind.TRANSIENT, DbExceptionClassifier.classify(sqlState("57P01"))); // admin_shutdown
assertEquals(Kind.TRANSIENT, DbExceptionClassifier.classify(sqlState("53300"))); // too_many_connections
assertEquals(Kind.TRANSIENT, DbExceptionClassifier.classify(sqlState("55P03"))); // lock_not_available
}
@@ -42,6 +43,18 @@ void constraintAndDataStatesArePermanent() {
assertEquals(Kind.PERMANENT, DbExceptionClassifier.classify(new RuntimeException("no sqlstate at all")));
}
+ @Test
+ void queryCanceledIsSplitOutOfTheOperatorInterventionBucket() {
+ // 57014 query_canceled is a per-statement cancellation/timeout, not the server going away.
+ // In-place retry can loop indefinitely on it, so it must reach the bounded DLQ/parking flow.
+ assertEquals(Kind.PERMANENT, DbExceptionClassifier.classify(sqlState("57014")));
+
+ // The rest of class 57 is a genuine outage and stays transient.
+ assertEquals(Kind.TRANSIENT, DbExceptionClassifier.classify(sqlState("57P01"))); // admin_shutdown
+ assertEquals(Kind.TRANSIENT, DbExceptionClassifier.classify(sqlState("57P02"))); // crash_shutdown
+ assertEquals(Kind.TRANSIENT, DbExceptionClassifier.classify(sqlState("57P03"))); // cannot_connect_now
+ }
+
@Test
void classifiesThroughTheCauseChain() {
// A real failure is wrapped several layers deep (JdbcTemplate -> DataAccessException -> SQLException).
@@ -57,4 +70,28 @@ void connectionAcquisitionFailureIsTransientEvenWithoutSqlState() {
Throwable ex = new CannotGetJdbcConnectionException("could not get connection");
assertEquals(Kind.TRANSIENT, DbExceptionClassifier.classify(ex));
}
+
+ @Test
+ void readOnlyTransactionIsTransientButTheRestOfClass25IsNot() {
+ // 25006: a managed Postgres can turn the whole server read-only for a bounded window
+ // (failover, maintenance, storage full), after which the identical write succeeds.
+ assertEquals(Kind.TRANSIENT, DbExceptionClassifier.classify(sqlState("25006"))); // read_only_sql_transaction
+
+ // Only that one member of class 25 is transient. The rest is genuine invalid-transaction-state
+ // and must stay permanent - retrying it would loop forever.
+ assertEquals(Kind.PERMANENT, DbExceptionClassifier.classify(sqlState("25001"))); // active_sql_transaction
+ assertEquals(Kind.PERMANENT, DbExceptionClassifier.classify(sqlState("25P02"))); // in_failed_sql_transaction
+ }
+
+ @Test
+ void readOnlyTransactionIsTransientInItsProductionWrapping() {
+ // The shape seen on ng-central-dev / mhbase 2026-08-25: Spring surfaces it as
+ // UncategorizedSQLException, whose class name the cause-chain scan does NOT match. This case
+ // therefore passes only via the SQLSTATE branch - it is the regression guard for that path.
+ Throwable ex = new UncategorizedSQLException(
+ "PreparedStatementCallback",
+ "UPDATE mhbase.eg_cm_campaign_data SET data = ?",
+ sqlState("25006"));
+ assertEquals(Kind.TRANSIENT, DbExceptionClassifier.classify(ex));
+ }
}
diff --git a/core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/DbHealthMonitorTest.java b/core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/DbHealthMonitorTest.java
new file mode 100644
index 00000000000..704c2b0afe2
--- /dev/null
+++ b/core-services/egov-persister/src/test/java/org/egov/infra/persist/consumer/DbHealthMonitorTest.java
@@ -0,0 +1,88 @@
+package org.egov.infra.persist.consumer;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * The monitor must gate on WRITABILITY, not reachability.
+ *
+ *
A read-only Postgres answers SELECTs normally, so the previous {@code SELECT 1} probe reported
+ * healthy for the whole of a read-only window while every write failed. On ng-central-dev / mhbase
+ * on 2026-08-25 that left the consumer running against a server that could not accept a single
+ * INSERT or UPDATE. These tests pin the corrected behaviour.
+ */
+class DbHealthMonitorTest {
+
+ private static final String READ_ONLY_PROBE = "SHOW transaction_read_only";
+ private static final String RECOVERY_PROBE = "SELECT pg_is_in_recovery()";
+
+ private static JdbcTemplate jdbc(String transactionReadOnly, Boolean inRecovery) {
+ JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class);
+ when(jdbcTemplate.queryForObject(eq(READ_ONLY_PROBE), eq(String.class))).thenReturn(transactionReadOnly);
+ when(jdbcTemplate.queryForObject(eq(RECOVERY_PROBE), eq(Boolean.class))).thenReturn(inRecovery);
+ return jdbcTemplate;
+ }
+
+ @Test
+ void pausesWhenServerIsReadOnly() {
+ PersisterConsumerConfig consumerConfig = mock(PersisterConsumerConfig.class);
+ new DbHealthMonitor(jdbc("on", Boolean.FALSE), consumerConfig).checkDatasource();
+
+ verify(consumerConfig).pauseContainer();
+ verify(consumerConfig, never()).resumeContainer();
+ }
+
+ @Test
+ void pausesWhenServerIsInRecovery() {
+ PersisterConsumerConfig consumerConfig = mock(PersisterConsumerConfig.class);
+ new DbHealthMonitor(jdbc("off", Boolean.TRUE), consumerConfig).checkDatasource();
+
+ verify(consumerConfig).pauseContainer();
+ }
+
+ @Test
+ void doesNotPauseWhenServerIsWritable() {
+ // Control: without this the read-only assertions above would pass even if the monitor
+ // paused unconditionally.
+ PersisterConsumerConfig consumerConfig = mock(PersisterConsumerConfig.class);
+ new DbHealthMonitor(jdbc("off", Boolean.FALSE), consumerConfig).checkDatasource();
+
+ verify(consumerConfig, never()).pauseContainer();
+ verify(consumerConfig, never()).resumeContainer();
+ }
+
+ @Test
+ void resumesOnceTheReadOnlyWindowCloses() {
+ PersisterConsumerConfig consumerConfig = mock(PersisterConsumerConfig.class);
+ JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class);
+ when(jdbcTemplate.queryForObject(eq(RECOVERY_PROBE), eq(Boolean.class))).thenReturn(Boolean.FALSE);
+ // Read-only on the first poll, writable on the second - the real flapping shape.
+ when(jdbcTemplate.queryForObject(eq(READ_ONLY_PROBE), eq(String.class))).thenReturn("on", "off");
+
+ DbHealthMonitor monitor = new DbHealthMonitor(jdbcTemplate, consumerConfig);
+ monitor.checkDatasource();
+ monitor.checkDatasource();
+
+ verify(consumerConfig).pauseContainer();
+ verify(consumerConfig).resumeContainer();
+ }
+
+ @Test
+ void pausesWhenTheProbeItselfFails() {
+ // Server unreachable: the probe throws rather than returning a value.
+ PersisterConsumerConfig consumerConfig = mock(PersisterConsumerConfig.class);
+ JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class);
+ when(jdbcTemplate.queryForObject(eq(READ_ONLY_PROBE), eq(String.class)))
+ .thenThrow(new RuntimeException("connection refused"));
+
+ new DbHealthMonitor(jdbcTemplate, consumerConfig).checkDatasource();
+
+ verify(consumerConfig).pauseContainer();
+ }
+}