Boundary and Persister Service changes

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

core-services/boundary-service/pom.xml

@@ -4,7 +4,7 @@
     <artifactId>boundary-service</artifactId>
     <packaging>jar</packaging>
     <name>boundary-service</name>
-    <version>1.0.1</version>
+    <version>1.0.2</version>
     <properties>
     <java.version>17</java.version>
     <maven.compiler.source>${java.version}</maven.compiler.source>
@@ -90,7 +90,7 @@
         <dependency>
             <groupId>org.egov.services</groupId>
             <artifactId>tracer</artifactId>
-            <version>2.9.2-SNAPSHOT</version>
+            <version>2.9.3-SNAPSHOT</version>
         </dependency>
         <dependency>
             <groupId>org.egov.services</groupId>

core-services/boundary-service/src/main/java/digit/config/ApplicationProperties.java

@@ -99,6 +99,17 @@ 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}")
+    private Integer bulkCreateMaxSize;
+
     @Value("${boundary.default.offset}")
     private Integer defaultOffset;
 

core-services/boundary-service/src/main/java/digit/errors/ErrorCodes.java

@@ -30,5 +30,23 @@ public class ErrorCodes {
     public static final String MULTIPLE_ROOT_NODES_ERR_MSG = "Hierarchy definition must have only one root node.";
     public static final String DUPLICATE_BOUNDARY_CODE = "DUPLICATE_BOUNDARY";
     public static final String DUPLICATE_BOUNDARY_MSG = "Duplicate boundary entities found in the request.";
+    public static final String DUPLICATE_RECORD_IN_REQUEST_CODE = "DUPLICATE_RECORD_IN_REQUEST";
+    public static final String DUPLICATE_RECORD_IN_REQUEST_MSG = "Duplicate boundary relationship (same tenantId, hierarchyType and code) found within the bulk request.";
+    public static final String BULK_RELATIONSHIP_VALIDATION_ERROR_CODE = "BULK_RELATIONSHIP_VALIDATION_ERROR";
+    public static final String BULK_RELATIONSHIP_VALIDATION_ERROR_MSG = "Boundary relationship could not be validated due to an unexpected error.";
+    public static final String BULK_RELATIONSHIP_PERSIST_TRANSIENT_CODE = "BULK_RELATIONSHIP_PERSIST_TRANSIENT";
+    public static final String BULK_RELATIONSHIP_PERSIST_TRANSIENT_MSG = "Boundary relationship persistence failed transiently and will be retried.";
+    public static final String BULK_REQUEST_EMPTY_CODE = "BULK_REQUEST_EMPTY";
+    public static final String BULK_REQUEST_EMPTY_MSG = "Bulk boundary relationship request must contain at least one relationship.";
+    public static final String BULK_REQUEST_SIZE_EXCEEDED_CODE = "BULK_REQUEST_SIZE_EXCEEDED";
+    public static final String BULK_REQUEST_SIZE_EXCEEDED_MSG = "Bulk boundary relationship request exceeds the maximum allowed size of ";
+    public static final String BULK_REQUEST_INFO_MISSING_CODE = "BULK_REQUEST_INFO_MISSING";
+    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): ";
 
 }

core-services/boundary-service/src/main/java/digit/kafka/Producer.java

@@ -17,4 +17,18 @@ public class Producer {
     public void push(String topic, Object value) {
         kafkaTemplate.send(topic, value);
     }
+
+    /**
+     * Keyed publish: routes the message to a partition by {@code key} so all messages sharing a key
+     * are ordered on the same partition. Used by the bulk path to key a batch by its parent code, so
+     * batches of siblings under the same parent keep a deterministic per-parent order. A null key
+     * falls back to the keyless (default-partitioner) behaviour of {@link #push(String, Object)}.
+     */
+    public void push(String topic, String key, Object value) {
+        if (key == null) {
+            kafkaTemplate.send(topic, value);
+        } else {
+            kafkaTemplate.send(topic, key, value);
+        }
+    }
 }

core-services/boundary-service/src/main/java/digit/repository/BoundaryRelationshipRepository.java

@@ -1,15 +1,36 @@
 package digit.repository;
 
+import digit.web.models.BoundaryRelation;
 import digit.web.models.BoundaryRelationshipDTO;
 import digit.web.models.BoundaryRelationshipRequest;
 import digit.web.models.BoundaryRelationshipRequestDTO;
 import digit.web.models.BoundaryRelationshipSearchCriteria;
