Fix NPE in DBFSVolumeClient when connectionContext is null

Existing tests create DBFSVolumeClient with a test constructor that sets
connectionContext to null. Guard getCustomHeaders() calls with null check
to avoid NPE in test paths while preserving SPOG header propagation in
production paths.

Co-authored-by: Isaac
Signed-off-by: Madhavendra Rathore <madhavendra.rathore@databricks.com>

Author: msrathore-db

src/main/java/com/databricks/jdbc/api/impl/volume/DBFSVolumeClient.java

@@ -492,7 +492,8 @@ CreateUploadUrlResponse getCreateUploadUrlResponse(String objectPath)
     CreateUploadUrlRequest request = new CreateUploadUrlRequest(objectPath);
     try {
       Request req = new Request(Request.POST, CREATE_UPLOAD_URL_PATH, apiClient.serialize(request));
-      req.withHeaders(JSON_HTTP_HEADERS).withHeaders(connectionContext.getCustomHeaders());
+      req.withHeaders(JSON_HTTP_HEADERS)
+          .withHeaders(connectionContext != null ? connectionContext.getCustomHeaders() : Map.of());
       return apiClient.execute(req, CreateUploadUrlResponse.class);
     } catch (IOException | DatabricksException e) {
       String errorMessage =
@@ -514,7 +515,8 @@ CreateDownloadUrlResponse getCreateDownloadUrlResponse(String objectPath)
     try {
       Request req =
           new Request(Request.POST, CREATE_DOWNLOAD_URL_PATH, apiClient.serialize(request));
-      req.withHeaders(JSON_HTTP_HEADERS).withHeaders(connectionContext.getCustomHeaders());
+      req.withHeaders(JSON_HTTP_HEADERS)
+          .withHeaders(connectionContext != null ? connectionContext.getCustomHeaders() : Map.of());
       return apiClient.execute(req, CreateDownloadUrlResponse.class);
     } catch (IOException | DatabricksException e) {
       String errorMessage =
@@ -534,7 +536,8 @@ CreateDeleteUrlResponse getCreateDeleteUrlResponse(String objectPath)
 
     try {
       Request req = new Request(Request.POST, CREATE_DELETE_URL_PATH, apiClient.serialize(request));
-      req.withHeaders(JSON_HTTP_HEADERS).withHeaders(connectionContext.getCustomHeaders());
+      req.withHeaders(JSON_HTTP_HEADERS)
+          .withHeaders(connectionContext != null ? connectionContext.getCustomHeaders() : Map.of());
       return apiClient.execute(req, CreateDeleteUrlResponse.class);
     } catch (IOException | DatabricksException e) {
       String errorMessage =
@@ -551,7 +554,8 @@ ListResponse getListResponse(String listPath) throws DatabricksVolumeOperationEx
     ListRequest request = new ListRequest(listPath);
     try {
       Request req = new Request(Request.GET, LIST_PATH);
-      req.withHeaders(JSON_HTTP_HEADERS).withHeaders(connectionContext.getCustomHeaders());
+      req.withHeaders(JSON_HTTP_HEADERS)
+          .withHeaders(connectionContext != null ? connectionContext.getCustomHeaders() : Map.of());
       ApiClient.setQuery(req, request);
       return apiClient.execute(req, ListResponse.class);
     } catch (IOException | DatabricksException e) {
@@ -888,7 +892,9 @@ private CompletableFuture<CreateUploadUrlResponse> requestPresignedUrlWithRetry(
         Map<String, String> authHeaders = workspaceClient.config().authenticate();
         authHeaders.forEach(requestBuilder::addHeader);
         JSON_HTTP_HEADERS.forEach(requestBuilder::addHeader);
-        connectionContext.getCustomHeaders().forEach(requestBuilder::addHeader);
+        if (connectionContext != null) {
+          connectionContext.getCustomHeaders().forEach(requestBuilder::addHeader);
+        }
 
         requestBuilder.setEntity(
             AsyncEntityProducers.create(requestBody.getBytes(), ContentType.APPLICATION_JSON));