Merge pull request #610 from heremaps/v1_int

hirenkp2000 · web-flow · commit 893fcbc76cbe · 2026-07-06T13:10:26.000+05:30
Rebase with v1_int
diff --git a/.github/workflows/reusable-build-and-publish.yml b/.github/workflows/reusable-build-and-publish.yml
@@ -48,7 +48,7 @@ jobs:
           cp -p $SERVICE_JAR_DIR/xyz-hub-service.jar $CODEDEPLOY_DIR/contents/xyz-hub/
       - name: List CodeDeploy artifact content
         run: |
-          ls -lR $CODEDEPLOY_DIR
+          ls -lRa $CODEDEPLOY_DIR
       - name: Save pipeline artifact name
         id: save-artifact-name
         run: echo "name=${{ env.GITHUB_CODEDEPLOY_ARTIFACT_NAME }}" >> "$GITHUB_OUTPUT"
@@ -58,6 +58,7 @@ jobs:
           name: ${{ env.GITHUB_CODEDEPLOY_ARTIFACT_NAME }}
           path: ${{ env.CODEDEPLOY_DIR }}
           if-no-files-found: error
+          include-hidden-files: true
       - name: Identify and save Application version
         id: save-app-version
         run: |
diff --git a/.github/workflows/reusable-s3-codedeploy-release.yml b/.github/workflows/reusable-s3-codedeploy-release.yml
@@ -49,7 +49,7 @@ jobs:
           path: download
       - name: List CodeDeploy artifacts
         run: |
-          ls -lR download/
+          ls -lRa download/
       - name: Preparing S3 artifact name
         id: s3-artifact-name
         run: |
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,6 +1,18 @@
+## Naksha_1.1.5
+
+- Improved DataSource cache refresh handling in txn-handler submodule to gracefully close active DB connections
+
+## Naksha_1.1.4
+
+- Increased header size limit to 24k
+
+## Naksha_1.1.3
+
+- Increased DB Pool size of Naksha Admin DB from 10 to 25 to allow additional sequencer/publisher jobs to run in parallel.
+
 ## Naksha_1.1.2
 
-- Increased header size limit to 24K
+- Fixed batch-size failure issue for SNS publishing where one of the message fails during publish, then we optimize by continuing from that failure instead of re-processing entire batch 
 
 ## Naksha_1.1.1
 
diff --git a/pom.xml b/pom.xml
@@ -41,7 +41,7 @@
 
   <groupId>com.here.xyz</groupId>
   <artifactId>xyz-hub</artifactId>
-  <version>1.1.2</version>
+  <version>1.1.5</version>
   <packaging>pom</packaging>
 
   <modules>
diff --git a/xyz-connectors/pom.xml b/xyz-connectors/pom.xml
@@ -25,7 +25,7 @@
     <groupId>com.here.xyz</groupId>
     <artifactId>xyz-hub</artifactId>
     <relativePath>../</relativePath>
-    <version>1.1.2</version>
+    <version>1.1.5</version>
   </parent>
 
   <licenses>
diff --git a/xyz-hub-service/pom.xml b/xyz-hub-service/pom.xml
@@ -25,7 +25,7 @@
     <groupId>com.here.xyz</groupId>
     <artifactId>xyz-hub</artifactId>
     <relativePath>../pom.xml</relativePath>
-    <version>1.1.2</version>
+    <version>1.1.5</version>
   </parent>
 
   <licenses>
diff --git a/xyz-hub-test/pom.xml b/xyz-hub-test/pom.xml
@@ -25,7 +25,7 @@
     <groupId>com.here.xyz</groupId>
     <artifactId>xyz-hub</artifactId>
     <relativePath>../</relativePath>
-    <version>1.1.2</version>
+    <version>1.1.5</version>
   </parent>
 
   <licenses>
diff --git a/xyz-models/pom.xml b/xyz-models/pom.xml
@@ -25,7 +25,7 @@
     <groupId>com.here.xyz</groupId>
     <artifactId>xyz-hub</artifactId>
     <relativePath>../</relativePath>
-    <version>1.1.2</version>
+    <version>1.1.5</version>
   </parent>
 
   <licenses>
