SK-2258 synchronized errors

Author: skyflow-bharti

v3/src/main/java/com/skyflow/vault/controller/VaultController.java

@@ -77,13 +77,14 @@ public CompletableFuture<com.skyflow.vault.data.InsertResponse> bulkInsertAsync(
             setBearerToken();
             configureInsertConcurrencyAndBatchSize(insertRequest.getValues().size());
             com.skyflow.generated.rest.resources.recordservice.requests.InsertRequest request = super.getBulkInsertRequestBody(insertRequest, super.getVaultConfig());
+            ExecutorService executor = Executors.newFixedThreadPool(insertConcurrencyLimit);
 
-            List<CompletableFuture<com.skyflow.vault.data.InsertResponse>> futures = this.insertBatchFutures(request);
+            List<ErrorRecord> errorRecords = new ArrayList<>();
+            List<CompletableFuture<com.skyflow.vault.data.InsertResponse>> futures = this.insertBatchFutures(request, errorRecords, executor);
 
             return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
                     .thenApply(v -> {
                         List<Success> successRecords = new ArrayList<>();
-                        List<ErrorRecord> errorRecords = new ArrayList<>();
 
                         for (CompletableFuture<com.skyflow.vault.data.InsertResponse> future : futures) {
                             com.skyflow.vault.data.InsertResponse futureResponse = future.join();
@@ -96,6 +97,7 @@ public CompletableFuture<com.skyflow.vault.data.InsertResponse> bulkInsertAsync(
                                 }
                             }
                         }
+                        executor.shutdown(); // Shutdown the executor after all tasks are completed
 
                         return new com.skyflow.vault.data.InsertResponse(successRecords, errorRecords, insertRequest.getValues());
                     });
@@ -113,9 +115,9 @@ private com.skyflow.vault.data.InsertResponse processSync(
         LogUtil.printInfoLog(InfoLogs.PROCESSING_BATCHES.getLog());
         List<ErrorRecord> errorRecords = new ArrayList<>();
         List<Success> successRecords = new ArrayList<>();
+        ExecutorService executor = Executors.newFixedThreadPool(insertConcurrencyLimit);
 
-        List<CompletableFuture<com.skyflow.vault.data.InsertResponse>> futures = this.insertBatchFutures(insertRequest);
-
+        List<CompletableFuture<com.skyflow.vault.data.InsertResponse>> futures = this.insertBatchFutures(insertRequest, errorRecords, executor);
         CompletableFuture<Void> allFutures = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
         allFutures.join();
 
@@ -129,6 +131,7 @@ private com.skyflow.vault.data.InsertResponse processSync(
                     errorRecords.addAll(futureResponse.getErrors());
                 }
             }
+            executor.shutdown(); // Shutdown the executor after all tasks are completed
         }
         com.skyflow.vault.data.InsertResponse response = new com.skyflow.vault.data.InsertResponse(successRecords, errorRecords, originalPayload);
         LogUtil.printInfoLog(InfoLogs.INSERT_REQUEST_RESOLVED.getLog());
@@ -137,11 +140,9 @@ private com.skyflow.vault.data.InsertResponse processSync(
 
 
     private List<CompletableFuture<com.skyflow.vault.data.InsertResponse>> insertBatchFutures(
-            com.skyflow.generated.rest.resources.recordservice.requests.InsertRequest insertRequest
-    ) {
+            com.skyflow.generated.rest.resources.recordservice.requests.InsertRequest insertRequest, List<ErrorRecord> errorRecords, ExecutorService executor) {
         List<InsertRecordData> records = insertRequest.getRecords().get();
 
-        ExecutorService executor = Executors.newFixedThreadPool(insertConcurrencyLimit);
         List<List<InsertRecordData>> batches = Utils.createBatches(records, insertBatchSize);
         List<CompletableFuture<com.skyflow.vault.data.InsertResponse>> futures = new ArrayList<>();
 
@@ -152,7 +153,12 @@ private List<CompletableFuture<com.skyflow.vault.data.InsertResponse>> insertBat
                 CompletableFuture<com.skyflow.vault.data.InsertResponse> future = CompletableFuture
                         .supplyAsync(() -> insertBatch(batch, insertRequest.getTableName().get()), executor)
                         .thenApply(response -> formatResponse(response, batchNumber, insertBatchSize))
-                        .exceptionally(ex -> new com.skyflow.vault.data.InsertResponse(null, handleBatchException(ex, batch, batchNumber, batches)));
+                        .exceptionally(ex -> {
+                            synchronized (errorRecords){
+                                errorRecords.addAll(handleBatchException(ex, batch, batchNumber, batches));
+                            }
+                            return null;
+                        });
                 futures.add(future);
             }
         } finally {