SK-3061: fix bulkInsert(sync) interceptor-exception leak

processBulkInsertSync called insertBatchFutures - which synchronously
invokes the caller's RequestInterceptor - before entering its own try
block, so a throwing interceptor escaped bulkInsert() as a raw,
undeclared exception instead of the documented SkyflowException.

processBulkDetokenizeSync/processBulkDeleteTokensSync/processBulkTokenizeSync
already call their own *BatchFutures inside their own try/catch(Exception),
which is why they were not exploitable the same way. This moves
insertBatchFutures's call inside processBulkInsertSync's existing try,
matching that same structure, instead of adding a new catch-all to the
4 public sync methods (the broader fix reverted here).

Added a regression test asserting a throwing interceptor surfaces as
SkyflowException from bulkInsert. Confirmed via TDD: failed before this
change (raw IllegalStateException), passes after.

mvn -pl common,flowvault -am test -> 709 (flowvault) + 106
(common/ValidationsTests) tests, 0 failures, BUILD SUCCESS.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: Devesh-Skyflow

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

@@ -778,9 +778,9 @@ private BulkInsertResponse processBulkInsertSync(
     ) throws ExecutionException, InterruptedException, SkyflowException {
         LogUtil.printInfoLog(InfoLogs.PROCESSING_BATCHES.getLog());
         List<BulkInsertResponseRecord> records = new ArrayList<>();
-        List<CompletableFuture<BulkInsertResponse>> futures = this.insertBatchFutures(insertRequest, interceptor, cfg);
 
         try {
+            List<CompletableFuture<BulkInsertResponse>> futures = this.insertBatchFutures(insertRequest, interceptor, cfg);
             CompletableFuture<Void> allFutures = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
             try {
                 allFutures.join();

flowvault/src/test/java/com/skyflow/vault/controller/VaultControllerTests.java

@@ -1164,4 +1164,30 @@ public void testBulkTokenize_interceptorInvokedOncePerBatchWithDistinctContext()
         Mockito.verify(mockRaw, Mockito.times(EXPECTED_BATCH_COUNT)).tokenize(any(), captor.capture());
         assertInterceptorRanOncePerBatch(interceptor, captor.getAllValues());
     }
+
+    @Test
+    public void testBulkInsert_throwingInterceptorWrappedAsSkyflowException() throws Exception {
+        ApiClient mockApi = Mockito.mock(ApiClient.class);
+        RawFlowserviceClient mockRaw = mockRawFlowservice(mockApi);
+        stubInsertEcho(mockRaw);
+        VaultController controller = createControllerWithMock(mockApi);
+
+        Map<String, Object> data = new HashMap<>();
+        data.put("name", "john");
+        ArrayList<InsertRequestRecord> records = new ArrayList<>();
+        records.add(BulkInsertRequestRecord.builder().tableName("table1").data(data).build());
+        BulkInsertRequest request = BulkInsertRequest.builder().records(records).build();
+
+        RequestInterceptor interceptor = ctx -> {
+            throw new IllegalStateException("sync insert interceptor blew up");
+        };
+        BulkInsertOptions options = BulkInsertOptions.builder().interceptor(interceptor).build();
+
+        try {
+            controller.bulkInsert(request, options);
+            Assert.fail(EXCEPTION_NOT_THROWN);
+        } catch (SkyflowException e) {
+            Assert.assertEquals("sync insert interceptor blew up", e.getMessage());
+        }
+    }
 }