Replace COS service singleton with project-ID cache

docs/content/services/document-processing/text-extraction-service.md

@@ -113,7 +113,7 @@ TextExtractionService service = TextExtractionService.builder()
     .build();
 ```
 
-The resolved endpoint URL and bucket are cached after the first lookup. When only `spaceId` is configured (no `projectId`), lazy COS resolution is not supported; set `cosUrl` explicitly in that case.
+The resolved endpoint URL and bucket are cached per Project ID after the first lookup for that project. When only `spaceId` is configured (no `projectId`), lazy COS resolution is not supported. Set `cosUrl` explicitly in that case.
 
 ### Using a Separate COS Authenticator
 

modules/watsonx-ai/src/main/java/com/ibm/watsonx/ai/textprocessing/CosDataLocation.java

@@ -15,4 +15,15 @@
  * @param bucket Optional override for the bucket name defined in the connection asset. Used only for {@code connection_asset} type.
  * @param path The file path within the container. Used only for {@code container} type.
  */
-public record CosDataLocation(String fileName, String bucket, String path) {}
+public record CosDataLocation(String fileName, String bucket, String path) {
+
+    public CosDataLocation {
+        boolean hasConnectionAssetFields = fileName != null || bucket != null;
+        boolean hasPath = path != null;
+        if (hasConnectionAssetFields == hasPath)
+            throw new IllegalArgumentException(
+                hasPath
+                    ? "fileName/bucket and path are mutually exclusive"
+                    : "Either fileName/bucket or path must be set");
+    }
+}

modules/watsonx-ai/src/main/java/com/ibm/watsonx/ai/textprocessing/storage/cos/CosStorageService.java

@@ -18,7 +18,6 @@
 import com.ibm.watsonx.ai.core.auth.ibmcloud.IBMCloudAuthenticator;
 import com.ibm.watsonx.ai.core.exception.WatsonxException;
 import com.ibm.watsonx.ai.core.exception.model.WatsonxError.Code;
-import com.ibm.watsonx.ai.core.exception.model.WatsonxError.Error;
 import com.ibm.watsonx.ai.core.http.logging.HttpRequestLogger;
 import com.ibm.watsonx.ai.core.http.logging.HttpResponseLogger;
 import com.ibm.watsonx.ai.textprocessing.DeleteFileRequest;
@@ -108,6 +107,7 @@ public boolean deleteFile(String requestId, String fileName) throws FileNotFound
         try {
             return client.deleteFileAsync(DeleteFileRequest.of(requestId, null, fileName)).get();
         } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
             throw new RuntimeException(e);
         } catch (ExecutionException e) {
             if (e.getCause() instanceof WatsonxException ex) {
@@ -352,12 +352,11 @@ public CosStorageService build() {
     private Exception mapIfCosFileNotFound(WatsonxException e) {
         if (e.statusCode() == 404 && e.details().isPresent()) {
             var details = e.details().get();
-            var fileNotFound = details.errors().stream()
+            return details.errors().stream()
                 .filter(error -> error.is(Code.COS_FILE_NOT_FOUND))
                 .findFirst()
-                .map(Error::message)
-                .orElse(e.getMessage());
-            return new FileNotFoundException(fileNotFound);
+                .<Exception>map(error -> new FileNotFoundException(error.message()))
+                .orElse(e);
         }
         return e;
     }

modules/watsonx-ai/src/main/java/com/ibm/watsonx/ai/textprocessing/storage/cos/DefaultRestClient.java

@@ -71,7 +71,10 @@ public boolean upload(String requestTrackingId, InputStream is, String fileName)
 
             var response = syncCosHttpClient.send(builder.build(), BodyHandlers.ofString());
             return response.statusCode() == 200;
-        } catch (IOException | InterruptedException | URISyntaxException e) {
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new RuntimeException(e);
+        } catch (IOException | URISyntaxException e) {
             throw new RuntimeException(e);
         }
     }
@@ -102,7 +105,10 @@ public String readFile(String requestTrackingId, String fileName) throws FileNot
                 builder.header(REQUEST_ID_HEADER, requestTrackingId);
 
             return syncCosHttpClient.send(builder.build(), BodyHandlers.ofString()).body();
-        } catch (IOException | InterruptedException | URISyntaxException e) {
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new RuntimeException(e);
+        } catch (IOException | URISyntaxException e) {
             throw new RuntimeException(e);
         }
     }

modules/watsonx-ai/src/main/java/com/ibm/watsonx/ai/textprocessing/textclassification/TextClassificationService.java

@@ -18,7 +18,7 @@
 import java.time.Duration;
 import java.util.Map;
 import java.util.UUID;
-import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.ConcurrentHashMap;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import com.ibm.watsonx.ai.CloudRegion;
@@ -64,8 +64,7 @@ public class TextClassificationService extends ScopedService {
     private final String cosUrl;
     private final DocumentReference documentReference;
     private final TextClassificationRestClient client;
-    private volatile StorageOperations cosService;
-    private final ReentrantLock cosServiceLock = new ReentrantLock();
+    private final ConcurrentHashMap<String, StorageOperations> cosServiceCache = new ConcurrentHashMap<>();
     private final ProjectService lazyProjectService;
     private final Authenticator authenticator;
     private final Authenticator cosAuthenticator;
@@ -379,8 +378,7 @@ public boolean uploadFile(InputStream inputStream, String fileName) {
         requireNonNull(inputStream, "inputStream value cannot be null");
         requireNonNull(fileName, "fileName value cannot be null");
         var requestId = UUID.randomUUID().toString();
-        upload(requestId, inputStream, fileName, null, false);
-        return true;
+        return upload(requestId, inputStream, fileName, null, false);
     }
 
     /**
@@ -442,81 +440,81 @@ public boolean deleteRequest(String id, TextClassificationDeleteParameters param
         return client.deleteClassification(request);
     }
 
-    // Returns the CosStorageService lazily on first use.
     private StorageOperations getOrResolveCosService() {
-        return getOrResolveCosService(null);
+        return getOrResolveCosService(this.projectId);
     }
 
     private StorageOperations getOrResolveCosService(CosReference cosRef) {
+        return StorageFactory.cos(cosUrl, cosRef.bucket(), authenticator, cosAuthenticator, httpClient,
+            timeout, logRequests, logResponses, requestLogger, requestLogLevel, responseLogger, responseLogLevel, verifySsl);
+    }
 
-        if (cosRef != null) {
-            // Per-call CosReference override.
-            return StorageFactory.cos(cosUrl, cosRef.bucket(), authenticator, cosAuthenticator, httpClient,
-                timeout, logRequests, logResponses, requestLogger, requestLogLevel, responseLogger, responseLogLevel, verifySsl);
-        }
-
-        if (cosService != null)
-            return cosService;
+    private StorageOperations getOrResolveCosService(String effectiveProjectId) {
+        if (effectiveProjectId == null)
+            throw new IllegalStateException(
+                "ContainerReference storage requires a projectId - spaceId alone is not supported for container uploads.");
 
-        cosServiceLock.lock();
+        var cached = cosServiceCache.get(effectiveProjectId);
+        if (cached != null)
+            return cached;
 
+        boolean wasInterrupted = Thread.interrupted();
         try {
+            return cosServiceCache.computeIfAbsent(effectiveProjectId, this::resolveCosServiceForProject);
+        } finally {
+            if (wasInterrupted)
+                Thread.currentThread().interrupt();
+        }
+    }
 
-            if (cosService != null)
-                return cosService;
-
-            if (projectId == null)
-                throw new IllegalStateException(
-                    "ContainerReference storage requires a projectId - spaceId alone is not supported for container uploads.");
-
-            ProjectService ps = lazyProjectService;
-
-            if (ps == null) {
+    private String resolveEffectiveProjectId(TextClassificationParameters parameters) {
+        if (parameters != null && (parameters.projectId() != null || parameters.spaceId() != null))
+            return parameters.projectId();
+        return this.projectId;
+    }
 
-                var region = CloudRegion.fromMlEndpoint(baseUrl).orElseThrow(() -> new IllegalStateException(
-                    "ContainerReference storage requires a ProjectService or a known CloudRegion. "
-                        + "Either pass projectService(ProjectService) on the builder, "
-                        + "or use baseUrl(CloudRegion) so the project storage can be resolved automatically."));
+    private StorageOperations resolveCosServiceForProject(String projectId) {
+        ProjectService ps = lazyProjectService;
 
-                var psBuilder = ProjectService.builder()
-                    .baseUrl(region.wxEndpoint().replace("/wx", ""))
-                    .authenticator(authenticator)
-                    .httpClient(httpClient)
-                    .timeout(timeout)
-                    .logRequests(logRequests)
-                    .logResponses(logResponses)
-                    .verifySsl(verifySsl);
+        if (ps == null) {
 
-                if (requestLogger != null)
-                    psBuilder.logRequests(requestLogger, requestLogLevel);
+            var region = CloudRegion.fromMlEndpoint(baseUrl).orElseThrow(() -> new IllegalStateException(
+                "ContainerReference storage requires a ProjectService or a known CloudRegion. "
+                    + "Either pass projectService(ProjectService) on the builder, "
+                    + "or use baseUrl(CloudRegion) so the project storage can be resolved automatically."));
 
-                if (responseLogger != null)
-                    psBuilder.logResponses(responseLogger, responseLogLevel);
+            var psBuilder = ProjectService.builder()
+                .baseUrl(region.wxEndpoint().replace("/wx", ""))
+                .authenticator(authenticator)
+                .httpClient(httpClient)
+                .timeout(timeout)
+                .logRequests(logRequests)
+                .logResponses(logResponses)
+                .verifySsl(verifySsl);
 
-                ps = psBuilder.build();
-            }
+            if (requestLogger != null)
+                psBuilder.logRequests(requestLogger, requestLogLevel);
 
-            var props = ps.findProject(projectId)
-                .orElseThrow(() -> new IllegalStateException("Project not found: " + projectId))
-                .storage()
-                .properties();
+            if (responseLogger != null)
+                psBuilder.logResponses(responseLogger, responseLogLevel);
 
-            var resolvedUrl = props.endpointUrl().endsWith("/")
-                ? props.endpointUrl().substring(0, props.endpointUrl().length() - 1)
-                : props.endpointUrl();
+            ps = psBuilder.build();
+        }
 
-            var resolvedBucket = props.bucketName();
-            cosService = StorageFactory.cos(resolvedUrl, resolvedBucket, authenticator, cosAuthenticator, httpClient,
-                timeout, logRequests, logResponses, requestLogger, requestLogLevel, responseLogger, responseLogLevel, verifySsl);
+        var props = ps.findProject(projectId)
+            .orElseThrow(() -> new IllegalStateException("Project not found: " + projectId))
+            .storage()
+            .properties();
 
-            return cosService;
+        var resolvedUrl = props.endpointUrl().endsWith("/")
+            ? props.endpointUrl().substring(0, props.endpointUrl().length() - 1)
+            : props.endpointUrl();
 
-        } finally {
-            cosServiceLock.unlock();
-        }
+        var resolvedBucket = props.bucketName();
+        return StorageFactory.cos(resolvedUrl, resolvedBucket, authenticator, cosAuthenticator, httpClient,
+            timeout, logRequests, logResponses, requestLogger, requestLogLevel, responseLogger, responseLogLevel, verifySsl);
     }
 
-    // Retrieves the classification result for the given request id.
     private TextClassificationResponse fetchClassificationRequest(String requestId, String id, TextClassificationFetchParameters parameters) {
         requireNonNull(requestId, "The requestId can not be null");
         requireNonNull(id, "The id can not be null");
@@ -550,13 +548,12 @@ private ClassificationResult classifyAndFetch(String requestId, String absoluteP
                 DocumentReference effectiveDoc = parameters.documentReference() != null
                     ? parameters.documentReference()
                     : this.documentReference;
-                cleanUpUploadedFile(requestId, uploadedPath, effectiveDoc);
+                cleanUpUploadedFile(requestId, uploadedPath, effectiveDoc, resolveEffectiveProjectId(parameters));
             }
         }
     }
 
-    // Uploads an input stream to COS or the container.
-    private void upload(String requestId, InputStream is, String fileName, TextClassificationParameters parameters,
+    private boolean upload(String requestId, InputStream is, String fileName, TextClassificationParameters parameters,
         boolean waitForClassification) {
         requireNonNull(requestId, "requestId value cannot be null");
         requireNonNull(is, "is value cannot be null");
@@ -573,10 +570,8 @@ private void upload(String requestId, InputStream is, String fileName, TextClass
             this.documentReference
         );
 
-        if (effectiveDoc instanceof ContainerReference) {
-            getOrResolveCosService().upload(requestId, is, fileName);
-            return;
-        }
+        if (effectiveDoc instanceof ContainerReference)
+            return getOrResolveCosService(resolveEffectiveProjectId(parameters)).upload(requestId, is, fileName);
 
         if (!(effectiveDoc instanceof CosReference cosRef))
             throw new UnsupportedOperationException(
@@ -586,7 +581,7 @@ private void upload(String requestId, InputStream is, String fileName, TextClass
             throw new IllegalStateException(
                 "cosUrl must be set on the service builder to perform COS upload operations.");
 
-        getOrResolveCosService(cosRef).upload(requestId, is, fileName);
+        return getOrResolveCosService(cosRef).upload(requestId, is, fileName);
     }
 
     // Starts the text classification process.
@@ -705,30 +700,21 @@ private void cleanUpAfterAbortedClassification(String processId, String projectI
         }
     }
 
-    // Deletes the uploaded file from COS or the container. All errors are logged and swallowed.
-    private void cleanUpUploadedFile(String requestId, String path, DocumentReference effectiveDoc) {
+    private void cleanUpUploadedFile(String requestId, String path, DocumentReference effectiveDoc, String effectiveProjectId) {
         try {
             if (effectiveDoc instanceof CosReference cosRef)
                 client.deleteFileAsync(DeleteFileRequest.of(requestId, cosRef.bucket(), path))
                     .exceptionally(ex -> {
                         logger.warn("Async COS delete failed for {}: {}", path, ex.getMessage());
                         return false;
                     });
-            else if (effectiveDoc instanceof ContainerReference) {
-                boolean wasInterrupted = Thread.interrupted();
-                try {
-                    getOrResolveCosService().deleteFileAsync(requestId, path);
-                } finally {
-                    if (wasInterrupted)
-                        Thread.currentThread().interrupt();
-                }
-            }
+            else if (effectiveDoc instanceof ContainerReference)
+                getOrResolveCosService(effectiveProjectId).deleteFileAsync(requestId, path);
         } catch (Exception e) {
             logger.warn("Failed to delete uploaded file {}: {}", path, e.getMessage(), e);
         }
     }
 
-    // Extracts the ClassificationResult from a completed-or-failed response.
     private ClassificationResult getClassificationResult(TextClassificationResponse textClassificationResponse)
         throws TextClassificationException {
 
@@ -834,6 +820,16 @@ public Builder documentReference(ContainerReference documentReference) {
             return this;
         }
 
+        /**
+         * Specifies the reference to the input document, whether a {@link CosReference} or a {@link ContainerReference}.
+         *
+         * @param documentReference the {@link DocumentReference} for the input file.
+         */
+        public Builder documentReference(DocumentReference documentReference) {
+            this.documentReference = documentReference;
+            return this;
+        }
+
         /**
          * Specifies the Cloud Object Storage connection and bucket where the input files are stored.
          *