diff --git a/xyz-psql-connector/pom.xml b/xyz-psql-connector/pom.xml
@@ -25,7 +25,7 @@
     <groupId>com.here.xyz</groupId>
     <artifactId>xyz-hub</artifactId>
     <relativePath>../</relativePath>
-    <version>1.1.2</version>
+    <version>1.1.5</version>
   </parent>
 
   <licenses>
diff --git a/xyz-txn-handler/pom.xml b/xyz-txn-handler/pom.xml
@@ -25,7 +25,7 @@
         <groupId>com.here.xyz</groupId>
         <artifactId>xyz-hub</artifactId>
         <relativePath>../</relativePath>
-        <version>1.1.2</version>
+        <version>1.1.5</version>
     </parent>
 
     <licenses>
diff --git a/xyz-txn-handler/src/main/java/com/here/xyz/pub/XYZTransactionHandler.java b/xyz-txn-handler/src/main/java/com/here/xyz/pub/XYZTransactionHandler.java
@@ -57,6 +57,7 @@ private static void readConfig() {
         adminDBConnParams.setDbUrl(pubCfg.STORAGE_DB_URL);
         adminDBConnParams.setUser(pubCfg.STORAGE_DB_USER);
         adminDBConnParams.setPswd(pubCfg.STORAGE_DB_PASSWORD);
+        adminDBConnParams.setMaxPoolSize(25); // keeping higher value to allow parallel seq/publisher jobs
         // Set AWS account access details
         System.setProperty("aws.accessKeyId", (pubCfg.AWS_ACCESS_KEY_ID!=null) ? pubCfg.AWS_ACCESS_KEY_ID : "");
         System.setProperty("aws.secretAccessKey", (pubCfg.AWS_SECRET_ACCESS_KEY!=null) ? pubCfg.AWS_SECRET_ACCESS_KEY : "");
diff --git a/xyz-txn-handler/src/main/java/com/here/xyz/pub/db/PubJdbcConnectionPool.java b/xyz-txn-handler/src/main/java/com/here/xyz/pub/db/PubJdbcConnectionPool.java
@@ -15,8 +15,8 @@
 public class PubJdbcConnectionPool {
     private static final Logger logger = LogManager.getLogger();
 
-    // Entire cache will be flushed every x hours (e.g. 8hrs) to prevent pile-up of stale data
-    final private static int dsCacheExpiryInMins = 8*60;
+    // Entire cache will be flushed every x hours (e.g. 1hrs) to prevent pile-up of stale data
+    final private static int dsCacheExpiryInMins = 1 * 60;
     private static long dsCacheExpiryEpochMs = System.currentTimeMillis() + TimeUnit.MINUTES.toMillis(dsCacheExpiryInMins);
     private static ConcurrentHashMap<JdbcConnectionParams, HikariDataSource> dsCache = new ConcurrentHashMap<>();
 
@@ -51,14 +51,16 @@ private synchronized static void flushDataSourceCache(final String spaceId) {
         for (JdbcConnectionParams connParams : dsCache.keySet()) {
             flushThisDS = (spaceId == null || spaceId.equals(connParams.getSpaceId()));
             if (flushThisDS) {
-                dsCache.get(connParams).close(); // close datasource
-                dsCache.remove(connParams); // remove datasource from cache
+                final HikariDataSource oldDs = dsCache.remove(connParams); // remove old datasource from cache
+                if (oldDs != null) {
+                    oldDs.close(); // close old datasource gracefully
+                }
                 logger.info("Removed old cached DataSource for spaceId {}.", connParams.getSpaceId());
             }
         }
         // reset cache (if it is full cache flush)
         if (spaceId == null) {
-            dsCache = new ConcurrentHashMap<>();
+            dsCache.clear();
             dsCacheExpiryEpochMs = System.currentTimeMillis() + TimeUnit.MINUTES.toMillis(dsCacheExpiryInMins);
             logger.info("Recreated DataSource Cache with expiry of {}mins.", dsCacheExpiryInMins);
         }
@@ -74,7 +76,7 @@ private static synchronized HikariDataSource createAndCacheDataSource(final Jdbc
         // Create new Datasource (connection pool) and add it to a cache
         final HikariConfig config = new HikariConfig();
         String dbUrl = dbConnParams.getDbUrl();
-        dbUrl += (dbUrl.contains("?") ? "&" : "?") + "ApplicationName=XYZ-Hub-Publisher";
+        dbUrl += (dbUrl.contains("?") ? "&" : "?") + "ApplicationName=XYZ-Hub-Publisher_"+dbConnParams.getSpaceId();
         config.setDriverClassName(dbConnParams.getDriveClass());
         config.setJdbcUrl(dbUrl);
         config.setUsername(dbConnParams.getUser());
diff --git a/xyz-txn-handler/src/main/java/com/here/xyz/pub/handlers/PubTxnListHandler.java b/xyz-txn-handler/src/main/java/com/here/xyz/pub/handlers/PubTxnListHandler.java
@@ -46,10 +46,13 @@ public void run() {
                 return;
             }
             // Handover transactions to appropriate Publisher (e.g. DefaultSNSPublisher)
-            final PublishEntryDTO lastTxn = PubUtil.getPubInstance(sub).publishTransactions(pubCfg, sub, txnList, lastTxnId, lastTxnRecId);
-
-            // Update last txn_id in AdminDB::xyz_config::xyz_txn_pub table
-            PubDatabaseHandler.saveLastTxnId(adminDBConnParams, subId, lastTxn);
+            final PublishEntryDTO lastTxn = new PublishEntryDTO(lastTxnId, lastTxnRecId);
+            try {
+                PubUtil.getPubInstance(sub).publishTransactions(pubCfg, sub, txnList, lastTxn);
+            } finally {
+                // Update last txn_id in AdminDB::xyz_config::xyz_txn_pub table
+                PubDatabaseHandler.saveLastTxnId(adminDBConnParams, subId, lastTxn);
+            }
         } catch (Exception ex) {
             logger.error("{} - Exception in publisher job for subId={}, spaceId={}. ",
                     PubLogConstants.LOG_CODE_PUBLISH_ERROR, subId, spaceId, ex);
diff --git a/xyz-txn-handler/src/main/java/com/here/xyz/pub/impl/DefaultSNSBatchPublisher.java b/xyz-txn-handler/src/main/java/com/here/xyz/pub/impl/DefaultSNSBatchPublisher.java
@@ -26,15 +26,14 @@ public class DefaultSNSBatchPublisher implements IPublisher {
 
     // Convert and publish transactions to desired SNS Topic
     @Override
-    public PublishEntryDTO publishTransactions(final PubConfig pubCfg, final Subscription sub,
+    public void publishTransactions(final PubConfig pubCfg, final Subscription sub,
                                                final List<PubTransactionData> txnList,
-                                               final long lastStoredTxnId, final long lastStoredTxnRecId) throws Exception {
+                                               final PublishEntryDTO pubDTO) throws Exception {
         final String subId = sub.getId();
         final String spaceId = sub.getSource();
         final String snsTopic = PubUtil.getSnsTopicARN(sub);
         final long lotStartTS = System.currentTimeMillis();
         // local counters
-        final PublishEntryDTO pubDTO = new PublishEntryDTO(lastStoredTxnId, lastStoredTxnRecId);
         int publishedRecCnt = 0;
         // Variables for batch publish
         final int TXN_LIST_SIZE = txnList.size();
@@ -64,15 +63,8 @@ public PublishEntryDTO publishTransactions(final PubConfig pubCfg, final Subscri
                 final int msgLength = msg.length();
                 final Map<String, MessageAttributeValue> msgAttrMap = populateMessageAttributeMap(txnData, sub, spaceId);
 
-                // Prepare PublishBatchEntry for current message
-                final PublishBatchRequestEntry batchEntry = PublishBatchRequestEntry.builder()
-                        .message(msg)
-                        .messageAttributes(msgAttrMap)
-                        .id(MSG_ID_PREFIX + batchEntryCounter)
-                        .build();
-
-                // publish batch if payload limit reached
-                if (msgLength+aggrBatchPayloadSize > MAX_ALLOWED_PAYLOAD_SIZE && batchEntries.size()>0) {
+                // publish accumulated batch if payload limit reached
+                if (msgLength+aggrBatchPayloadSize > MAX_ALLOWED_PAYLOAD_SIZE && !batchEntries.isEmpty()) {
                     // publish current batch
                     publishBatchEntriesAndCheckResult(batchEntries, snsTopic, snsClient, txnList, subId, publishedRecCnt, MSG_ID_PREFIX, pubCfg, pubDTO);
                     // update batch variables
@@ -82,6 +74,12 @@ public PublishEntryDTO publishTransactions(final PubConfig pubCfg, final Subscri
                     batchEntryCounter = 0;
                 }
 
+                // Prepare PublishBatchEntry for current message
+                final PublishBatchRequestEntry batchEntry = PublishBatchRequestEntry.builder()
+                        .message(msg)
+                        .messageAttributes(msgAttrMap)
+                        .id(MSG_ID_PREFIX + batchEntryCounter)
+                        .build();
                 // add current message to the batch
                 batchEntries.add(batchEntry);
                 aggrBatchPayloadSize += msgLength;
@@ -106,8 +104,6 @@ public PublishEntryDTO publishTransactions(final PubConfig pubCfg, final Subscri
             logger.info("Transaction publish stats for SNS [{}] [format => eventType,subId,spaceId,msgCount,timeTakenMs,lastTxnId,lastTxnRecId] - {} {} {} {} {} {} {}",
                     snsTopic, PubLogConstants.LOG_EVT_TXN_PUBLISH_STATS, subId, spaceId, publishedRecCnt, lotTimeTaken, pubDTO.getLastTxnId(), pubDTO.getLastTxnRecId());
         }
-
-        return pubDTO;
     }
 
 
diff --git a/xyz-txn-handler/src/main/java/com/here/xyz/pub/impl/DefaultSNSSinglePublisher.java b/xyz-txn-handler/src/main/java/com/here/xyz/pub/impl/DefaultSNSSinglePublisher.java
@@ -25,15 +25,14 @@ public class DefaultSNSSinglePublisher implements IPublisher {
 
     // Convert and publish transactions to desired SNS Topic
     @Override
-    public PublishEntryDTO publishTransactions(final PubConfig pubCfg, final Subscription sub,
+    public void publishTransactions(final PubConfig pubCfg, final Subscription sub,
                                                final List<PubTransactionData> txnList,
-                                               final long lastStoredTxnId, final long lastStoredTxnRecId) throws Exception {
+                                               final PublishEntryDTO pubDTO) throws Exception {
         final String subId = sub.getId();
         final String spaceId = sub.getSource();
         final String snsTopic = PubUtil.getSnsTopicARN(sub);
         final long lotStartTS = System.currentTimeMillis();
         // local counters
-        final PublishEntryDTO pubDTO = new PublishEntryDTO(lastStoredTxnId, lastStoredTxnRecId);
         long publishedRecCnt = 0;
 
         try {
@@ -82,8 +81,6 @@ public PublishEntryDTO publishTransactions(final PubConfig pubCfg, final Subscri
             logger.info("Transaction publish stats for SNS [{}] [format => eventType,subId,spaceId,msgCount,timeTakenMs,lastTxnId,lastTxnRecId] - {} {} {} {} {} {} {}",
                     snsTopic, PubLogConstants.LOG_EVT_TXN_PUBLISH_STATS, subId, spaceId, publishedRecCnt, lotTimeTaken, pubDTO.getLastTxnId(), pubDTO.getLastTxnRecId());
         }
-
-        return pubDTO;
     }
 
 
diff --git a/xyz-txn-handler/src/main/java/com/here/xyz/pub/impl/IPublisher.java b/xyz-txn-handler/src/main/java/com/here/xyz/pub/impl/IPublisher.java
@@ -8,8 +8,8 @@
 import java.util.List;
 
 public interface IPublisher {
-    PublishEntryDTO publishTransactions(final PubConfig pubCfg, final Subscription sub,
+    void publishTransactions(final PubConfig pubCfg, final Subscription sub,
                                         final List<PubTransactionData> txnList,
-                                        final long lastTxnId, final long lastTxnRecId) throws Exception;
+                                        final PublishEntryDTO pubDTO) throws Exception;
 
 }
