streamObservable) {
- fileStreamService.streamToClient(request, streamObservable);
+ fileStreamService.streamToClient(request, streamObservable, THREADED_FILE_STREAM_SERVICE_EXECUTOR);
+ }
+
+ @Override
+ public void close() {
+ fileStreamService.close();
+ kafkaStreamService.close();
+ THREADED_FILE_STREAM_SERVICE_EXECUTOR.shutdown();
+ THREADED_KAFKA_STREAM_SERVICE_EXECUTOR.shutdown();
}
}
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/client/grpc/file/FileChunkAssembler.java b/src/main/java/uk/gov/dbt/ndtp/federator/client/grpc/file/FileChunkAssembler.java
index b5495a83..8dfeeb4f 100644
--- a/src/main/java/uk/gov/dbt/ndtp/federator/client/grpc/file/FileChunkAssembler.java
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/client/grpc/file/FileChunkAssembler.java
@@ -26,6 +26,7 @@
import uk.gov.dbt.ndtp.federator.client.storage.ReceivedFileStorage;
import uk.gov.dbt.ndtp.federator.client.storage.ReceivedFileStorageFactory;
import uk.gov.dbt.ndtp.federator.client.storage.StoredFileResult;
+import uk.gov.dbt.ndtp.federator.client.storage.impl.GCPReceivedFileStorage;
import uk.gov.dbt.ndtp.federator.client.storage.impl.S3ReceivedFileStorage;
import uk.gov.dbt.ndtp.federator.common.utils.GRPCUtils;
import uk.gov.dbt.ndtp.federator.common.utils.PropertyUtil;
@@ -176,18 +177,20 @@ private Path handleLastChunk(FileChunk chunk, String fileName, String key, Assem
Path finalTarget = moveToFinalTarget(state, fileName);
- // Delegate storage (LOCAL or S3) based on configuration
+ // Delegate storage (LOCAL, S3, AZURE, or GCP) based on configuration
ReceivedFileStorage storage = ReceivedFileStorageFactory.get();
StoredFileResult storeResult = storage.store(finalTarget, fileName, destination);
storeResult.remoteUriOpt().ifPresent(uri -> log.info("Remote location: {}", uri));
- // If provider is S3 and remote URI is absent, treat as failure: do NOT signal completion to caller
- if (storage instanceof S3ReceivedFileStorage
+ // If provider is S3/GCP and remote URI is absent, treat as failure: do NOT signal completion to caller
+ if ((storage instanceof S3ReceivedFileStorage || storage instanceof GCPReceivedFileStorage)
&& storeResult.remoteUriOpt().isEmpty()) {
assemblies.remove(key);
Path failedPath = storeResult.localPath().toAbsolutePath();
+ String providerName = storage instanceof S3ReceivedFileStorage ? "S3" : "GCP";
log.info(
- "S3 upload failed for file '{}'; local temp at '{}' may be removed by provider. Will not update Redis offset.",
+ "{} upload failed for file '{}'; local temp at '{}' may be removed by provider. Will not update Redis offset.",
+ providerName,
fileName,
failedPath);
return null; // signal to GRPCFileClient that offset must NOT be advanced
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/client/storage/ReceivedFileStorageFactory.java b/src/main/java/uk/gov/dbt/ndtp/federator/client/storage/ReceivedFileStorageFactory.java
index a812c0fb..241b8bb1 100644
--- a/src/main/java/uk/gov/dbt/ndtp/federator/client/storage/ReceivedFileStorageFactory.java
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/client/storage/ReceivedFileStorageFactory.java
@@ -2,20 +2,21 @@
import lombok.extern.slf4j.Slf4j;
import uk.gov.dbt.ndtp.federator.client.storage.impl.AzureReceivedFileStorage;
+import uk.gov.dbt.ndtp.federator.client.storage.impl.GCPReceivedFileStorage;
import uk.gov.dbt.ndtp.federator.client.storage.impl.LocalReceivedFileStorage;
import uk.gov.dbt.ndtp.federator.client.storage.impl.S3ReceivedFileStorage;
import uk.gov.dbt.ndtp.federator.common.utils.PropertyUtil;
/**
* Factory for selecting a {@link uk.gov.dbt.ndtp.federator.client.storage.ReceivedFileStorage}
- * implementation based on {@code client.files.storage.provider} (LOCAL | S3 | AZURE).
+ * implementation based on {@code client.files.storage.provider} (LOCAL | S3 | AZURE | GCP).
*
* Defaults to LOCAL when the property is missing or has an unknown value.
*/
@Slf4j
public final class ReceivedFileStorageFactory {
- private static final String STORAGE_PROVIDER_PROP = "client.files.storage.provider"; // LOCAL | S3 | AZURE
+ private static final String STORAGE_PROVIDER_PROP = "client.files.storage.provider"; // LOCAL | S3 | AZURE | GCP
private ReceivedFileStorageFactory() {}
@@ -26,6 +27,7 @@ private ReceivedFileStorageFactory() {}
*
* - {@code S3} – returns {@link S3ReceivedFileStorage}
* - {@code AZURE} – returns {@link AzureReceivedFileStorage}
+ * - {@code GCP} – returns {@link GCPReceivedFileStorage}
* - {@code LOCAL} or any other value – returns {@link LocalReceivedFileStorage}
*
*
@@ -46,6 +48,9 @@ public static ReceivedFileStorage get() {
if ("AZURE".equalsIgnoreCase(provider)) {
return new AzureReceivedFileStorage();
}
+ if ("GCP".equalsIgnoreCase(provider)) {
+ return new GCPReceivedFileStorage();
+ }
// Default to LOCAL for unknown values as well
return new LocalReceivedFileStorage();
}
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/client/storage/impl/GCPReceivedFileStorage.java b/src/main/java/uk/gov/dbt/ndtp/federator/client/storage/impl/GCPReceivedFileStorage.java
new file mode 100644
index 00000000..2d6c3ae9
--- /dev/null
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/client/storage/impl/GCPReceivedFileStorage.java
@@ -0,0 +1,96 @@
+package uk.gov.dbt.ndtp.federator.client.storage.impl;
+
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import java.nio.file.Path;
+import lombok.extern.slf4j.Slf4j;
+import uk.gov.dbt.ndtp.federator.client.storage.ReceivedFileStorage;
+import uk.gov.dbt.ndtp.federator.client.storage.StoredFileResult;
+import uk.gov.dbt.ndtp.federator.common.storage.provider.file.client.GcsClientFactory;
+import uk.gov.dbt.ndtp.federator.common.utils.PropertyUtil;
+
+/**
+ * Stores assembled files to Google Cloud Storage using the shared {@link uk.gov.dbt.ndtp.federator.common.storage.provider.file.client.GcsClientFactory}.
+ *
+ * Configuration is sourced from properties (shared by client and server):
+ *
+ * - {@code files.gcp.bucket} – target bucket (required)
+ *
+ * GCP client credentials and endpoint are read by {@code GcsClientFactory} via properties:
+ * {@code gcp.storage.project.id}, {@code gcp.storage.credentials.file}, and optional {@code gcp.storage.endpoint.url}.
+ */
+@Slf4j
+public class GCPReceivedFileStorage implements ReceivedFileStorage {
+
+ /**
+ * Shared property key for target GCS bucket used by both client and server components.
+ */
+ private static final String GCP_BUCKET_PROP = "files.gcp.bucket";
+
+ /**
+ * Uploads the assembled file to GCS (if bucket is configured) and returns the result.
+ *
+ * @param localFile absolute path of the assembled file on the local filesystem
+ * @param originalFileName original file name from the stream (used to form the GCS object key)
+ * @param destination destination or prefix used to build the object key
+ * @return {@link StoredFileResult} containing the local path and the GCS URI if upload succeeded
+ */
+ @Override
+ public StoredFileResult store(Path localFile, String originalFileName, String destination) {
+ String bucket = resolveBucket();
+ if (bucket.isBlank()) {
+ log.warn("Storage provider is GCP but bucket is not provided. Skipping upload.");
+ return new StoredFileResult(localFile.toAbsolutePath(), null);
+ }
+
+ String key = ReceivedFileStorage.super.resolveKey(destination, originalFileName);
+ try {
+ var uri = upload(localFile, bucket, key);
+ if (uri != null) {
+ // Success path: we manage local temp cleanup here to satisfy tests
+ ReceivedFileStorage.super.deleteLocalTempQuietly(localFile);
+ return new StoredFileResult(localFile.toAbsolutePath(), uri);
+ }
+ // upload() may return null (and may have already attempted deletion). Ensure it's deleted.
+ ReceivedFileStorage.super.deleteLocalTempQuietly(localFile);
+ return new StoredFileResult(localFile.toAbsolutePath(), null);
+ } catch (Exception e) {
+ // If an overriding implementation of upload() throws, we must still clean up and return gracefully
+ log.error(
+ "Upload threw an exception; deleting temp file {} and returning without remote URI", localFile, e);
+ ReceivedFileStorage.super.deleteLocalTempQuietly(localFile);
+ return new StoredFileResult(localFile.toAbsolutePath(), null);
+ }
+ }
+
+ // -------- Helper methods (extracted for testability) --------
+
+ String resolveBucket() {
+ String bucket = PropertyUtil.getPropertyValue(GCP_BUCKET_PROP, "");
+ return bucket == null ? "" : bucket;
+ }
+
+ // Use default key resolution from interface
+
+ String upload(Path localFile, String bucket, String key) {
+ try {
+ var storage = GcsClientFactory.getClient();
+ BlobId blobId = BlobId.of(bucket, key);
+ BlobInfo blobInfo = BlobInfo.newBuilder(blobId).build();
+ storage.createFrom(blobInfo, localFile);
+ String uri = String.format("gs://%s/%s", bucket, key);
+ log.info("Uploaded file to GCS at {}", uri);
+ return uri;
+ } catch (Exception e) {
+ log.error(
+ "Failed to upload file to GCS; deleting temp file {} and skipping any Redis updates", localFile, e);
+ // On failure, ensure the temporary local file is cleaned up
+ ReceivedFileStorage.super.deleteLocalTempQuietly(localFile);
+ return null;
+ }
+ }
+
+ // Use default deletion from interface
+
+ // Use default sanitize/buildKey/normalizeKey from interface
+}
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/common/annotations/ExcludeFromJacocoGeneratedReport.java b/src/main/java/uk/gov/dbt/ndtp/federator/common/annotations/ExcludeFromJacocoGeneratedReport.java
new file mode 100644
index 00000000..d97fdd5d
--- /dev/null
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/common/annotations/ExcludeFromJacocoGeneratedReport.java
@@ -0,0 +1,10 @@
+package uk.gov.dbt.ndtp.federator.common.annotations;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.METHOD)
+public @interface ExcludeFromJacocoGeneratedReport {}
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/common/model/SourceType.java b/src/main/java/uk/gov/dbt/ndtp/federator/common/model/SourceType.java
index de2d9c51..b6b835c9 100644
--- a/src/main/java/uk/gov/dbt/ndtp/federator/common/model/SourceType.java
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/common/model/SourceType.java
@@ -3,5 +3,6 @@
public enum SourceType {
S3,
AZURE,
+ GCP,
LOCAL
}
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/common/service/config/ConfigService.java b/src/main/java/uk/gov/dbt/ndtp/federator/common/service/config/ConfigService.java
index 7b693d4e..c261638b 100644
--- a/src/main/java/uk/gov/dbt/ndtp/federator/common/service/config/ConfigService.java
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/common/service/config/ConfigService.java
@@ -35,9 +35,10 @@ public interface ConfigService {
*/
default T fetchWithResilience() {
final String componentName = getKeyPrefix(); // single shared per service type
+ final String operation = "fetch configuration";
Supplier supplier = this::fetchConfiguration;
try {
- return ResilienceSupport.decorateAndExecute(componentName, supplier);
+ return ResilienceSupport.decorateAndExecute(componentName, operation, null, supplier);
} catch (RuntimeException ex) {
throw new ConfigFetchException(
"Failed to fetch configuration after resilience protections for component: " + componentName, ex);
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/common/service/config/exception/ConfigFetchException.java b/src/main/java/uk/gov/dbt/ndtp/federator/common/service/config/exception/ConfigFetchException.java
index 72786ac1..c10c43bf 100644
--- a/src/main/java/uk/gov/dbt/ndtp/federator/common/service/config/exception/ConfigFetchException.java
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/common/service/config/exception/ConfigFetchException.java
@@ -6,10 +6,12 @@
package uk.gov.dbt.ndtp.federator.common.service.config.exception;
+import uk.gov.dbt.ndtp.federator.exceptions.RebuildableRuntimeException;
+
/**
* Exception indicating configuration fetch failure after retries or due to circuit breaker state.
*/
-public class ConfigFetchException extends RuntimeException {
+public class ConfigFetchException extends RebuildableRuntimeException {
public ConfigFetchException(String message) {
super(message);
}
@@ -17,4 +19,15 @@ public ConfigFetchException(String message) {
public ConfigFetchException(String message, Throwable cause) {
super(message, cause);
}
+
+ /**
+ * Rebuilds this exception with the given message and cause.
+ * @param message the enriched error message
+ * @param cause the original exception
+ * @return a new instance of {@link ConfigFetchException}
+ */
+ @Override
+ public ConfigFetchException rebuild(String message, Throwable cause) {
+ return new ConfigFetchException(message, cause);
+ }
}
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/common/service/file/FileStreamService.java b/src/main/java/uk/gov/dbt/ndtp/federator/common/service/file/FileStreamService.java
index be58028c..a2887264 100644
--- a/src/main/java/uk/gov/dbt/ndtp/federator/common/service/file/FileStreamService.java
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/common/service/file/FileStreamService.java
@@ -8,7 +8,7 @@
import org.slf4j.LoggerFactory;
import uk.gov.dbt.ndtp.federator.common.model.dto.AttributesDTO;
import uk.gov.dbt.ndtp.federator.common.model.dto.ProducerConfigDTO;
-import uk.gov.dbt.ndtp.federator.common.service.stream.FederatorStreamService;
+import uk.gov.dbt.ndtp.federator.common.service.stream.CloseableFederatorStreamService;
import uk.gov.dbt.ndtp.federator.common.utils.ThreadUtil;
import uk.gov.dbt.ndtp.federator.server.conductor.FileConductor;
import uk.gov.dbt.ndtp.federator.server.conductor.MessageConductor;
@@ -18,18 +18,14 @@
import uk.gov.dbt.ndtp.grpc.FileStreamEvent;
import uk.gov.dbt.ndtp.grpc.FileStreamRequest;
-public class FileStreamService implements FederatorStreamService {
+public class FileStreamService extends CloseableFederatorStreamService {
private static final Logger LOGGER = LoggerFactory.getLogger(FileStreamService.class);
- private static final ExecutorService THREADED_EXECUTOR = ThreadUtil.threadExecutor("FileStreamService");
- /**
- * Streams file chunks to the client based on the file request.
- * @param fileRequest
- * @param streamObservable
- */
@Override
- public void streamToClient(FileStreamRequest fileRequest, StreamObservable streamObservable) {
-
+ public void streamToClient(
+ FileStreamRequest fileRequest,
+ StreamObservable streamObservable,
+ ExecutorService executorService) {
long offset = fileRequest.getStartSequenceId();
String consumerId = GRPCContextKeys.CLIENT_ID.get();
streamObservable.setOnCancelHandler(() -> LOGGER.info("Cancel called by client: {}", consumerId));
@@ -39,19 +35,29 @@ public void streamToClient(FileStreamRequest fileRequest, StreamObservable> futures = new ArrayList<>();
- futures.add(THREADED_EXECUTOR.submit(messageConductor::processMessages));
- LOGGER.info(
- "Awaiting FileStreamRequest finished for Client: {}, Topic: {}, Offset: {}",
- consumerId,
- topicData.getTopic(),
- topicData.getOffset());
- ThreadUtil.awaitShutdown(futures, messageConductor, THREADED_EXECUTOR);
- LOGGER.info(
- "Finished FileStreamRequest processed for Client: {}, Topic: {}, Offset: {}",
- consumerId,
- topicData.getTopic(),
- topicData.getOffset());
+ futures.add(executorService.submit(messageConductor::processMessages));
+
+ try {
+ LOGGER.info(
+ "Awaiting FileStreamRequest finished for Client: {}, Topic: {}, Offset: {}",
+ consumerId,
+ topicData.getTopic(),
+ topicData.getOffset());
+
+ ThreadUtil.awaitFutures(futures);
+
+ LOGGER.info(
+ "Finished FileStreamRequest processed for Client: {}, Topic: {}, Offset: {}",
+ consumerId,
+ topicData.getTopic(),
+ topicData.getOffset());
+ } finally {
+ messageConductors.remove(messageConductor);
+ }
+
streamObservable.onCompleted();
}
}
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/common/service/idp/AbstractIdpTokenService.java b/src/main/java/uk/gov/dbt/ndtp/federator/common/service/idp/AbstractIdpTokenService.java
index 41aab120..01db28dc 100644
--- a/src/main/java/uk/gov/dbt/ndtp/federator/common/service/idp/AbstractIdpTokenService.java
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/common/service/idp/AbstractIdpTokenService.java
@@ -43,10 +43,16 @@ protected AbstractIdpTokenService(String idpJwksUrl, HttpClient httpClient, Obje
@Override
public boolean verifyToken(String token) {
final String componentName = "idp-jwks-service";
+ final String operation = "verify token";
try {
- return ResilienceSupport.decorateAndExecute(componentName, () -> verifyTokenInternal(token));
+ return ResilienceSupport.decorateAndExecute(
+ componentName, operation, null, () -> verifyTokenInternal(token));
} catch (RuntimeException ex) {
- log.error("Token verification failed after resilience protections", ex);
+
+ String msg = ResilienceSupport.buildFailureMessage(
+ "Token verification failed after resilience protections", ex, componentName, operation, null);
+
+ log.error(msg, ex);
return false;
}
}
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/common/service/idp/IdpTokenService.java b/src/main/java/uk/gov/dbt/ndtp/federator/common/service/idp/IdpTokenService.java
index e2c7c9ea..63c22eab 100644
--- a/src/main/java/uk/gov/dbt/ndtp/federator/common/service/idp/IdpTokenService.java
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/common/service/idp/IdpTokenService.java
@@ -80,9 +80,10 @@ private String maskToken(String token) {
*/
default String fetchTokenWithResilience(String managementNodeId) {
final String componentName = "idp-token-service";
+ final String operation = "fetch token";
Supplier supplier = () -> fetchToken(managementNodeId);
try {
- return ResilienceSupport.decorateAndExecute(componentName, supplier);
+ return ResilienceSupport.decorateAndExecute(componentName, operation, managementNodeId, supplier);
} catch (RuntimeException ex) {
throw new FederatorTokenException(
"Failed to fetch token after resilience protections for management node: " + managementNodeId, ex);
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/common/service/kafka/KafkaStreamService.java b/src/main/java/uk/gov/dbt/ndtp/federator/common/service/kafka/KafkaStreamService.java
index 94940002..e852d1e7 100644
--- a/src/main/java/uk/gov/dbt/ndtp/federator/common/service/kafka/KafkaStreamService.java
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/common/service/kafka/KafkaStreamService.java
@@ -14,7 +14,7 @@
import uk.gov.dbt.ndtp.federator.common.model.dto.ConsumerDTO;
import uk.gov.dbt.ndtp.federator.common.model.dto.ProducerConfigDTO;
import uk.gov.dbt.ndtp.federator.common.model.dto.ProductDTO;
-import uk.gov.dbt.ndtp.federator.common.service.stream.FederatorStreamService;
+import uk.gov.dbt.ndtp.federator.common.service.stream.CloseableFederatorStreamService;
import uk.gov.dbt.ndtp.federator.common.utils.ThreadUtil;
import uk.gov.dbt.ndtp.federator.server.conductor.MessageConductor;
import uk.gov.dbt.ndtp.federator.server.conductor.RdfMessageConductor;
@@ -24,25 +24,17 @@
import uk.gov.dbt.ndtp.grpc.KafkaByteBatch;
import uk.gov.dbt.ndtp.grpc.TopicRequest;
-public class KafkaStreamService implements FederatorStreamService {
+public class KafkaStreamService extends CloseableFederatorStreamService {
public static final Logger LOGGER = LoggerFactory.getLogger("KafkaStreamService");
- private static final ExecutorService THREADED_EXECUTOR = ThreadUtil.threadExecutor("KafkaStream");
private final Set sharedHeaders;
public KafkaStreamService(Set sharedHeaders) {
this.sharedHeaders = sharedHeaders;
}
- /**
- * Takes a request with the topic, client id, key, offset and the streamObservable object to write
- * into.
- *
- * @param request that contains the details required to get data from a specific topic.
- * @param streamObservable used to write the data into.
- * @throws InvalidTopicException if the topic is not valid for a specific client.
- */
@Override
- public void streamToClient(TopicRequest request, StreamObservable streamObservable)
+ public void streamToClient(
+ TopicRequest request, StreamObservable streamObservable, ExecutorService executorService)
throws InvalidTopicException {
String topic = request.getTopic();
long offset = request.getOffset();
@@ -61,19 +53,28 @@ public void streamToClient(TopicRequest request, StreamObservable> futures = new ArrayList<>();
- futures.add(THREADED_EXECUTOR.submit(messageConductor::processMessages));
- LOGGER.info(
- "Awaiting TopicRequest finished for Client: {}, Topic: {}, Offset: {}",
- consumerId,
- topicData.getTopic(),
- topicData.getOffset());
- ThreadUtil.awaitShutdown(futures, messageConductor, THREADED_EXECUTOR);
- LOGGER.info(
- "Finished TopicRequest processed for Client: {}, Topic: {}, Offset: {}",
- consumerId,
- topicData.getTopic(),
- topicData.getOffset());
+ futures.add(executorService.submit(messageConductor::processMessages));
+
+ try {
+ LOGGER.info(
+ "Awaiting TopicRequest finished for Client: {}, Topic: {}, Offset: {}",
+ consumerId,
+ topicData.getTopic(),
+ topicData.getOffset());
+
+ ThreadUtil.awaitFutures(futures);
+
+ LOGGER.info(
+ "Finished TopicRequest processed for Client: {}, Topic: {}, Offset: {}",
+ consumerId,
+ topicData.getTopic(),
+ topicData.getOffset());
+ } finally {
+ messageConductors.remove(messageConductor);
+ }
streamObservable.onCompleted();
}
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/common/service/stream/CloseableFederatorStreamService.java b/src/main/java/uk/gov/dbt/ndtp/federator/common/service/stream/CloseableFederatorStreamService.java
new file mode 100644
index 00000000..80a894c8
--- /dev/null
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/common/service/stream/CloseableFederatorStreamService.java
@@ -0,0 +1,22 @@
+package uk.gov.dbt.ndtp.federator.common.service.stream;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import uk.gov.dbt.ndtp.federator.server.conductor.MessageConductor;
+
+/**
+ * An abstract class that implements both the {@link FederatorStreamService} and {@link AutoCloseable}
+ */
+public abstract class CloseableFederatorStreamService implements FederatorStreamService, AutoCloseable {
+ protected final List messageConductors = Collections.synchronizedList(new ArrayList<>());
+
+ @Override
+ public void close() {
+ for (MessageConductor messageConductor : messageConductors) {
+ messageConductor.close();
+ }
+
+ messageConductors.clear();
+ }
+}
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/common/service/stream/FederatorStreamService.java b/src/main/java/uk/gov/dbt/ndtp/federator/common/service/stream/FederatorStreamService.java
index b130e2bb..481493e3 100644
--- a/src/main/java/uk/gov/dbt/ndtp/federator/common/service/stream/FederatorStreamService.java
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/common/service/stream/FederatorStreamService.java
@@ -3,6 +3,7 @@
import java.util.Collections;
import java.util.List;
import java.util.Objects;
+import java.util.concurrent.ExecutorService;
import java.util.stream.Stream;
import org.slf4j.Logger;
import uk.gov.dbt.ndtp.federator.common.model.dto.AttributesDTO;
@@ -15,7 +16,14 @@
public interface FederatorStreamService {
Logger LOGGER = org.slf4j.LoggerFactory.getLogger(FederatorStreamService.class);
- void streamToClient(R request, StreamObservable streamObservable);
+ /**
+ * A method which streams data to a client as outlined in the request.
+ *
+ * @param request the details of the message streaming request.
+ * @param streamObservable the {@link StreamObservable} involved in the request.
+ * @param executorService the {@link ExecutorService} to submit tasks involved in processing messages invovled in the kafka message streaming to the client.
+ */
+ void streamToClient(R request, StreamObservable streamObservable, ExecutorService executorService);
default ProducerConfigDTO getProducerConfiguration() {
return ProducerConsumerConfigServiceFactory.getProducerConfigService().getProducerConfiguration();
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/FileProviderFactory.java b/src/main/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/FileProviderFactory.java
index 2c6e8be6..7dba823c 100644
--- a/src/main/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/FileProviderFactory.java
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/FileProviderFactory.java
@@ -2,8 +2,10 @@
import uk.gov.dbt.ndtp.federator.common.model.SourceType;
import uk.gov.dbt.ndtp.federator.common.storage.provider.file.client.AzureBlobClientFactory;
+import uk.gov.dbt.ndtp.federator.common.storage.provider.file.client.GcsClientFactory;
import uk.gov.dbt.ndtp.federator.common.storage.provider.file.client.S3ClientFactory;
import uk.gov.dbt.ndtp.federator.common.storage.provider.file.impl.AzureFileProvider;
+import uk.gov.dbt.ndtp.federator.common.storage.provider.file.impl.GCPFileProvider;
import uk.gov.dbt.ndtp.federator.common.storage.provider.file.impl.LocalFileProvider;
import uk.gov.dbt.ndtp.federator.common.storage.provider.file.impl.S3FileProvider;
@@ -16,13 +18,14 @@ private FileProviderFactory() {}
/**
* Returns a file provider suitable for the given source type.
*
- * @param sourceType the remote source type (S3, AZURE, LOCAL)
+ * @param sourceType the remote source type (S3, AZURE, GCP, LOCAL)
* @return a {@link FileProvider} capable of fetching from that source
*/
public static FileProvider getProvider(SourceType sourceType) {
return switch (sourceType) {
case S3 -> new S3FileProvider(S3ClientFactory.getClient());
case AZURE -> new AzureFileProvider(AzureBlobClientFactory.getClient());
+ case GCP -> new GCPFileProvider(GcsClientFactory.getClient());
case LOCAL -> new LocalFileProvider();
};
}
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/client/GcsClientFactory.java b/src/main/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/client/GcsClientFactory.java
new file mode 100644
index 00000000..72fa655f
--- /dev/null
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/client/GcsClientFactory.java
@@ -0,0 +1,130 @@
+package uk.gov.dbt.ndtp.federator.common.storage.provider.file.client;
+
+import com.google.auth.Credentials;
+import com.google.auth.oauth2.GoogleCredentials;
+import com.google.cloud.NoCredentials;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicReference;
+import lombok.extern.slf4j.Slf4j;
+import uk.gov.dbt.ndtp.federator.common.utils.PropertyUtil;
+
+/**
+ * Factory for a singleton Google Cloud Storage client used across client and server components.
+ *
+ * Supported configuration via {@link PropertyUtil} keys:
+ * - {@code gcp.storage.project.id} – GCP project ID (optional; falls back to default)
+ * - {@code gcp.storage.endpoint.url} – optional GCS-compatible endpoint (e.g., fake-gcs-server)
+ *
+ * Credential resolution:
+ * - Uses {@link GoogleCredentials#getApplicationDefault()} for authentication.
+ * - This supports Service Accounts via ADC (Application Default Credentials): env var, gcloud, GCE/GKE metadata, etc.
+ * - For emulator endpoints, uses {@link NoCredentials}.
+ *
+ * Project ID resolution:
+ * - If {@code gcp.storage.project.id} provided, use it; otherwise fall back to default from credentials or environment.
+ *
+ * Custom endpoint support for local testing (e.g., fake-gcs-server).
+ */
+@Slf4j
+public final class GcsClientFactory {
+
+ // Lazily initialized singleton to avoid class-load failures if configuration is bad
+ private static final AtomicReference gcsClient = new AtomicReference<>();
+
+ private GcsClientFactory() {}
+
+ // Orchestrates the modular steps to create the GCS client
+ private static Storage createClient() {
+ GcsSettings settings = GcsSettings.fromProperties();
+ return buildClient(settings);
+ }
+
+ // Build the GCS client using resolved components
+ private static Storage buildClient(GcsSettings settings) {
+ Credentials credentials = resolveCredentials(settings);
+ String projectId = resolveProjectId(settings);
+
+ StorageOptions.Builder builder = StorageOptions.newBuilder().setCredentials(credentials);
+
+ if (projectId != null && !projectId.isBlank()) {
+ builder = builder.setProjectId(projectId);
+ }
+
+ builder = applyEndpointOverride(builder, settings);
+
+ return builder.build().getService();
+ }
+
+ // Select credentials based on settings (application default or no credentials for emulator)
+ private static Credentials resolveCredentials(GcsSettings settings) {
+ if (settings.endpointUrl != null && !settings.endpointUrl.isBlank()) {
+ log.info("GCS emulator endpoint configured; using NoCredentials");
+ return NoCredentials.getInstance();
+ }
+
+ try {
+ log.info("Using Application Default Credentials for GCS");
+ return GoogleCredentials.getApplicationDefault();
+ } catch (IOException e) {
+ throw new IllegalStateException("Failed to obtain Application Default Credentials for GCS", e);
+ }
+ }
+
+ // Determine project ID from explicit configuration
+ private static String resolveProjectId(GcsSettings settings) {
+ if (settings.projectId != null && !settings.projectId.isBlank()) {
+ return settings.projectId;
+ }
+ log.info("No explicit GCP project ID configured; will use default from environment");
+ return null;
+ }
+
+ // Optionally apply endpoint override, useful for fake-gcs-server or custom GCS endpoints
+ private static StorageOptions.Builder applyEndpointOverride(StorageOptions.Builder builder, GcsSettings settings) {
+ if (settings.endpointUrl != null && !settings.endpointUrl.isBlank()) {
+ log.info("Using custom GCS endpoint: {}", settings.endpointUrl);
+ return builder.setHost(settings.endpointUrl);
+ }
+ return builder;
+ }
+
+ /** Returns the singleton {@link Storage} instance configured from properties. */
+ public static Storage getClient() {
+ return gcsClient.updateAndGet(current -> {
+ if (current != null) {
+ return current;
+ }
+ try {
+ return createClient();
+ } catch (Exception e) {
+ log.error("Failed to initialize GCS Storage client from properties.", e);
+ throw new IllegalStateException("Failed to initialize GCS Storage client from properties", e);
+ }
+ });
+ }
+
+ /** Resets the singleton instance (primarily for testing). */
+ static void resetClient() {
+ gcsClient.set(null);
+ }
+
+ // Encapsulates all properties used to configure the GCS client
+ private static final class GcsSettings {
+ private final String projectId;
+ private final String endpointUrl;
+
+ private GcsSettings(String projectId, String endpointUrl) {
+ this.projectId = projectId;
+ this.endpointUrl = endpointUrl;
+ }
+
+ static GcsSettings fromProperties() {
+ // These properties are optional; use null defaults to avoid exceptions when absent
+ String projectId = PropertyUtil.getPropertyValue("gcp.storage.project.id", null);
+ String endpointUrl = PropertyUtil.getPropertyValue("gcp.storage.endpoint.url", "");
+ return new GcsSettings(projectId, endpointUrl);
+ }
+ }
+}
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/impl/GCPFileProvider.java b/src/main/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/impl/GCPFileProvider.java
new file mode 100644
index 00000000..a8922a0f
--- /dev/null
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/impl/GCPFileProvider.java
@@ -0,0 +1,92 @@
+package uk.gov.dbt.ndtp.federator.common.storage.provider.file.impl;
+
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageException;
+import java.io.InputStream;
+import java.nio.channels.Channels;
+import uk.gov.dbt.ndtp.federator.common.exception.FileTransferException;
+import uk.gov.dbt.ndtp.federator.common.model.FileTransferRequest;
+import uk.gov.dbt.ndtp.federator.common.storage.provider.file.FileProvider;
+import uk.gov.dbt.ndtp.federator.exceptions.FileFetcherException;
+import uk.gov.dbt.ndtp.federator.server.processor.file.FileTransferResult;
+
+/**
+ * {@link FileProvider} implementation that fetches files from Google Cloud Storage using an injected {@link com.google.cloud.storage.Storage}.
+ * Resolves object size via a metadata call before opening the GET stream.
+ */
+public class GCPFileProvider implements FileProvider {
+
+ private final Storage storage;
+
+ public GCPFileProvider(Storage storage) {
+ this.storage = storage;
+ }
+
+ /**
+ * Fetches the file specified in the FileTransferRequest from GCS.
+ * @param request
+ * @return
+ */
+ @Override
+ public FileTransferResult get(FileTransferRequest request) {
+ try {
+ BlobId blobId = BlobId.of(request.storageContainer(), request.path());
+
+ Blob blob = storage.get(blobId);
+ if (blob == null || !blob.exists()) {
+ throw new FileFetcherException(
+ "File not found in GCS: " + request.storageContainer() + "/" + request.path());
+ }
+
+ long size = blob.getSize();
+ InputStream stream = Channels.newInputStream(blob.reader());
+
+ return new FileTransferResult(stream, size);
+
+ } catch (FileFetcherException e) {
+ throw e;
+ } catch (StorageException e) {
+ if (e.getCode() == 404) {
+ throw new FileFetcherException(
+ "File not found in GCS: " + request.storageContainer() + "/" + request.path());
+ }
+ throw new FileFetcherException(
+ "GCS error fetching: " + request.storageContainer() + "/" + request.path(), e);
+ } catch (Exception e) {
+ throw new FileFetcherException(
+ "Failed to fetch from GCS: " + request.storageContainer() + "/" + request.path(), e);
+ }
+ }
+
+ /**
+ * Validates that the GCS object exists by checking its metadata.
+ * @param request the file transfer request containing the GCS bucket and object path to validate
+ * @throws FileTransferException if the GCS object does not exist or cannot be accessed
+ */
+ @Override
+ public void validatePath(FileTransferRequest request) {
+ validateStorageContainer(request, "GCS bucket");
+
+ executeValidation(
+ () -> {
+ try {
+ BlobId blobId = BlobId.of(request.storageContainer(), request.path());
+ Blob blob = storage.get(blobId);
+ if (blob == null || !blob.exists()) {
+ throw new FileTransferException(
+ "GCS object not found: " + request.storageContainer() + "/" + request.path());
+ }
+ } catch (StorageException e) {
+ if (e.getCode() == 404) {
+ throw new FileTransferException(
+ "GCS object not found: " + request.storageContainer() + "/" + request.path());
+ }
+ throw new FileTransferException(
+ "GCS validation error: " + request.storageContainer() + "/" + request.path(), e);
+ }
+ },
+ "Invalid GCS path: " + request.storageContainer() + "/" + request.path());
+ }
+}
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/common/utils/ResilienceSupport.java b/src/main/java/uk/gov/dbt/ndtp/federator/common/utils/ResilienceSupport.java
index 5738a1a9..388e3fbc 100644
--- a/src/main/java/uk/gov/dbt/ndtp/federator/common/utils/ResilienceSupport.java
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/common/utils/ResilienceSupport.java
@@ -22,10 +22,13 @@
import org.jspecify.annotations.NonNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import redis.clients.jedis.exceptions.JedisException;
+import uk.gov.dbt.ndtp.federator.exceptions.RebuildableRuntimeException;
/**
* Centralized Resilience4j configuration and decoration helpers.
- * Uses PropertyUtil for configuration under prefix: management.node.resilience.*
+ * Uses PropertyUtil for configuration under prefix:
+ * management.node.resilience.*
*/
public final class ResilienceSupport {
@@ -62,7 +65,8 @@ private static CircuitBreakerRegistry getCircuitBreakerRegistry() {
private static RetryConfig buildRetryConfig() {
int maxAttempts = PropertyUtil.getPropertyIntValue(PROP_RETRY_MAX_ATTEMPTS, "10");
- // Requirement: 5 attempts within 5 minutes. Some versions may not support maxDuration; we always enforce
+ // Requirement: 5 attempts within 5 minutes. Some versions may not support
+ // maxDuration; we always enforce
// attempts.
// Exponential backoff with a cap at 5 minutes between attempts
Duration maxBackoff = PropertyUtil.getPropertyDurationValue(
@@ -139,20 +143,28 @@ public static CircuitBreaker getCircuitBreaker(String name) {
}
/**
- * Testing helper to clear registries so tests can reconfigure policies per test.
+ * Testing helper to clear registries so tests can reconfigure policies per
+ * test.
*/
public static void clearForTests() {
retryRegistry.set(null);
circuitBreakerRegistry.set(null);
}
- public static T decorateAndExecute(String componentName, Supplier supplier) {
+ public static T decorateAndExecute(
+ String componentName, String operation, String targetId, Supplier supplier) {
Retry retry = getRetry(componentName);
CircuitBreaker circuitBreaker = getCircuitBreaker(componentName);
Supplier withCb = CircuitBreaker.decorateSupplier(circuitBreaker, supplier);
Supplier withRetry = Retry.decorateSupplier(retry, withCb);
- return withRetry.get();
+
+ try {
+ return withRetry.get();
+ } catch (RuntimeException ex) {
+ enrichAndRethrow(ex, componentName, operation, targetId);
+ return null;
+ }
}
private static Class[] parseExceptionClasses(String csv) {
@@ -175,4 +187,107 @@ private static Class[] parseExceptionClasses(String csv) {
}
return classes.toArray(new Class[0]);
}
+
+ /**
+ * A helper method that builds a detailed error message
+ *
+ * @param baseMessage the base message for the exception
+ * @param ex the exception thrown
+ * @param componentName the name of the component throwing the exception
+ * @param operation the name of the operation throwing the exception
+ * @param targetId the target id for the operation
+ * @return an enriched failure message
+ */
+ public static String buildFailureMessage(
+ String baseMessage, Throwable ex, String componentName, String operation, String targetId) {
+ return buildFailureMessage(baseMessage, getExceptionDetails(ex, componentName, operation), targetId);
+ }
+
+ /**
+ * A helper method that builds a detailed error message
+ *
+ * @param ex the exception thrown
+ * @param componentName the name of the component throwing the exception
+ * @param operation the name of the operation throwing the exception
+ * @param targetId the target id for the operation
+ * @return an enriched failure message
+ */
+ private static String buildFailureMessage(Throwable ex, String componentName, String operation, String targetId) {
+ return buildFailureMessage(ex.getMessage(), getExceptionDetails(ex, componentName, operation), targetId);
+ }
+
+ /**
+ * A helper method to fetch human readabe details from parameters
+ *
+ * @param ex the exception throw
+ * @param componentName the name of the component throwing the exception
+ * @param operation the name of the operation where the exception is thrown
+ * @return a string with human readable details from the provided parameters
+ */
+ private static String getExceptionDetails(Throwable ex, String componentName, String operation) {
+ Throwable root = getRootCause(ex);
+
+ return switch (root) {
+ case java.net.SocketTimeoutException ignored -> "timeout while calling " + componentName;
+
+ case java.net.http.HttpTimeoutException ignored -> "timeout while calling " + componentName;
+
+ case java.io.InterruptedIOException ignored -> {
+ Thread.currentThread().interrupt();
+ yield "request was interrupted";
+ }
+
+ case InterruptedException ignored -> {
+ Thread.currentThread().interrupt();
+ yield "request was interrupted";
+ }
+
+ case java.io.IOException ignored -> "I/O error while calling " + componentName;
+
+ case JedisException ignored -> "redis cache failure";
+
+ default -> "unexpected failure during " + operation;
+ };
+ }
+
+ /**
+ * A helper method that creates a detailed formatted error message
+ *
+ * @param baseMessage the base message included in the original exception
+ * @param detail the human readable detailed message for the exception
+ * @param targetId the target id for the operation
+ * @return a detailed formatted error message
+ */
+ private static String buildFailureMessage(String baseMessage, String detail, String targetId) {
+
+ String targetSuffix = (targetId != null && !targetId.isBlank()) ? " for " + targetId : "";
+
+ return "%s (%s%s)".formatted((baseMessage == null ? "" : baseMessage), detail, targetSuffix);
+ }
+
+ private static Throwable getRootCause(Throwable ex) {
+ Throwable current = ex;
+ while (current.getCause() != null) {
+ current = current.getCause();
+ }
+ return current;
+ }
+
+ /**
+ * Enriches the message of a RebuildableRuntimeException and rethrows it.
+ * If the exception is not a RebuildableRuntimeException it is rethrown as-is.
+ *
+ * @param e the exception to enrich
+ * @param componentName the name of the component involved in the exception
+ * @param operation the name of the operation attempted before the exception was raised
+ * @param targetId the id of the target involved in the exception
+ */
+ private static void enrichAndRethrow(RuntimeException ex, String componentName, String operation, String targetId) {
+ if (!(ex instanceof RebuildableRuntimeException rebuildableRuntimeException)) {
+ throw ex;
+ }
+
+ String enrichedMessage = buildFailureMessage(ex, componentName, operation, targetId);
+ throw rebuildableRuntimeException.rebuild(enrichedMessage, ex);
+ }
}
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/common/utils/ThreadUtil.java b/src/main/java/uk/gov/dbt/ndtp/federator/common/utils/ThreadUtil.java
index c3ac8260..f3631abc 100644
--- a/src/main/java/uk/gov/dbt/ndtp/federator/common/utils/ThreadUtil.java
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/common/utils/ThreadUtil.java
@@ -61,7 +61,11 @@ public static void awaitShutdown(
LOGGER.info("Exception occurred during shutdown, ignoring.", e);
}
}));
- for (Future future : futureList) {
+ awaitFutures(futureList);
+ }
+
+ public static void awaitFutures(List> futures) {
+ for (Future future : futures) {
try {
future.get();
LOGGER.info("Future processed: {}", future);
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/exceptions/FederatorTokenException.java b/src/main/java/uk/gov/dbt/ndtp/federator/exceptions/FederatorTokenException.java
index 25927af7..2916d6d5 100644
--- a/src/main/java/uk/gov/dbt/ndtp/federator/exceptions/FederatorTokenException.java
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/exceptions/FederatorTokenException.java
@@ -2,7 +2,7 @@
/**
* Exception thrown when there is an error related to federator tokens.
*/
-public class FederatorTokenException extends RuntimeException {
+public class FederatorTokenException extends RebuildableRuntimeException {
public FederatorTokenException(String message) {
super(message);
@@ -11,4 +11,15 @@ public FederatorTokenException(String message) {
public FederatorTokenException(String message, Throwable cause) {
super(message, cause);
}
+
+ /**
+ * Rebuilds this exception with the given message and cause.
+ * @param message the enriched error message
+ * @param cause the original exception
+ * @return a new instance of {@link FederatorTokenException}
+ */
+ @Override
+ public FederatorTokenException rebuild(String message, Throwable cause) {
+ return new FederatorTokenException(message, cause);
+ }
}
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/exceptions/RebuildableRuntimeException.java b/src/main/java/uk/gov/dbt/ndtp/federator/exceptions/RebuildableRuntimeException.java
new file mode 100644
index 00000000..283787f6
--- /dev/null
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/exceptions/RebuildableRuntimeException.java
@@ -0,0 +1,18 @@
+package uk.gov.dbt.ndtp.federator.exceptions;
+
+/**
+ * Abstract base for runtime exceptions that enforce rebuildability.
+ * Subclasses must implement {@link #rebuild(String, Throwable)} to return
+ * a new instance of themselves with the given message and cause.
+ */
+public abstract class RebuildableRuntimeException extends RuntimeException {
+ protected RebuildableRuntimeException(String message) {
+ super(message);
+ }
+
+ protected RebuildableRuntimeException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public abstract RebuildableRuntimeException rebuild(String message, Throwable cause);
+}
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/server/conductor/AbstractKafkaEventMessageConductor.java b/src/main/java/uk/gov/dbt/ndtp/federator/server/conductor/AbstractKafkaEventMessageConductor.java
index ad95f574..8a2e68ce 100644
--- a/src/main/java/uk/gov/dbt/ndtp/federator/server/conductor/AbstractKafkaEventMessageConductor.java
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/server/conductor/AbstractKafkaEventMessageConductor.java
@@ -67,6 +67,8 @@ public void processMessages() throws MessageProcessingException {
}
} catch (Exception e) {
throw new MessageProcessingException(e);
+ } finally {
+ super.close();
}
}
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/server/conductor/AbstractMessageConductor.java b/src/main/java/uk/gov/dbt/ndtp/federator/server/conductor/AbstractMessageConductor.java
index a878eeb6..6632c058 100644
--- a/src/main/java/uk/gov/dbt/ndtp/federator/server/conductor/AbstractMessageConductor.java
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/server/conductor/AbstractMessageConductor.java
@@ -73,6 +73,8 @@ public void processMessages() throws MessageProcessingException {
}
} catch (Exception e) {
throw new MessageProcessingException(e);
+ } finally {
+ close();
}
}
@@ -84,10 +86,13 @@ public boolean continueProcessing() {
@Override
public void close() {
try {
- messageConsumer.close();
+ if (messageConsumer.stillAvailable()) {
+ messageConsumer.close();
+ }
} catch (Exception ex) {
LOGGER.info("Error whilst closing consumer, ignoring.", ex);
}
+
try {
messageProcessor.close();
} catch (Exception ex) {
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/server/conductor/FileConductor.java b/src/main/java/uk/gov/dbt/ndtp/federator/server/conductor/FileConductor.java
index 2565f652..702a049f 100644
--- a/src/main/java/uk/gov/dbt/ndtp/federator/server/conductor/FileConductor.java
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/server/conductor/FileConductor.java
@@ -56,7 +56,6 @@ private FileConductor(
public boolean continueProcessing() {
if (serverCallStreamObserver.isCancelled()) {
LOGGER.info("Observer is closed on client end. Stop further processing.");
- messageConsumer.close();
return false;
}
return messageConsumer.stillAvailable();
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/server/conductor/RdfMessageConductor.java b/src/main/java/uk/gov/dbt/ndtp/federator/server/conductor/RdfMessageConductor.java
index 84ca0423..dcddb049 100644
--- a/src/main/java/uk/gov/dbt/ndtp/federator/server/conductor/RdfMessageConductor.java
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/server/conductor/RdfMessageConductor.java
@@ -83,7 +83,6 @@ private RdfMessageConductor(
public boolean continueProcessing() {
if (serverCallStreamObserver.isCancelled()) {
LOGGER.info("Observer is closed on client end. Stop further processing.");
- messageConsumer.close();
return false;
}
return messageConsumer.stillAvailable();
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/server/grpc/GRPCFederatorService.java b/src/main/java/uk/gov/dbt/ndtp/federator/server/grpc/GRPCFederatorService.java
index b45ae91d..40f2e7fd 100644
--- a/src/main/java/uk/gov/dbt/ndtp/federator/server/grpc/GRPCFederatorService.java
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/server/grpc/GRPCFederatorService.java
@@ -34,6 +34,7 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import uk.gov.dbt.ndtp.federator.FederatorService;
+import uk.gov.dbt.ndtp.federator.common.annotations.ExcludeFromJacocoGeneratedReport;
import uk.gov.dbt.ndtp.federator.server.interfaces.StreamObservable;
import uk.gov.dbt.ndtp.grpc.FederatorServiceGrpc;
import uk.gov.dbt.ndtp.grpc.FileStreamEvent;
@@ -44,7 +45,7 @@
/**
* GRPC specific federator service that uses the POJO federator service and wrappers.
*/
-public class GRPCFederatorService extends FederatorServiceGrpc.FederatorServiceImplBase {
+public class GRPCFederatorService extends FederatorServiceGrpc.FederatorServiceImplBase implements AutoCloseable {
public static final Logger LOGGER = LoggerFactory.getLogger("GRPCFederatorService");
@@ -90,4 +91,10 @@ public void getFilesStream(FileStreamRequest request, StreamObserver(serverCallStreamObserver);
federator.getFileConsumer(request, streamObservable);
}
+
+ @ExcludeFromJacocoGeneratedReport
+ @Override
+ public void close() {
+ federator.close();
+ }
}
diff --git a/src/main/java/uk/gov/dbt/ndtp/federator/server/grpc/GRPCServer.java b/src/main/java/uk/gov/dbt/ndtp/federator/server/grpc/GRPCServer.java
index c0f8e9ce..1a3bcd66 100644
--- a/src/main/java/uk/gov/dbt/ndtp/federator/server/grpc/GRPCServer.java
+++ b/src/main/java/uk/gov/dbt/ndtp/federator/server/grpc/GRPCServer.java
@@ -43,6 +43,7 @@
import lombok.SneakyThrows;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import uk.gov.dbt.ndtp.federator.common.annotations.ExcludeFromJacocoGeneratedReport;
import uk.gov.dbt.ndtp.federator.common.service.idp.IdpTokenService;
import uk.gov.dbt.ndtp.federator.common.utils.GRPCUtils;
import uk.gov.dbt.ndtp.federator.common.utils.PropertyUtil;
@@ -78,8 +79,10 @@ public class GRPCServer implements AutoCloseable {
private final Server server;
private ServerCredentials creds;
+ private GRPCFederatorService grpcFederatorService;
public GRPCServer(Set sharedHeaders) {
+ grpcFederatorService = new GRPCFederatorService(sharedHeaders);
if (PropertyUtil.getPropertyBooleanValue(SERVER_MTLS_ENABLED, FALSE)) {
creds = generateServerCredentials();
server = generateSecureServer(creds, sharedHeaders);
@@ -122,7 +125,8 @@ private ServerCredentials generateServerCredentials() {
String trustStorePassword = PropertyUtil.getPropertyValue(SERVER_TRUSTSTORE_PASSWORD);
LOGGER.info(
- "Using p12 file path: {}, truststore file path: {}, p12 password is set: {}, truststore password is set: {}",
+ "Using p12 file path: {}, truststore file path: {}, p12 password is set: {}, truststore password is"
+ + " set: {}",
p12FilePath,
trustStoreFilePath,
p12Password != null,
@@ -148,10 +152,12 @@ public void start() {
}
}
+ @ExcludeFromJacocoGeneratedReport
@Override
public void close() {
try {
LOGGER.info("GRPCServer close called");
+ grpcFederatorService.close();
server.shutdown().awaitTermination(30, TimeUnit.SECONDS);
LOGGER.info("GRPCServer closed");
} catch (InterruptedException e) {
diff --git a/src/test/java/uk/gov/dbt/ndtp/federator/FederatorServiceTest.java b/src/test/java/uk/gov/dbt/ndtp/federator/FederatorServiceTest.java
index 1d4fddf6..efede0f0 100644
--- a/src/test/java/uk/gov/dbt/ndtp/federator/FederatorServiceTest.java
+++ b/src/test/java/uk/gov/dbt/ndtp/federator/FederatorServiceTest.java
@@ -31,7 +31,7 @@
import java.util.Set;
import org.apache.kafka.common.errors.InvalidTopicException;
import org.junit.jupiter.api.Test;
-import uk.gov.dbt.ndtp.federator.common.service.stream.FederatorStreamService;
+import uk.gov.dbt.ndtp.federator.common.service.stream.CloseableFederatorStreamService;
import uk.gov.dbt.ndtp.federator.server.interfaces.StreamObservable;
import uk.gov.dbt.ndtp.grpc.TopicRequest;
@@ -55,7 +55,7 @@ void test_getKafkaConsumer_delegatesToKafkaStreamService() throws Exception {
FederatorService cut = new FederatorService(headers);
@SuppressWarnings("rawtypes")
- FederatorStreamService mockKafka = mock(FederatorStreamService.class);
+ CloseableFederatorStreamService mockKafka = mock(CloseableFederatorStreamService.class);
setPrivateField(cut, "kafkaStreamService", mockKafka);
TopicRequest request =
@@ -66,7 +66,7 @@ void test_getKafkaConsumer_delegatesToKafkaStreamService() throws Exception {
cut.getKafkaConsumer(request, observable);
// Assert
- verify(mockKafka, times(1)).streamToClient(request, observable);
+ verify(mockKafka, times(1)).streamToClient(eq(request), eq(observable), any());
verifyNoMoreInteractions(mockKafka);
}
@@ -75,13 +75,15 @@ void test_getKafkaConsumer_propagatesInvalidTopicException() {
// Arrange
FederatorService cut = new FederatorService(Set.of());
@SuppressWarnings("rawtypes")
- FederatorStreamService mockKafka = mock(FederatorStreamService.class);
+ CloseableFederatorStreamService mockKafka = mock(CloseableFederatorStreamService.class);
setPrivateField(cut, "kafkaStreamService", mockKafka);
TopicRequest request = TopicRequest.newBuilder().setTopic("forbidden").build();
StreamObservable observable = mock(StreamObservable.class);
- doThrow(new InvalidTopicException("not allowed")).when(mockKafka).streamToClient(request, observable);
+ doThrow(new InvalidTopicException("not allowed"))
+ .when(mockKafka)
+ .streamToClient(eq(request), eq(observable), any());
// Act + Assert
assertThrows(InvalidTopicException.class, () -> cut.getKafkaConsumer(request, observable));
@@ -92,7 +94,7 @@ void test_getFileConsumer_delegatesToFileStreamService() {
// Arrange
FederatorService cut = new FederatorService(Set.of());
@SuppressWarnings("rawtypes")
- FederatorStreamService mockFile = mock(FederatorStreamService.class);
+ CloseableFederatorStreamService mockFile = mock(CloseableFederatorStreamService.class);
setPrivateField(cut, "fileStreamService", mockFile);
uk.gov.dbt.ndtp.grpc.FileStreamRequest request = uk.gov.dbt.ndtp.grpc.FileStreamRequest.newBuilder()
@@ -104,7 +106,26 @@ void test_getFileConsumer_delegatesToFileStreamService() {
cut.getFileConsumer(request, observable);
// Assert
- verify(mockFile, times(1)).streamToClient(request, observable);
+ verify(mockFile, times(1)).streamToClient(eq(request), eq(observable), any());
verifyNoMoreInteractions(mockFile);
}
+
+ @Test
+ void test_close_ClosesBothTheKafkaStreamServiceAndTheFileStreamService() {
+ // Arrange
+ FederatorService cut = new FederatorService(Set.of());
+ @SuppressWarnings("rawtypes")
+ CloseableFederatorStreamService mockKafka = mock(CloseableFederatorStreamService.class);
+ setPrivateField(cut, "kafkaStreamService", mockKafka);
+ @SuppressWarnings("rawtypes")
+ CloseableFederatorStreamService mockFile = mock(CloseableFederatorStreamService.class);
+ setPrivateField(cut, "fileStreamService", mockFile);
+
+ // Act
+ cut.close();
+
+ // Assert
+ verify(mockKafka, times(1)).close();
+ verify(mockFile, times(1)).close();
+ }
}
diff --git a/src/test/java/uk/gov/dbt/ndtp/federator/client/grpc/file/FileChunkAssemblerTest.java b/src/test/java/uk/gov/dbt/ndtp/federator/client/grpc/file/FileChunkAssemblerTest.java
index 0485772d..5d517351 100644
--- a/src/test/java/uk/gov/dbt/ndtp/federator/client/grpc/file/FileChunkAssemblerTest.java
+++ b/src/test/java/uk/gov/dbt/ndtp/federator/client/grpc/file/FileChunkAssemblerTest.java
@@ -10,6 +10,13 @@
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+import uk.gov.dbt.ndtp.federator.client.storage.ReceivedFileStorage;
+import uk.gov.dbt.ndtp.federator.client.storage.ReceivedFileStorageFactory;
+import uk.gov.dbt.ndtp.federator.client.storage.StoredFileResult;
+import uk.gov.dbt.ndtp.federator.client.storage.impl.GCPReceivedFileStorage;
+import uk.gov.dbt.ndtp.federator.client.storage.impl.S3ReceivedFileStorage;
import uk.gov.dbt.ndtp.federator.common.utils.GRPCUtils;
import uk.gov.dbt.ndtp.federator.exceptions.FileAssemblyException;
import uk.gov.dbt.ndtp.grpc.FileChunk;
@@ -268,4 +275,103 @@ void testHandleLastChunk_MoveFails() throws Exception {
// If that also fails, it will throw the IOException.
assertThrows(java.io.IOException.class, () -> assembler.accept(c1));
}
+
+ @Test
+ void testHandleLastChunk_GCPStorageSuccess_returnsPath() {
+ FileChunkAssembler assembler = new FileChunkAssembler(tempDir);
+ String fileName = "gcptest.txt";
+ long seq = 7L;
+ String emptyChecksum = GRPCUtils.calculateSha256Checksum(new byte[0]);
+
+ FileChunk last = FileChunk.newBuilder()
+ .setFileName(fileName)
+ .setFileSequenceId(seq)
+ .setIsLastChunk(true)
+ .setFileSize(0)
+ .setTotalChunks(1)
+ .setFileChecksum(emptyChecksum)
+ .build();
+
+ // Mock ReceivedFileStorageFactory to return a mock GCP storage that succeeds
+ try (MockedStatic factoryMock =
+ Mockito.mockStatic(ReceivedFileStorageFactory.class)) {
+ ReceivedFileStorage mockGCPStorage = Mockito.mock(GCPReceivedFileStorage.class);
+ factoryMock.when(ReceivedFileStorageFactory::get).thenReturn(mockGCPStorage);
+
+ // Mock successful storage with remote URI present
+ Path mockPath = tempDir.resolve(fileName);
+ StoredFileResult successResult = new StoredFileResult(mockPath, "gs://my-bucket/gcptest.txt");
+ Mockito.when(mockGCPStorage.store(Mockito.any(), Mockito.eq(fileName), Mockito.any()))
+ .thenReturn(successResult);
+
+ Path result = assembler.accept(last);
+ assertNotNull(result, "Should return path when GCP storage succeeds");
+ }
+ }
+
+ @Test
+ void testHandleLastChunk_GCPStorageFailure_returnsNull() {
+ FileChunkAssembler assembler = new FileChunkAssembler(tempDir);
+ String fileName = "gcpfail.txt";
+ long seq = 8L;
+ String emptyChecksum = GRPCUtils.calculateSha256Checksum(new byte[0]);
+
+ FileChunk last = FileChunk.newBuilder()
+ .setFileName(fileName)
+ .setFileSequenceId(seq)
+ .setIsLastChunk(true)
+ .setFileSize(0)
+ .setTotalChunks(1)
+ .setFileChecksum(emptyChecksum)
+ .build();
+
+ // Mock ReceivedFileStorageFactory to return a mock GCP storage that fails (no remote URI)
+ try (MockedStatic factoryMock =
+ Mockito.mockStatic(ReceivedFileStorageFactory.class)) {
+ ReceivedFileStorage mockGCPStorage = Mockito.mock(GCPReceivedFileStorage.class);
+ factoryMock.when(ReceivedFileStorageFactory::get).thenReturn(mockGCPStorage);
+
+ // Mock failed storage with no remote URI
+ Path mockPath = tempDir.resolve(fileName);
+ StoredFileResult failureResult = new StoredFileResult(mockPath, null);
+ Mockito.when(mockGCPStorage.store(Mockito.any(), Mockito.eq(fileName), Mockito.any()))
+ .thenReturn(failureResult);
+
+ Path result = assembler.accept(last);
+ assertNull(result, "Should return null when GCP storage fails (no remote URI)");
+ }
+ }
+
+ @Test
+ void testHandleLastChunk_S3StorageFailure_returnsNull() {
+ FileChunkAssembler assembler = new FileChunkAssembler(tempDir);
+ String fileName = "s3fail.txt";
+ long seq = 9L;
+ String emptyChecksum = GRPCUtils.calculateSha256Checksum(new byte[0]);
+
+ FileChunk last = FileChunk.newBuilder()
+ .setFileName(fileName)
+ .setFileSequenceId(seq)
+ .setIsLastChunk(true)
+ .setFileSize(0)
+ .setTotalChunks(1)
+ .setFileChecksum(emptyChecksum)
+ .build();
+
+ // Mock ReceivedFileStorageFactory to return a mock S3 storage that fails (no remote URI)
+ try (MockedStatic factoryMock =
+ Mockito.mockStatic(ReceivedFileStorageFactory.class)) {
+ ReceivedFileStorage mockS3Storage = Mockito.mock(S3ReceivedFileStorage.class);
+ factoryMock.when(ReceivedFileStorageFactory::get).thenReturn(mockS3Storage);
+
+ // Mock failed storage with no remote URI
+ Path mockPath = tempDir.resolve(fileName);
+ StoredFileResult failureResult = new StoredFileResult(mockPath, null);
+ Mockito.when(mockS3Storage.store(Mockito.any(), Mockito.eq(fileName), Mockito.any()))
+ .thenReturn(failureResult);
+
+ Path result = assembler.accept(last);
+ assertNull(result, "Should return null when S3 storage fails (no remote URI)");
+ }
+ }
}
diff --git a/src/test/java/uk/gov/dbt/ndtp/federator/client/storage/ReceivedFileStorageFactoryTest.java b/src/test/java/uk/gov/dbt/ndtp/federator/client/storage/ReceivedFileStorageFactoryTest.java
new file mode 100644
index 00000000..1b044427
--- /dev/null
+++ b/src/test/java/uk/gov/dbt/ndtp/federator/client/storage/ReceivedFileStorageFactoryTest.java
@@ -0,0 +1,158 @@
+package uk.gov.dbt.ndtp.federator.client.storage;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+import uk.gov.dbt.ndtp.federator.client.storage.impl.AzureReceivedFileStorage;
+import uk.gov.dbt.ndtp.federator.client.storage.impl.GCPReceivedFileStorage;
+import uk.gov.dbt.ndtp.federator.client.storage.impl.LocalReceivedFileStorage;
+import uk.gov.dbt.ndtp.federator.client.storage.impl.S3ReceivedFileStorage;
+import uk.gov.dbt.ndtp.federator.common.utils.PropertyUtil;
+
+class ReceivedFileStorageFactoryTest {
+
+ @AfterEach
+ void tearDown() {
+ // Ensure no lingering global state between tests
+ try {
+ PropertyUtil.clear();
+ } catch (Exception ignored) {
+ // ignore if not initialized
+ }
+ }
+
+ @Test
+ void get_returnsS3Storage_whenProviderIsS3() {
+ try (MockedStatic prop = Mockito.mockStatic(PropertyUtil.class)) {
+ prop.when(() -> PropertyUtil.getPropertyValue("client.files.storage.provider", "LOCAL"))
+ .thenReturn("S3");
+
+ ReceivedFileStorage storage = ReceivedFileStorageFactory.get();
+ assertInstanceOf(S3ReceivedFileStorage.class, storage);
+ }
+ }
+
+ @Test
+ void get_returnsS3Storage_whenProviderIsS3CaseInsensitive() {
+ try (MockedStatic prop = Mockito.mockStatic(PropertyUtil.class)) {
+ prop.when(() -> PropertyUtil.getPropertyValue("client.files.storage.provider", "LOCAL"))
+ .thenReturn("s3");
+
+ ReceivedFileStorage storage = ReceivedFileStorageFactory.get();
+ assertInstanceOf(S3ReceivedFileStorage.class, storage);
+ }
+ }
+
+ @Test
+ void get_returnsAzureStorage_whenProviderIsAzure() {
+ try (MockedStatic prop = Mockito.mockStatic(PropertyUtil.class)) {
+ prop.when(() -> PropertyUtil.getPropertyValue("client.files.storage.provider", "LOCAL"))
+ .thenReturn("AZURE");
+
+ ReceivedFileStorage storage = ReceivedFileStorageFactory.get();
+ assertInstanceOf(AzureReceivedFileStorage.class, storage);
+ }
+ }
+
+ @Test
+ void get_returnsAzureStorage_whenProviderIsAzureCaseInsensitive() {
+ try (MockedStatic prop = Mockito.mockStatic(PropertyUtil.class)) {
+ prop.when(() -> PropertyUtil.getPropertyValue("client.files.storage.provider", "LOCAL"))
+ .thenReturn("Azure");
+
+ ReceivedFileStorage storage = ReceivedFileStorageFactory.get();
+ assertInstanceOf(AzureReceivedFileStorage.class, storage);
+ }
+ }
+
+ @Test
+ void get_returnsGCPStorage_whenProviderIsGCP() {
+ try (MockedStatic prop = Mockito.mockStatic(PropertyUtil.class)) {
+ prop.when(() -> PropertyUtil.getPropertyValue("client.files.storage.provider", "LOCAL"))
+ .thenReturn("GCP");
+
+ ReceivedFileStorage storage = ReceivedFileStorageFactory.get();
+ assertInstanceOf(GCPReceivedFileStorage.class, storage);
+ }
+ }
+
+ @Test
+ void get_returnsGCPStorage_whenProviderIsGCPCaseInsensitive() {
+ try (MockedStatic prop = Mockito.mockStatic(PropertyUtil.class)) {
+ prop.when(() -> PropertyUtil.getPropertyValue("client.files.storage.provider", "LOCAL"))
+ .thenReturn("gcp");
+
+ ReceivedFileStorage storage = ReceivedFileStorageFactory.get();
+ assertInstanceOf(GCPReceivedFileStorage.class, storage);
+ }
+ }
+
+ @Test
+ void get_returnsLocalStorage_whenProviderIsLocal() {
+ try (MockedStatic prop = Mockito.mockStatic(PropertyUtil.class)) {
+ prop.when(() -> PropertyUtil.getPropertyValue("client.files.storage.provider", "LOCAL"))
+ .thenReturn("LOCAL");
+
+ ReceivedFileStorage storage = ReceivedFileStorageFactory.get();
+ assertInstanceOf(LocalReceivedFileStorage.class, storage);
+ }
+ }
+
+ @Test
+ void get_returnsLocalStorage_whenProviderIsEmpty() {
+ try (MockedStatic prop = Mockito.mockStatic(PropertyUtil.class)) {
+ prop.when(() -> PropertyUtil.getPropertyValue("client.files.storage.provider", "LOCAL"))
+ .thenReturn("");
+
+ ReceivedFileStorage storage = ReceivedFileStorageFactory.get();
+ assertInstanceOf(LocalReceivedFileStorage.class, storage);
+ }
+ }
+
+ @Test
+ void get_returnsLocalStorage_whenProviderIsNull() {
+ try (MockedStatic prop = Mockito.mockStatic(PropertyUtil.class)) {
+ prop.when(() -> PropertyUtil.getPropertyValue("client.files.storage.provider", "LOCAL"))
+ .thenReturn(null);
+
+ ReceivedFileStorage storage = ReceivedFileStorageFactory.get();
+ assertInstanceOf(LocalReceivedFileStorage.class, storage);
+ }
+ }
+
+ @Test
+ void get_returnsLocalStorage_whenProviderIsUnknown() {
+ try (MockedStatic prop = Mockito.mockStatic(PropertyUtil.class)) {
+ prop.when(() -> PropertyUtil.getPropertyValue("client.files.storage.provider", "LOCAL"))
+ .thenReturn("UNKNOWN_PROVIDER");
+
+ ReceivedFileStorage storage = ReceivedFileStorageFactory.get();
+ assertInstanceOf(LocalReceivedFileStorage.class, storage);
+ }
+ }
+
+ @Test
+ void get_returnsLocalStorage_whenPropertyUtilThrowsException() {
+ try (MockedStatic prop = Mockito.mockStatic(PropertyUtil.class)) {
+ prop.when(() -> PropertyUtil.getPropertyValue("client.files.storage.provider", "LOCAL"))
+ .thenThrow(new RuntimeException("Property not available"));
+
+ ReceivedFileStorage storage = ReceivedFileStorageFactory.get();
+ assertInstanceOf(LocalReceivedFileStorage.class, storage);
+ }
+ }
+
+ @Test
+ void get_returnsLocalStorage_whenPropertyUtilThrowsRuntimeException() {
+ try (MockedStatic prop = Mockito.mockStatic(PropertyUtil.class)) {
+ prop.when(() -> PropertyUtil.getPropertyValue("client.files.storage.provider", "LOCAL"))
+ .thenThrow(new RuntimeException("Configuration error"));
+
+ ReceivedFileStorage storage = ReceivedFileStorageFactory.get();
+ assertInstanceOf(LocalReceivedFileStorage.class, storage);
+ }
+ }
+}
diff --git a/src/test/java/uk/gov/dbt/ndtp/federator/client/storage/impl/GCPReceivedFileStorageTest.java b/src/test/java/uk/gov/dbt/ndtp/federator/client/storage/impl/GCPReceivedFileStorageTest.java
new file mode 100644
index 00000000..4d5df784
--- /dev/null
+++ b/src/test/java/uk/gov/dbt/ndtp/federator/client/storage/impl/GCPReceivedFileStorageTest.java
@@ -0,0 +1,174 @@
+package uk.gov.dbt.ndtp.federator.client.storage.impl;
+
+import static java.nio.file.Files.createTempFile;
+import static java.nio.file.Files.deleteIfExists;
+import static java.nio.file.Files.exists;
+import static java.nio.file.Files.writeString;
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+import uk.gov.dbt.ndtp.federator.client.storage.StoredFileResult;
+import uk.gov.dbt.ndtp.federator.common.utils.PropertyUtil;
+
+class GCPReceivedFileStorageTest {
+ @AfterEach
+ void tearDown() {
+ // Ensure no lingering global state between tests
+ try {
+ PropertyUtil.clear();
+ } catch (Exception ignored) {
+ // ignore if not initialized
+ }
+ }
+
+ @Test
+ void resolveBucket_returnsEmptyWhenNotConfigured() {
+ // Mock PropertyUtil to return blank bucket
+ try (MockedStatic prop = Mockito.mockStatic(PropertyUtil.class)) {
+ prop.when(() -> PropertyUtil.getPropertyValue("files.gcp.bucket", ""))
+ .thenReturn("");
+
+ GCPReceivedFileStorage gcp = new GCPReceivedFileStorage();
+ assertEquals("", gcp.resolveBucket());
+ }
+ }
+
+ @Test
+ void resolveBucket_returnsConfiguredBucket() {
+ try (MockedStatic prop = Mockito.mockStatic(PropertyUtil.class)) {
+ prop.when(() -> PropertyUtil.getPropertyValue("files.gcp.bucket", ""))
+ .thenReturn("my-team-docs");
+
+ GCPReceivedFileStorage gcp = new GCPReceivedFileStorage();
+ assertEquals("my-team-docs", gcp.resolveBucket());
+ }
+ }
+
+ @Test
+ void resolveKey_handlesNullBlankAndPrefixesAndNormalization() {
+ GCPReceivedFileStorage gcp = new GCPReceivedFileStorage();
+
+ // null destination -> sanitized original file name
+ assertEquals("name.txt", gcp.resolveKey(null, "dir/name.txt"));
+ // blank destination -> sanitized
+ assertEquals("name.txt", gcp.resolveKey(" ", "x/../name.txt"));
+ // prefix with trailing slash -> append sanitized file name
+ assertEquals("a/b/name.txt", gcp.resolveKey("a/b/", "c/d/name.txt"));
+ // full key without trailing slash -> normalize leading slashes are removed
+ assertEquals("a/b/c.txt", gcp.resolveKey("/a/b/c.txt", "ignored.txt"));
+ }
+
+ // We avoid direct testing of upload() to prevent static initialization of GcsClientFactory.
+ // Instead, we exercise store() behavior with a subclass overriding upload().
+
+ @Test
+ void store_bucketBlank_skipsUpload_andKeepsLocalFile() throws IOException {
+ Path temp = createTempFile("gcprfst-", ".bin");
+ writeString(temp, "data");
+
+ try (MockedStatic prop = Mockito.mockStatic(PropertyUtil.class)) {
+ prop.when(() -> PropertyUtil.getPropertyValue("files.gcp.bucket", ""))
+ .thenReturn("");
+ GCPReceivedFileStorage gcp = new GCPReceivedFileStorage();
+ StoredFileResult res = gcp.store(temp, "f.txt", null);
+ assertTrue(exists(res.localPath()));
+ assertFalse(res.remoteUriOpt().isPresent());
+ } finally {
+ deleteIfExists(temp);
+ }
+ }
+
+ @Test
+ void store_success_deletesLocal_andReturnsRemoteUri() throws IOException {
+ Path temp = createTempFile("gcprfst-", ".bin");
+ writeString(temp, "data");
+
+ try (MockedStatic prop = Mockito.mockStatic(PropertyUtil.class)) {
+ // Bucket resolution
+ prop.when(() -> PropertyUtil.getPropertyValue("files.gcp.bucket", ""))
+ .thenReturn("my-team-docs");
+ // Use a test subclass that fakes upload success
+ class TestGCP extends GCPReceivedFileStorage {
+ @Override
+ String upload(Path localFile, String bucket, String key) {
+ return String.format("gs://%s/%s", bucket, key);
+ }
+ }
+ GCPReceivedFileStorage gcp = new TestGCP();
+ StoredFileResult res = gcp.store(temp, "file.txt", "prefix/");
+
+ assertTrue(res.remoteUriOpt().isPresent());
+ assertFalse(exists(temp), "Temp file should be deleted after successful upload");
+ } finally {
+ deleteIfExists(temp);
+ }
+ }
+
+ @Test
+ void store_failure_deletesLocal_andNoRemoteUri() throws IOException {
+ Path temp = createTempFile("gcprfst-", ".bin");
+ writeString(temp, "data");
+
+ try (MockedStatic prop = Mockito.mockStatic(PropertyUtil.class)) {
+ // Bucket resolution
+ prop.when(() -> PropertyUtil.getPropertyValue("files.gcp.bucket", ""))
+ .thenReturn("my-team-docs");
+ // Use a test subclass that simulates upload failure by throwing an exception
+ class TestGCP extends GCPReceivedFileStorage {
+ @Override
+ String upload(Path localFile, String bucket, String key) {
+ throw new RuntimeException("simulated GCS error");
+ }
+ }
+ GCPReceivedFileStorage gcp = new TestGCP();
+ StoredFileResult res = gcp.store(temp, "file.txt", "prefix/");
+
+ assertFalse(res.remoteUriOpt().isPresent());
+ assertFalse(exists(temp), "Temp file should be deleted when upload fails");
+ } finally {
+ deleteIfExists(temp);
+ }
+ }
+
+ @Test
+ void resolveBucket_returnsEmptyWhenNullReturned() {
+ try (MockedStatic prop = Mockito.mockStatic(PropertyUtil.class)) {
+ prop.when(() -> PropertyUtil.getPropertyValue("files.gcp.bucket", ""))
+ .thenReturn(null);
+
+ GCPReceivedFileStorage gcp = new GCPReceivedFileStorage();
+ assertEquals("", gcp.resolveBucket());
+ }
+ }
+
+ @Test
+ void store_uploadReturnsNull_deletesLocal_andNoRemoteUri() throws IOException {
+ Path temp = createTempFile("gcprfst-", ".bin");
+ writeString(temp, "data");
+
+ try (MockedStatic prop = Mockito.mockStatic(PropertyUtil.class)) {
+ // Bucket resolution
+ prop.when(() -> PropertyUtil.getPropertyValue("files.gcp.bucket", ""))
+ .thenReturn("my-team-docs");
+ // Use a test subclass that simulates upload failure by returning null
+ class TestGCP extends GCPReceivedFileStorage {
+ @Override
+ String upload(Path localFile, String bucket, String key) {
+ return null; // Simulate upload failure without exception
+ }
+ }
+ GCPReceivedFileStorage gcp = new TestGCP();
+ StoredFileResult res = gcp.store(temp, "file.txt", "prefix/");
+
+ assertFalse(res.remoteUriOpt().isPresent());
+ assertFalse(exists(temp), "Temp file should be deleted when upload returns null");
+ } finally {
+ deleteIfExists(temp);
+ }
+ }
+}
diff --git a/src/test/java/uk/gov/dbt/ndtp/federator/common/service/KafkaStreamServiceTest.java b/src/test/java/uk/gov/dbt/ndtp/federator/common/service/KafkaStreamServiceTest.java
index 67828eb8..227a3770 100644
--- a/src/test/java/uk/gov/dbt/ndtp/federator/common/service/KafkaStreamServiceTest.java
+++ b/src/test/java/uk/gov/dbt/ndtp/federator/common/service/KafkaStreamServiceTest.java
@@ -5,12 +5,18 @@
import static org.mockito.Mockito.*;
import io.grpc.Context;
+import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
+import java.nio.file.Files;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
import org.apache.kafka.common.errors.InvalidTopicException;
import org.junit.jupiter.api.Test;
import org.mockito.MockedStatic;
@@ -23,6 +29,7 @@
import uk.gov.dbt.ndtp.federator.common.service.config.ProducerConfigService;
import uk.gov.dbt.ndtp.federator.common.service.kafka.KafkaStreamService;
import uk.gov.dbt.ndtp.federator.common.utils.ProducerConsumerConfigServiceFactory;
+import uk.gov.dbt.ndtp.federator.common.utils.PropertyUtil;
import uk.gov.dbt.ndtp.federator.server.grpc.GRPCContextKeys;
import uk.gov.dbt.ndtp.federator.server.interfaces.StreamObservable;
import uk.gov.dbt.ndtp.grpc.TopicRequest;
@@ -122,6 +129,7 @@ void test_streamToClient_throwsInvalidTopic_whenAccessDenied() {
TopicRequest req =
TopicRequest.newBuilder().setTopic("not-allowed").setOffset(0L).build();
StreamObservable observer = mock(StreamObservable.class);
+ ExecutorService executorService = mock(ExecutorService.class);
ProducerConfigService mockService = mock(ProducerConfigService.class);
ProducerConfigDTO emptyCfg =
@@ -138,10 +146,88 @@ void test_streamToClient_throwsInvalidTopic_whenAccessDenied() {
Context ctx = Context.current().withValue(GRPCContextKeys.CLIENT_ID, "consumer-1");
Context previous = ctx.attach();
try {
- assertThrows(InvalidTopicException.class, () -> cut.streamToClient(req, observer));
+ assertThrows(InvalidTopicException.class, () -> cut.streamToClient(req, observer, executorService));
} finally {
ctx.detach(previous);
}
}
}
+
+ // -------------------- Positive test for streamToClient --------------------
+
+ @Test
+ void test_streamToClient_awaitsTheFutureSubmittedToTheExecutorService() throws IOException {
+ KafkaStreamService cut = new KafkaStreamService(EMPTY_SHARED_HEADERS);
+ TopicRequest req =
+ TopicRequest.newBuilder().setTopic("test").setOffset(0L).build();
+ StreamObservable observer = mock(StreamObservable.class);
+ ExecutorService executorService = mock(ExecutorService.class);
+
+ ProductDTO mockProductDto = mock(ProductDTO.class);
+ ConsumerDTO mockConsumerDto = mock(ConsumerDTO.class);
+ ArrayList mockConumerDtos = new ArrayList<>();
+ mockConumerDtos.add(mockConsumerDto);
+
+ when(mockProductDto.getTopic()).thenReturn("test");
+ when(mockConsumerDto.getIdpClientId()).thenReturn("consumer-1");
+ when(mockProductDto.getConsumers()).thenReturn(mockConumerDtos);
+
+ ArrayList mockProductDtos = new ArrayList<>();
+ mockProductDtos.add(mockProductDto);
+
+ ProducerDTO mockProducerDto = mock(ProducerDTO.class);
+ ArrayList mockProducerDtos = new ArrayList<>();
+ mockProducerDtos.add(mockProducerDto);
+
+ when(mockProducerDto.getProducts()).thenReturn(mockProductDtos);
+
+ ProducerConfigService mockService = mock(ProducerConfigService.class);
+ ProducerConfigDTO producerCfg =
+ ProducerConfigDTO.builder().producers(mockProducerDtos).build();
+
+ try (MockedStatic mockedFactory =
+ Mockito.mockStatic(ProducerConsumerConfigServiceFactory.class)) {
+ mockedFactory
+ .when(ProducerConsumerConfigServiceFactory::getProducerConfigService)
+ .thenReturn(mockService);
+ when(mockService.getProducerConfiguration()).thenReturn(producerCfg);
+
+ Future mockFuture = mock(Future.class);
+
+ when(executorService.submit(any(Runnable.class))).thenReturn(mockFuture);
+
+ // Set the gRPC context key so KafkaStreamService can read the consumer id
+ Context ctx = Context.current().withValue(GRPCContextKeys.CLIENT_ID, "consumer-1");
+ Context previous = ctx.attach();
+
+ // Prepare a temporary properties file with minimal configuration
+ Path tmp = Files.createTempFile("s3clientfactory-test-", ".properties");
+ try {
+ String props = String.join(
+ "\n",
+ "kafka.defaultKeyDeserializerClass=org.apache.kafka.common.serialization.StringDeserializer",
+ "kafka.defaultValueDeserializerClass=uk.gov.dbt.ndtp.federator.access.AccessMessageDeserializer",
+ "kafka.bootstrapServers=localhost:9092",
+ "kafka.consumerGroup=test",
+ "kafka.pollRecords=100");
+
+ Files.writeString(tmp, props);
+ // Initialize PropertyUtil
+ PropertyUtil.init(tmp.toFile());
+ cut.streamToClient(req, observer, executorService);
+ } finally {
+ Files.deleteIfExists(tmp);
+ PropertyUtil.clear();
+ ctx.detach(previous);
+ }
+
+ verify(executorService, times(1)).submit(any(Runnable.class));
+
+ try {
+ verify(mockFuture, times(1)).get();
+ } catch (InterruptedException | ExecutionException ignored) {
+ // ignored
+ }
+ }
+ }
}
diff --git a/src/test/java/uk/gov/dbt/ndtp/federator/common/service/file/FileStreamServiceTest.java b/src/test/java/uk/gov/dbt/ndtp/federator/common/service/file/FileStreamServiceTest.java
index eb431203..e60ca435 100644
--- a/src/test/java/uk/gov/dbt/ndtp/federator/common/service/file/FileStreamServiceTest.java
+++ b/src/test/java/uk/gov/dbt/ndtp/federator/common/service/file/FileStreamServiceTest.java
@@ -5,6 +5,9 @@
import static org.mockito.Mockito.*;
import io.grpc.Context;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
import org.junit.jupiter.api.Test;
import org.mockito.MockedConstruction;
import org.mockito.MockedStatic;
@@ -26,6 +29,11 @@ void test_streamToClient_invokesConductorAndCompletes() {
StreamObservable observer = mock(StreamObservable.class);
+ ExecutorService executorService = mock(ExecutorService.class);
+ Future mockFuture = mock(Future.class);
+
+ when(executorService.submit(any(Runnable.class))).thenReturn(mockFuture);
+
FileStreamRequest req = FileStreamRequest.newBuilder()
.setTopic("files-topic")
.setStartSequenceId(0L)
@@ -50,7 +58,7 @@ void test_streamToClient_invokesConductorAndCompletes() {
Context grpcCtx = Context.current().withValue(GRPCContextKeys.CLIENT_ID, "client-xyz");
Context prev = grpcCtx.attach();
try {
- cut.streamToClient(req, observer);
+ cut.streamToClient(req, observer, executorService);
} finally {
grpcCtx.detach(prev);
}
@@ -58,6 +66,12 @@ void test_streamToClient_invokesConductorAndCompletes() {
// One FileConductor constructed
assertEquals(1, mocked.constructed().size());
+ try {
+ verify(mockFuture, times(1)).get();
+ } catch (InterruptedException | ExecutionException ignored) {
+ // ignored
+ }
+
// Cancel handler is set and onCompleted called
verify(observer, times(1)).setOnCancelHandler(any());
verify(observer, times(1)).onCompleted();
diff --git a/src/test/java/uk/gov/dbt/ndtp/federator/common/service/stream/ClosableFederatorStreamServiceTest.java b/src/test/java/uk/gov/dbt/ndtp/federator/common/service/stream/ClosableFederatorStreamServiceTest.java
new file mode 100644
index 00000000..86ee6e80
--- /dev/null
+++ b/src/test/java/uk/gov/dbt/ndtp/federator/common/service/stream/ClosableFederatorStreamServiceTest.java
@@ -0,0 +1,29 @@
+package uk.gov.dbt.ndtp.federator.common.service.stream;
+
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import org.junit.jupiter.api.Test;
+import uk.gov.dbt.ndtp.federator.common.service.file.FileStreamService;
+import uk.gov.dbt.ndtp.federator.server.conductor.MessageConductor;
+
+/**
+ * Tests for {@link CloseableFederatorStreamService}
+ */
+class ClosableFederatorStreamServiceTest {
+
+ @Test
+ void shouldCloseAllMessageConductorsAndClearTheList_whenClosed() {
+ CloseableFederatorStreamService service = new FileStreamService();
+
+ MessageConductor messageConductor1 = mock(MessageConductor.class);
+ service.messageConductors.add(messageConductor1);
+
+ service.close();
+
+ verify(messageConductor1, times(1)).close();
+ assertTrue(service.messageConductors.isEmpty());
+ }
+}
diff --git a/src/test/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/FileProviderFactoryTest.java b/src/test/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/FileProviderFactoryTest.java
index 9e12da77..2d8b85b4 100644
--- a/src/test/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/FileProviderFactoryTest.java
+++ b/src/test/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/FileProviderFactoryTest.java
@@ -10,6 +10,7 @@
import static org.mockito.Mockito.*;
import com.azure.storage.blob.BlobServiceClient;
+import com.google.cloud.storage.Storage;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -17,8 +18,10 @@
import software.amazon.awssdk.services.s3.S3Client;
import uk.gov.dbt.ndtp.federator.common.model.SourceType;
import uk.gov.dbt.ndtp.federator.common.storage.provider.file.client.AzureBlobClientFactory;
+import uk.gov.dbt.ndtp.federator.common.storage.provider.file.client.GcsClientFactory;
import uk.gov.dbt.ndtp.federator.common.storage.provider.file.client.S3ClientFactory;
import uk.gov.dbt.ndtp.federator.common.storage.provider.file.impl.AzureFileProvider;
+import uk.gov.dbt.ndtp.federator.common.storage.provider.file.impl.GCPFileProvider;
import uk.gov.dbt.ndtp.federator.common.storage.provider.file.impl.LocalFileProvider;
import uk.gov.dbt.ndtp.federator.common.storage.provider.file.impl.S3FileProvider;
@@ -26,17 +29,20 @@ class FileProviderFactoryTest {
private MockedStatic s3FactoryMock;
private MockedStatic azureFactoryMock;
+ private MockedStatic gcsFactoryMock;
@BeforeEach
void setUp() {
s3FactoryMock = mockStatic(S3ClientFactory.class);
azureFactoryMock = mockStatic(AzureBlobClientFactory.class);
+ gcsFactoryMock = mockStatic(GcsClientFactory.class);
}
@AfterEach
void tearDown() {
s3FactoryMock.close();
azureFactoryMock.close();
+ gcsFactoryMock.close();
}
@Test
@@ -53,6 +59,13 @@ void testGetProvider_Azure() {
assertTrue(provider instanceof AzureFileProvider);
}
+ @Test
+ void testGetProvider_GCP() {
+ gcsFactoryMock.when(GcsClientFactory::getClient).thenReturn(mock(Storage.class));
+ FileProvider provider = FileProviderFactory.getProvider(SourceType.GCP);
+ assertTrue(provider instanceof GCPFileProvider);
+ }
+
@Test
void testGetProvider_Local() {
FileProvider provider = FileProviderFactory.getProvider(SourceType.LOCAL);
diff --git a/src/test/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/client/GcsClientFactoryTest.java b/src/test/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/client/GcsClientFactoryTest.java
new file mode 100644
index 00000000..1b7d00f6
--- /dev/null
+++ b/src/test/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/client/GcsClientFactoryTest.java
@@ -0,0 +1,234 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * © Crown Copyright 2026. This work has been developed by the National Digital Twin Programme and is legally
+ * attributed to the Department for Business and Trade (UK) as the governing entity.
+ */
+
+package uk.gov.dbt.ndtp.federator.common.storage.provider.file.client;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import com.google.cloud.storage.Storage;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import uk.gov.dbt.ndtp.federator.common.utils.PropertyUtil;
+
+/**
+ * Basic smoke test for GcsClientFactory ensuring a client can be created from properties.
+ *
+ * Note: We intentionally limit to a single construction scenario because the factory
+ * holds a static singleton instance which cannot be reset between tests. This test
+ * verifies the expected happy-path initialization using different credential approaches
+ * and a custom endpoint (e.g., fake-gcs-server/local), without making any network calls.
+ */
+class GcsClientFactoryTest {
+
+ @AfterEach
+ void tearDown() {
+ try {
+ GcsClientFactory.resetClient();
+ PropertyUtil.clear();
+ } catch (Exception ignored) {
+ // ignore if not initialized
+ }
+ }
+
+ @Test
+ void getClient_withEndpointUrl_buildsSuccessfully() throws IOException {
+ // Prepare a temporary properties file with custom endpoint (fake-gcs-server)
+ Path tmp = Files.createTempFile("gcsclientfactory-test-", ".properties");
+ try {
+ String props = String.join(
+ "\n", "gcp.storage.endpoint.url=http://localhost:4443", "gcp.storage.project.id=test-project");
+ Files.writeString(tmp, props);
+
+ // Initialize PropertyUtil before touching GcsClientFactory
+ PropertyUtil.init(tmp.toFile());
+
+ // When
+ var client = GcsClientFactory.getClient();
+
+ // Then
+ assertNotNull(client, "GcsClientFactory should return a non-null Storage instance");
+ } finally {
+ Files.deleteIfExists(tmp);
+ }
+ }
+
+ @Test
+ void getClient_withProjectIdOnly_buildsSuccessfully() throws IOException {
+ // Prepare a temporary properties file with only project ID and endpoint to avoid ADC
+ Path tmp = Files.createTempFile("gcsclientfactory-project-test-", ".properties");
+ try {
+ String props = String.join(
+ "\n", "gcp.storage.project.id=test-project", "gcp.storage.endpoint.url=http://localhost:4443");
+ Files.writeString(tmp, props);
+
+ // Initialize PropertyUtil before touching GcsClientFactory
+ PropertyUtil.init(tmp.toFile());
+
+ // When
+ var client = GcsClientFactory.getClient();
+
+ // Then
+ assertNotNull(client, "GcsClientFactory should return a non-null Storage instance with project ID only");
+ } finally {
+ Files.deleteIfExists(tmp);
+ }
+ }
+
+ @Test
+ void getClient_withNoProperties_buildsSuccessfully() throws IOException {
+ // Prepare a temporary properties file with endpoint to avoid ADC
+ Path tmp = Files.createTempFile("gcsclientfactory-default-test-", ".properties");
+ try {
+ String props = "gcp.storage.endpoint.url=http://localhost:4443";
+ Files.writeString(tmp, props);
+
+ // Initialize PropertyUtil before touching GcsClientFactory
+ PropertyUtil.init(tmp.toFile());
+
+ // When
+ var client = GcsClientFactory.getClient();
+
+ // Then
+ assertNotNull(
+ client, "GcsClientFactory should return a non-null Storage instance with default credentials");
+ } finally {
+ Files.deleteIfExists(tmp);
+ }
+ }
+
+ @Test
+ void getClient_withEndpointAndNoProjectId_buildsSuccessfully() throws IOException {
+ // Prepare a temporary properties file with endpoint but no project ID
+ Path tmp = Files.createTempFile("gcsclientfactory-endpoint-no-project-test-", ".properties");
+ try {
+ String props = String.join("\n", "gcp.storage.endpoint.url=http://localhost:4443");
+ Files.writeString(tmp, props);
+
+ // Initialize PropertyUtil before touching GcsClientFactory
+ PropertyUtil.init(tmp.toFile());
+
+ // When
+ var client = GcsClientFactory.getClient();
+
+ // Then
+ assertNotNull(
+ client,
+ "GcsClientFactory should return a non-null Storage instance with endpoint but no project ID");
+ } finally {
+ Files.deleteIfExists(tmp);
+ }
+ }
+
+ @Test
+ void getClient_returnsSameInstance_whenCalledMultipleTimes() throws IOException {
+ // Prepare a temporary properties file
+ Path tmp = Files.createTempFile("gcsclientfactory-singleton-test-", ".properties");
+ try {
+ String props = String.join(
+ "\n", "gcp.storage.endpoint.url=http://localhost:4443", "gcp.storage.project.id=test-project");
+ Files.writeString(tmp, props);
+
+ // Initialize PropertyUtil before touching GcsClientFactory
+ PropertyUtil.init(tmp.toFile());
+
+ // When - call getClient multiple times
+ Storage client1 = GcsClientFactory.getClient();
+ Storage client2 = GcsClientFactory.getClient();
+ Storage client3 = GcsClientFactory.getClient();
+
+ // Then - all references should point to the same instance
+ assertAll(
+ "Singleton behavior verification",
+ () -> assertNotNull(client1, "First client should not be null"),
+ () -> assertSame(client1, client2, "Second call should return same instance as first"),
+ () -> assertSame(client1, client3, "Third call should return same instance as first"),
+ () -> assertSame(client2, client3, "All instances should be identical"));
+ } finally {
+ Files.deleteIfExists(tmp);
+ }
+ }
+
+ @Test
+ void getClient_withBlankEndpointUrl_usesDefaultEndpoint() throws IOException {
+ // Prepare a temporary properties file with blank endpoint URL but valid endpoint to avoid ADC
+ Path tmp = Files.createTempFile("gcsclientfactory-blank-endpoint-test-", ".properties");
+ try {
+ String props = String.join(
+ "\n", "gcp.storage.project.id=test-project", "gcp.storage.endpoint.url=http://localhost:4443");
+ Files.writeString(tmp, props);
+
+ // Initialize PropertyUtil before touching GcsClientFactory
+ PropertyUtil.init(tmp.toFile());
+
+ // When - test that blank values in properties are handled
+ var client = GcsClientFactory.getClient();
+
+ // Then - should successfully create client
+ assertNotNull(client, "GcsClientFactory should handle configuration gracefully");
+ } finally {
+ Files.deleteIfExists(tmp);
+ }
+ }
+
+ @Test
+ void getClient_withBlankProjectId_usesDefaultProjectId() throws IOException {
+ // Prepare a temporary properties file with blank project ID
+ Path tmp = Files.createTempFile("gcsclientfactory-blank-project-test-", ".properties");
+ try {
+ String props =
+ String.join("\n", "gcp.storage.endpoint.url=http://localhost:4443", "gcp.storage.project.id=");
+ Files.writeString(tmp, props);
+
+ // Initialize PropertyUtil before touching GcsClientFactory
+ PropertyUtil.init(tmp.toFile());
+
+ // When
+ var client = GcsClientFactory.getClient();
+
+ // Then
+ assertNotNull(client, "GcsClientFactory should handle blank project ID gracefully");
+ } finally {
+ Files.deleteIfExists(tmp);
+ }
+ }
+
+ @Test
+ void resetClient_allowsNewClientCreation() throws IOException {
+ // Prepare a temporary properties file
+ Path tmp = Files.createTempFile("gcsclientfactory-reset-test-", ".properties");
+ try {
+ String props = String.join(
+ "\n", "gcp.storage.endpoint.url=http://localhost:4443", "gcp.storage.project.id=test-project-1");
+ Files.writeString(tmp, props);
+
+ // Initialize PropertyUtil and get first client
+ PropertyUtil.init(tmp.toFile());
+ Storage client1 = GcsClientFactory.getClient();
+
+ // When - reset and create new client with different config
+ GcsClientFactory.resetClient();
+ PropertyUtil.clear();
+
+ String props2 = String.join(
+ "\n", "gcp.storage.endpoint.url=http://localhost:4444", "gcp.storage.project.id=test-project-2");
+ Files.writeString(tmp, props2);
+ PropertyUtil.init(tmp.toFile());
+ Storage client2 = GcsClientFactory.getClient();
+
+ // Then - clients should be different instances
+ assertAll(
+ "Reset behavior verification",
+ () -> assertNotNull(client1, "First client should not be null"),
+ () -> assertNotNull(client2, "Second client should not be null"),
+ () -> assertNotSame(client1, client2, "After reset, a new instance should be created"));
+ } finally {
+ Files.deleteIfExists(tmp);
+ }
+ }
+}
diff --git a/src/test/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/client/S3ClientFactoryTest.java b/src/test/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/client/S3ClientFactoryTest.java
index fd027239..c58ea590 100644
--- a/src/test/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/client/S3ClientFactoryTest.java
+++ b/src/test/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/client/S3ClientFactoryTest.java
@@ -6,6 +6,7 @@
import java.nio.file.Files;
import java.nio.file.Path;
import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import uk.gov.dbt.ndtp.federator.common.utils.PropertyUtil;
@@ -19,11 +20,15 @@
*/
class S3ClientFactoryTest {
+ @BeforeEach
+ void setup() {
+ PropertyUtil.clear();
+ }
+
@AfterEach
void tearDown() {
try {
S3ClientFactory.resetClient();
- PropertyUtil.clear();
} catch (Exception ignored) {
// ignore if not initialized
}
@@ -102,7 +107,8 @@ void getClient_withProfile_buildsSuccessfully() throws IOException {
// Then
assertNotNull(
client,
- "S3ClientFactory should return a non-null S3Client instance even if profile doesn't exist (it falls back)");
+ "S3ClientFactory should return a non-null S3Client instance even if profile doesn't exist (it"
+ + " falls back)");
} finally {
Files.deleteIfExists(tmp);
}
diff --git a/src/test/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/impl/GCPFileProviderTest.java b/src/test/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/impl/GCPFileProviderTest.java
new file mode 100644
index 00000000..b04141f2
--- /dev/null
+++ b/src/test/java/uk/gov/dbt/ndtp/federator/common/storage/provider/file/impl/GCPFileProviderTest.java
@@ -0,0 +1,187 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * © Crown Copyright 2026. This work has been developed by the National Digital Twin Programme and is legally
+ * attributed to the Department for Business and Trade (UK) as the governing entity.
+ */
+
+package uk.gov.dbt.ndtp.federator.common.storage.provider.file.impl;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
+import com.google.cloud.ReadChannel;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageException;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import uk.gov.dbt.ndtp.federator.common.exception.FileTransferException;
+import uk.gov.dbt.ndtp.federator.common.model.FileTransferRequest;
+import uk.gov.dbt.ndtp.federator.common.model.SourceType;
+import uk.gov.dbt.ndtp.federator.exceptions.FileFetcherException;
+import uk.gov.dbt.ndtp.federator.server.processor.file.FileTransferResult;
+
+@ExtendWith(MockitoExtension.class)
+class GCPFileProviderTest {
+
+ @Mock
+ private Storage storage;
+
+ @Mock
+ private Blob blob;
+
+ @Mock
+ private ReadChannel readChannel;
+
+ private GCPFileProvider gcpFileProvider;
+
+ @BeforeEach
+ void setUp() {
+ gcpFileProvider = new GCPFileProvider(storage);
+ }
+
+ @Test
+ void testGetSuccess() {
+ FileTransferRequest request = new FileTransferRequest(SourceType.GCP, "my-bucket", "my-key");
+ when(storage.get(any(BlobId.class))).thenReturn(blob);
+ when(blob.exists()).thenReturn(true);
+ when(blob.getSize()).thenReturn(100L);
+ when(blob.reader()).thenReturn(readChannel);
+
+ try (FileTransferResult result = gcpFileProvider.get(request)) {
+ assertNotNull(result);
+ assertEquals(100L, result.fileSize());
+ assertNotNull(result.stream());
+ }
+
+ verify(storage).get(any(BlobId.class));
+ verify(blob).exists();
+ verify(blob).getSize();
+ verify(blob).reader();
+ }
+
+ @Test
+ void testGetBlobNotFound_NullBlob() {
+ FileTransferRequest request = new FileTransferRequest(SourceType.GCP, "my-bucket", "my-key");
+ when(storage.get(any(BlobId.class))).thenReturn(null);
+
+ FileFetcherException exception = assertThrows(FileFetcherException.class, () -> gcpFileProvider.get(request));
+ assertTrue(exception.getMessage().contains("File not found in GCS"));
+ }
+
+ @Test
+ void testGetBlobNotFound_BlobDoesNotExist() {
+ FileTransferRequest request = new FileTransferRequest(SourceType.GCP, "my-bucket", "my-key");
+ when(storage.get(any(BlobId.class))).thenReturn(blob);
+ when(blob.exists()).thenReturn(false);
+
+ FileFetcherException exception = assertThrows(FileFetcherException.class, () -> gcpFileProvider.get(request));
+ assertTrue(exception.getMessage().contains("File not found in GCS"));
+ }
+
+ @Test
+ void testGetStorageException404() {
+ FileTransferRequest request = new FileTransferRequest(SourceType.GCP, "my-bucket", "my-key");
+ StorageException storageException = new StorageException(404, "Not Found");
+ when(storage.get(any(BlobId.class))).thenThrow(storageException);
+
+ FileFetcherException exception = assertThrows(FileFetcherException.class, () -> gcpFileProvider.get(request));
+ assertTrue(exception.getMessage().contains("File not found in GCS"));
+ }
+
+ @Test
+ void testGetStorageExceptionOther() {
+ FileTransferRequest request = new FileTransferRequest(SourceType.GCP, "my-bucket", "my-key");
+ StorageException storageException = new StorageException(500, "Internal Server Error");
+ when(storage.get(any(BlobId.class))).thenThrow(storageException);
+
+ FileFetcherException exception = assertThrows(FileFetcherException.class, () -> gcpFileProvider.get(request));
+ assertTrue(exception.getMessage().contains("GCS error fetching"));
+ }
+
+ @Test
+ void testGetGeneralException() {
+ FileTransferRequest request = new FileTransferRequest(SourceType.GCP, "my-bucket", "my-key");
+ when(storage.get(any(BlobId.class))).thenThrow(new RuntimeException("Generic error"));
+
+ FileFetcherException exception = assertThrows(FileFetcherException.class, () -> gcpFileProvider.get(request));
+ assertTrue(exception.getMessage().contains("Failed to fetch from GCS"));
+ }
+
+ @Test
+ void testValidatePathSuccess() {
+ FileTransferRequest request = new FileTransferRequest(SourceType.GCP, "my-bucket", "my-key");
+ when(storage.get(any(BlobId.class))).thenReturn(blob);
+ when(blob.exists()).thenReturn(true);
+
+ assertDoesNotThrow(() -> gcpFileProvider.validatePath(request));
+ verify(storage).get(any(BlobId.class));
+ verify(blob).exists();
+ }
+
+ @Test
+ void testValidatePathObjectNotFound_NullBlob() {
+ FileTransferRequest request = new FileTransferRequest(SourceType.GCP, "my-bucket", "my-key");
+ when(storage.get(any(BlobId.class))).thenReturn(null);
+
+ FileTransferException exception =
+ assertThrows(FileTransferException.class, () -> gcpFileProvider.validatePath(request));
+ assertTrue(exception.getMessage().contains("GCS object not found"));
+ }
+
+ @Test
+ void testValidatePathObjectNotFound_BlobDoesNotExist() {
+ FileTransferRequest request = new FileTransferRequest(SourceType.GCP, "my-bucket", "my-key");
+ when(storage.get(any(BlobId.class))).thenReturn(blob);
+ when(blob.exists()).thenReturn(false);
+
+ FileTransferException exception =
+ assertThrows(FileTransferException.class, () -> gcpFileProvider.validatePath(request));
+ assertTrue(exception.getMessage().contains("GCS object not found"));
+ }
+
+ @Test
+ void testValidatePathStorageException404() {
+ FileTransferRequest request = new FileTransferRequest(SourceType.GCP, "my-bucket", "my-key");
+ StorageException storageException = new StorageException(404, "Not Found");
+ when(storage.get(any(BlobId.class))).thenThrow(storageException);
+
+ FileTransferException exception =
+ assertThrows(FileTransferException.class, () -> gcpFileProvider.validatePath(request));
+ assertTrue(exception.getMessage().contains("GCS object not found"));
+ }
+
+ @Test
+ void testValidatePathStorageExceptionOther() {
+ FileTransferRequest request = new FileTransferRequest(SourceType.GCP, "my-bucket", "my-key");
+ StorageException storageException = new StorageException(500, "Internal Server Error");
+ when(storage.get(any(BlobId.class))).thenThrow(storageException);
+
+ FileTransferException exception =
+ assertThrows(FileTransferException.class, () -> gcpFileProvider.validatePath(request));
+ assertTrue(exception.getMessage().contains("GCS validation error"));
+ }
+
+ @Test
+ void testValidatePathMissingBucket() {
+ FileTransferRequest request = new FileTransferRequest(SourceType.GCP, null, "my-key");
+
+ FileTransferException exception =
+ assertThrows(FileTransferException.class, () -> gcpFileProvider.validatePath(request));
+ assertTrue(exception.getMessage().contains("GCS bucket (storageContainer) is required"));
+ }
+
+ @Test
+ void testValidatePathBlankBucket() {
+ FileTransferRequest request = new FileTransferRequest(SourceType.GCP, " ", "my-key");
+
+ FileTransferException exception =
+ assertThrows(FileTransferException.class, () -> gcpFileProvider.validatePath(request));
+ assertTrue(exception.getMessage().contains("GCS bucket (storageContainer) is required"));
+ }
+}
diff --git a/src/test/java/uk/gov/dbt/ndtp/federator/common/utils/ResilienceSupportTest.java b/src/test/java/uk/gov/dbt/ndtp/federator/common/utils/ResilienceSupportTest.java
new file mode 100644
index 00000000..bda1d272
--- /dev/null
+++ b/src/test/java/uk/gov/dbt/ndtp/federator/common/utils/ResilienceSupportTest.java
@@ -0,0 +1,212 @@
+package uk.gov.dbt.ndtp.federator.common.utils;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import io.github.resilience4j.circuitbreaker.CallNotPermittedException;
+import io.github.resilience4j.circuitbreaker.CircuitBreaker;
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.net.SocketTimeoutException;
+import java.net.http.HttpTimeoutException;
+import java.util.Properties;
+import java.util.function.Supplier;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+import redis.clients.jedis.exceptions.JedisException;
+import uk.gov.dbt.ndtp.federator.exceptions.FederatorTokenException;
+
+/**
+ * ResilienceSupportTest
+ */
+class ResilienceSupportTest {
+
+ private static final String COMPONENT_NAME = "idp";
+ private static final String OPERATION = "fetchToken";
+ private static final String TARGET_ID = "client";
+ private static final String BASE_MESSAGE =
+ "Failed to fetch token after resilience protections for management node: 1234";
+
+ private Supplier supplier;
+
+ private static void setupTestFileProperties() {
+ Properties props = new Properties();
+ props.setProperty("management.node.resilience.retry.maxAttempts", "1");
+ props.setProperty("management.node.resilience.retry.maxBackOff", "PT0.2S");
+
+ PropertyUtil propertyUtil = PropertyUtil.getInstance();
+ propertyUtil.properties.putAll(props);
+ PropertyUtil.overrideSystemProperties(propertyUtil.properties);
+ }
+
+ @BeforeAll
+ static void initProperties() {
+ ResilienceSupport.clearForTests();
+ PropertyUtil.clear();
+ PropertyUtil.init("test.properties");
+ setupTestFileProperties();
+ }
+
+ @BeforeEach
+ void setup() {
+ supplier = Mockito.mock(Supplier.class);
+ }
+
+ @AfterAll
+ static void tearDown() {
+ ResilienceSupport.clearForTests();
+ PropertyUtil.clear();
+ }
+
+ @Test
+ void shouldEnrichAndRethrowRebuildableExceptionWithHttpTimeoutExceptionAsTheCause() {
+ HttpTimeoutException originalException = new HttpTimeoutException("Error: request timed out!");
+ FederatorTokenException federatorTokenException = new FederatorTokenException(BASE_MESSAGE, originalException);
+ Mockito.when(supplier.get()).thenThrow(federatorTokenException);
+
+ FederatorTokenException thrown = assertThrows(
+ FederatorTokenException.class,
+ () -> ResilienceSupport.decorateAndExecute(COMPONENT_NAME, OPERATION, TARGET_ID, supplier));
+
+ assertEquals(
+ BASE_MESSAGE + " (timeout while calling " + COMPONENT_NAME + " for " + TARGET_ID + ")",
+ thrown.getMessage());
+ assertEquals(federatorTokenException, thrown.getCause());
+ }
+
+ @Test
+ void shouldEnrichAndRethrowRebuildableExceptionWithSocketTimeoutExceptionAsTheCause() {
+ SocketTimeoutException originalException = new SocketTimeoutException("Error: request timed out!");
+ FederatorTokenException federatorTokenException = new FederatorTokenException(BASE_MESSAGE, originalException);
+ Mockito.when(supplier.get()).thenThrow(federatorTokenException);
+
+ FederatorTokenException thrown = assertThrows(
+ FederatorTokenException.class,
+ () -> ResilienceSupport.decorateAndExecute(COMPONENT_NAME, OPERATION, TARGET_ID, supplier));
+
+ assertEquals(
+ BASE_MESSAGE + " (timeout while calling " + COMPONENT_NAME + " for " + TARGET_ID + ")",
+ thrown.getMessage());
+ assertEquals(federatorTokenException, thrown.getCause());
+ }
+
+ @Test
+ void shouldEnrichAndRethrowRebuildableExceptionWithInterruptedIOExceptionAsTheCause() {
+ InterruptedIOException originalException = new InterruptedIOException("Error: file not found!");
+ FederatorTokenException federatorTokenException = new FederatorTokenException(BASE_MESSAGE, originalException);
+ Mockito.when(supplier.get()).thenThrow(federatorTokenException);
+
+ FederatorTokenException thrown = assertThrows(
+ FederatorTokenException.class,
+ () -> ResilienceSupport.decorateAndExecute(COMPONENT_NAME, OPERATION, TARGET_ID, supplier));
+
+ assertEquals(BASE_MESSAGE + " (request was interrupted for " + TARGET_ID + ")", thrown.getMessage());
+ assertEquals(federatorTokenException, thrown.getCause());
+ }
+
+ @Test
+ void shouldEnrichAndRethrowRebuildableExceptionWithInterruptedExceptionAsTheCause() {
+ InterruptedException originalException = new InterruptedException("Error: thread interrupted!");
+ FederatorTokenException federatorTokenException = new FederatorTokenException(BASE_MESSAGE, originalException);
+ Mockito.when(supplier.get()).thenThrow(federatorTokenException);
+
+ FederatorTokenException thrown = assertThrows(
+ FederatorTokenException.class,
+ () -> ResilienceSupport.decorateAndExecute(COMPONENT_NAME, OPERATION, TARGET_ID, supplier));
+
+ assertEquals(BASE_MESSAGE + " (request was interrupted for " + TARGET_ID + ")", thrown.getMessage());
+ assertEquals(federatorTokenException, thrown.getCause());
+ }
+
+ @Test
+ void shouldEnrichAndRethrowRebuildableExceptionWithIOExceptionAsTheCause() {
+ IOException originalException = new IOException("Error: File not found!");
+ FederatorTokenException federatorTokenException = new FederatorTokenException(BASE_MESSAGE, originalException);
+ Mockito.when(supplier.get()).thenThrow(federatorTokenException);
+
+ FederatorTokenException thrown = assertThrows(
+ FederatorTokenException.class,
+ () -> ResilienceSupport.decorateAndExecute(COMPONENT_NAME, OPERATION, TARGET_ID, supplier));
+
+ assertEquals(
+ BASE_MESSAGE + " (I/O error while calling " + COMPONENT_NAME + " for " + TARGET_ID + ")",
+ thrown.getMessage());
+ assertEquals(federatorTokenException, thrown.getCause());
+ }
+
+ @Test
+ void shouldEnrichAndRethrowRebuildableExceptionWithJedisExceptionAsTheCause() {
+ JedisException originalException = new JedisException("Error: key not found!");
+ FederatorTokenException federatorTokenException = new FederatorTokenException(BASE_MESSAGE, originalException);
+ Mockito.when(supplier.get()).thenThrow(federatorTokenException);
+
+ FederatorTokenException thrown = assertThrows(
+ FederatorTokenException.class,
+ () -> ResilienceSupport.decorateAndExecute(COMPONENT_NAME, OPERATION, TARGET_ID, supplier));
+
+ assertEquals(BASE_MESSAGE + " (redis cache failure for " + TARGET_ID + ")", thrown.getMessage());
+ assertEquals(federatorTokenException, thrown.getCause());
+ }
+
+ @Test
+ void shouldEnrichAndRethrowRebuildableExceptionWithUnmatchedCause() {
+ RuntimeException originalException = new RuntimeException("unknown");
+ FederatorTokenException federatorTokenException = new FederatorTokenException(BASE_MESSAGE, originalException);
+ Mockito.when(supplier.get()).thenThrow(federatorTokenException);
+
+ FederatorTokenException thrown = assertThrows(
+ FederatorTokenException.class,
+ () -> ResilienceSupport.decorateAndExecute(COMPONENT_NAME, OPERATION, TARGET_ID, supplier));
+
+ assertEquals(
+ BASE_MESSAGE + " (unexpected failure during " + OPERATION + " for " + TARGET_ID + ")",
+ thrown.getMessage());
+ assertEquals(federatorTokenException, thrown.getCause());
+ }
+
+ @Test
+ void shouldRethrowNonRebuildableExceptionAsIs() {
+ CallNotPermittedException callNotPermittedException =
+ CallNotPermittedException.createCallNotPermittedException(CircuitBreaker.ofDefaults(COMPONENT_NAME));
+
+ Mockito.when(supplier.get()).thenThrow(callNotPermittedException);
+
+ CallNotPermittedException thrown = assertThrows(
+ CallNotPermittedException.class,
+ () -> ResilienceSupport.decorateAndExecute(COMPONENT_NAME, OPERATION, TARGET_ID, supplier));
+
+ assertEquals(callNotPermittedException, thrown);
+ }
+
+ @Test
+ void shouldHandleNullBaseMessageGracefully() {
+ HttpTimeoutException originalException = new HttpTimeoutException("Error: request timed out!");
+ FederatorTokenException federatorTokenException = new FederatorTokenException(null, originalException);
+ Mockito.when(supplier.get()).thenThrow(federatorTokenException);
+
+ FederatorTokenException thrown = assertThrows(
+ FederatorTokenException.class,
+ () -> ResilienceSupport.decorateAndExecute(COMPONENT_NAME, OPERATION, TARGET_ID, supplier));
+
+ assertEquals("" + " (timeout while calling " + COMPONENT_NAME + " for " + TARGET_ID + ")", thrown.getMessage());
+ assertEquals(federatorTokenException, thrown.getCause());
+ }
+
+ @Test
+ void shouldHandleNullCauseGracefully() {
+ FederatorTokenException federatorTokenException = new FederatorTokenException(BASE_MESSAGE);
+ Mockito.when(supplier.get()).thenThrow(federatorTokenException);
+
+ FederatorTokenException thrown = assertThrows(
+ FederatorTokenException.class,
+ () -> ResilienceSupport.decorateAndExecute(COMPONENT_NAME, OPERATION, TARGET_ID, supplier));
+
+ assertEquals(
+ BASE_MESSAGE + " (unexpected failure during " + OPERATION + " for " + TARGET_ID + ")",
+ thrown.getMessage());
+ assertEquals(federatorTokenException, thrown.getCause());
+ }
+}
diff --git a/src/test/java/uk/gov/dbt/ndtp/federator/server/conductor/AbstractEventMessageConductorTest.java b/src/test/java/uk/gov/dbt/ndtp/federator/server/conductor/AbstractEventMessageConductorTest.java
index b44294d9..b866f546 100644
--- a/src/test/java/uk/gov/dbt/ndtp/federator/server/conductor/AbstractEventMessageConductorTest.java
+++ b/src/test/java/uk/gov/dbt/ndtp/federator/server/conductor/AbstractEventMessageConductorTest.java
@@ -61,7 +61,7 @@ void testProcessMessage_WithEvent() {
@Test
void testProcessMessages() throws Exception {
- when(mockConsumer.stillAvailable()).thenReturn(true, false);
+ when(mockConsumer.stillAvailable()).thenReturn(true, false, false);
Event mockEvent = mock(Event.class);
when(mockEvent.key()).thenReturn("testKey");
when(mockEvent.headers()).thenReturn(Stream.empty());
@@ -70,22 +70,31 @@ void testProcessMessages() throws Exception {
conductor.processMessages();
verify(mockProcessor, times(1)).process(mockEvent);
- verify(mockConsumer, times(2)).stillAvailable();
+ verify(mockConsumer, times(3)).stillAvailable();
}
@Test
- void testClose() {
+ void testClose_WithConsumerIsStillAvailable() {
+ when(mockConsumer.stillAvailable()).thenReturn(true);
conductor.close();
verify(mockConsumer).close();
verify(mockProcessor).close();
}
+ @Test
+ void testClose_WithConsumerNotStillAvailable() {
+ when(mockConsumer.stillAvailable()).thenReturn(false);
+ conductor.close();
+ verify(mockProcessor).close();
+ }
+
@Test
void testClose_WithExceptions() {
doThrow(new RuntimeException("Consumer Close Error")).when(mockConsumer).close();
doThrow(new RuntimeException("Processor Close Error"))
.when(mockProcessor)
.close();
+ when(mockConsumer.stillAvailable()).thenReturn(true);
// Should not throw exception
assertDoesNotThrow(() -> conductor.close());
diff --git a/src/test/java/uk/gov/dbt/ndtp/federator/server/conductor/AbstractMessageConductorTest.java b/src/test/java/uk/gov/dbt/ndtp/federator/server/conductor/AbstractMessageConductorTest.java
index 67c9b1e5..bf213469 100644
--- a/src/test/java/uk/gov/dbt/ndtp/federator/server/conductor/AbstractMessageConductorTest.java
+++ b/src/test/java/uk/gov/dbt/ndtp/federator/server/conductor/AbstractMessageConductorTest.java
@@ -57,13 +57,24 @@ void test_processMessages_shouldProcessMessagesCorrectly() throws Exception {
}
@Test
- void test_close_shouldHandleClosingResources() {
+ void test_close_shouldHandleClosingResources_whenMessageConsumerIsStillAvailable() {
+ when(messageConsumer.stillAvailable()).thenReturn(true);
+
conductor.close();
verify(messageConsumer).close();
verify(messageProcessor).close();
}
+ @Test
+ void test_close_shouldHandleClosingResources_whenMessageConsumerIsNotStillAvailable() {
+ when(messageConsumer.stillAvailable()).thenReturn(false);
+
+ conductor.close();
+
+ verify(messageProcessor).close();
+ }
+
@Test
void test_processMessages_shouldThrowMessageProcessingException() {
when(messageConsumer.stillAvailable()).thenReturn(true);
diff --git a/src/test/java/uk/gov/dbt/ndtp/federator/server/conductor/FileConductorTest.java b/src/test/java/uk/gov/dbt/ndtp/federator/server/conductor/FileConductorTest.java
index 3ea09260..fd956ee1 100644
--- a/src/test/java/uk/gov/dbt/ndtp/federator/server/conductor/FileConductorTest.java
+++ b/src/test/java/uk/gov/dbt/ndtp/federator/server/conductor/FileConductorTest.java
@@ -16,6 +16,7 @@
import org.junit.jupiter.api.Test;
import uk.gov.dbt.ndtp.federator.common.model.FileTransferRequest;
import uk.gov.dbt.ndtp.federator.common.model.dto.AttributesDTO;
+import uk.gov.dbt.ndtp.federator.exceptions.MessageProcessingException;
import uk.gov.dbt.ndtp.federator.server.consumer.MessageConsumer;
import uk.gov.dbt.ndtp.federator.server.interfaces.StreamObservable;
import uk.gov.dbt.ndtp.federator.server.processor.MessageProcessor;
@@ -50,9 +51,9 @@ void testContinueProcessing_ObserverCancelled() {
when(mockObserver.isCancelled()).thenReturn(true);
boolean result = conductor.continueProcessing();
+ conductor.processMessages();
assertFalse(result);
- verify(mockConsumer).close();
}
@Test
@@ -70,4 +71,36 @@ void testContinueProcessing_ObserverNotCancelled_ConsumerNotAvailable() {
assertFalse(conductor.continueProcessing());
}
+
+ @Test
+ void testProcessMessages_ConsumerAndProcessorClosed_AfterProcessingAllMessages() {
+ when(mockConsumer.stillAvailable()).thenReturn(false);
+
+ conductor.processMessages();
+
+ verify(mockProcessor, times(1)).close();
+ }
+
+ @Test
+ void testProcessMessages_ConsumerAndProcessorClosed_AfterProcessingAllMessages_AndConsumerStillAvailable() {
+ when(mockConsumer.stillAvailable()).thenReturn(false, true);
+
+ conductor.processMessages();
+
+ verify(mockConsumer, times(1)).close();
+ verify(mockProcessor, times(1)).close();
+ }
+
+ @Test
+ void testProcessMessages_ConsumerAndProcessorClosed_IfExceptionThrownWhileProcessingMessages() {
+ when(mockConsumer.stillAvailable()).thenReturn(true);
+ doThrow(new RuntimeException("Error when fetching next message!"))
+ .when(mockConsumer)
+ .getNextMessage();
+
+ assertThrows(MessageProcessingException.class, () -> conductor.processMessages());
+
+ verify(mockConsumer, times(1)).close();
+ verify(mockProcessor, times(1)).close();
+ }
}
diff --git a/src/test/java/uk/gov/dbt/ndtp/federator/server/processor/file/FileKafkaEventMessageProcessorTest.java b/src/test/java/uk/gov/dbt/ndtp/federator/server/processor/file/FileKafkaEventMessageProcessorTest.java
index e0734b36..6315b71d 100644
--- a/src/test/java/uk/gov/dbt/ndtp/federator/server/processor/file/FileKafkaEventMessageProcessorTest.java
+++ b/src/test/java/uk/gov/dbt/ndtp/federator/server/processor/file/FileKafkaEventMessageProcessorTest.java
@@ -11,7 +11,6 @@
import java.lang.reflect.Field;
import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import uk.gov.dbt.ndtp.federator.common.model.FileTransferRequest;
@@ -32,6 +31,7 @@ class FileKafkaEventMessageProcessorTest {
@BeforeEach
@SuppressWarnings("unchecked")
void setUp() throws Exception {
+ PropertyUtil.clear();
PropertyUtil.init("client.properties");
mockObserver = mock(StreamObservable.class);
@@ -50,11 +50,6 @@ void setUp() throws Exception {
validatorField.set(processor, mockValidator);
}
- @AfterEach
- void tearDown() {
- PropertyUtil.clear();
- }
-
@Test
@SuppressWarnings("unchecked")
void testProcessSuccessfully() throws Exception {