+import org.egov.common.contract.request.RequestInfo;
 import java.util.List;
 
 public interface BoundaryRelationshipRepository {
 
     public void create(BoundaryRelationshipRequest boundaryRelationshipRequest);
 
+    /**
+     * 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 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
+     */
+    public void createBulk(List<BoundaryRelation> boundaryRelationships, RequestInfo requestInfo);
+
     public void update(BoundaryRelationshipRequestDTO boundaryRelationshipRequest);
 
     public List<BoundaryRelationshipDTO> search(BoundaryRelationshipSearchCriteria boundaryRelationshipSearchCriteria);

core-services/boundary-service/src/main/java/digit/repository/impl/BoundaryRelationshipRepositoryImpl.java

@@ -6,9 +6,11 @@
 import digit.repository.querybuilder.BoundaryRelationshipQueryBuilder;
 import digit.repository.rowmapper.BoundaryRelationshipRowMapper;
 import digit.web.models.*;
+import org.egov.common.contract.request.RequestInfo;
 import org.springframework.beans.BeanUtils;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.stereotype.Repository;
+import org.springframework.util.CollectionUtils;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -49,6 +51,87 @@ public void create(BoundaryRelationshipRequest boundaryRelationshipRequest) {
         producer.push(applicationProperties.getCreateBoundaryRelationshipTopic(), boundaryRelationshipRequestDTO);
     }
 
+    /**
+     * 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 — 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.
+     *
+     * <p>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.</p>
+     *
+     * <p>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.</p>
+     *
+     * @param boundaryRelationships validated and enriched relationships to persist
+     * @param requestInfo request info propagated onto the published message
+     */
+    @Override
+    public void createBulk(List<BoundaryRelation> 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<BoundaryRelationshipDTO> boundaryRelationshipDTOs = new ArrayList<>(boundaryRelationships.size());
+        for (BoundaryRelation boundaryRelationship : boundaryRelationships) {
+            boundaryRelationshipDTOs.add(convertRelationPOJOToDTO(boundaryRelationship));
+        }
+
+        BulkBoundaryRelationshipRequestDTO batchMessage = BulkBoundaryRelationshipRequestDTO.builder()
+                .requestInfo(requestInfo)
+                .boundaryRelationship(boundaryRelationshipDTOs)
+                .build();
+
+        // Publish the whole validated list as ONE message to the dedicated bulk topic.
+        producer.push(applicationProperties.getBulkCreateBoundaryRelationshipJobTopic(), 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<BoundaryRelation> 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;
+    }
+
     /**
      * This method implements boundary relationship interface's update method. In this implementation
      * it pushes the request to kafka for persister to pick it up and perform update.

core-services/boundary-service/src/main/java/digit/repository/querybuilder/BoundaryEntityQueryBuilder.java

@@ -51,8 +51,12 @@ private String buildQuery(BoundarySearchCriteria boundarySearchCriteria , List<O
         }
         if (!Objects.isNull(boundarySearchCriteria.getCodes())) {
             QueryUtil.addClauseIfRequired(builder , preparedStmtList);
-            builder.append(" boundary.code IN ( ").append(QueryUtil.createQuery(boundarySearchCriteria.getCodes().size())).append(" )");
+            // Deduplicate first so the placeholder count matches the (deduped) bind values. Using the
+            // raw list size for the "?" placeholders while binding a HashSet of values leaves a trailing
+            // placeholder unbound when the batch contains duplicate codes -> PostgreSQL "No value
+            // specified for parameter".
             Set<String> codes = new HashSet<>(boundarySearchCriteria.getCodes());
+            builder.append(" boundary.code IN ( ").append(QueryUtil.createQuery(codes.size())).append(" )");
             QueryUtil.addToPreparedStatement(preparedStmtList , codes);
         }
         return builder.toString();

core-services/boundary-service/src/main/java/digit/repository/querybuilder/BoundaryRelationshipQueryBuilder.java

@@ -7,6 +7,7 @@
 import org.springframework.util.ObjectUtils;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 
 @Component
 public class BoundaryRelationshipQueryBuilder {
@@ -52,8 +53,11 @@ private String buildQuery(BoundaryRelationshipSearchCriteria boundaryRelationshi
 
             if (!CollectionUtils.isEmpty(boundaryRelationshipSearchCriteria.getCodes())) {
                 QueryUtil.addClauseIfRequired(builder, preparedStmtList);
-                builder.append(" code IN ( ").append(QueryUtil.createQuery(boundaryRelationshipSearchCriteria.getCodes().size())).append(" )");
-                QueryUtil.addToPreparedStatement(preparedStmtList, new HashSet<>(boundaryRelationshipSearchCriteria.getCodes()));
+                // Deduplicate so the placeholder count matches the bind values (duplicate codes would
+                // otherwise leave a "?" without a value -> "No value specified for parameter").
+                Set<String> codeSet = new HashSet<>(boundaryRelationshipSearchCriteria.getCodes());
+                builder.append(" code IN ( ").append(QueryUtil.createQuery(codeSet.size())).append(" )");
+                QueryUtil.addToPreparedStatement(preparedStmtList, codeSet);
             }
         }
 
@@ -64,9 +68,11 @@ private String buildQuery(BoundaryRelationshipSearchCriteria boundaryRelationshi
 
         if(!CollectionUtils.isEmpty(boundaryRelationshipSearchCriteria.getCurrentBoundaryCodes())) {
             QueryUtil.addClauseIfRequired(builder, preparedStmtList);
-            builder.append(" ARRAY [ ").append(QueryUtil.createQuery(boundaryRelationshipSearchCriteria.getCurrentBoundaryCodes().size())).append(" ]").append("::text[] ");
+            // Deduplicate so the placeholder count matches the bind values (see note above).
+            Set<String> currentCodeSet = new HashSet<>(boundaryRelationshipSearchCriteria.getCurrentBoundaryCodes());
+            builder.append(" ARRAY [ ").append(QueryUtil.createQuery(currentCodeSet.size())).append(" ]").append("::text[] ");
             builder.append(" && string_to_array(ancestralmaterializedpath, '|') ");
-            QueryUtil.addToPreparedStatement(preparedStmtList, new HashSet<>(boundaryRelationshipSearchCriteria.getCurrentBoundaryCodes()));
+            QueryUtil.addToPreparedStatement(preparedStmtList, currentCodeSet);
         }
 
         return builder.toString();