refactor: introduce null checks in manager

core/control-plane/control-plane-aggregate-services/src/test/java/org/eclipse/edc/connector/service/transferprocess/TransferProcessEventDispatchTest.java

@@ -10,6 +10,7 @@
  *  Contributors:
  *       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - initial API and implementation
  *       Masatake Iwasaki (NTT DATA) - fixed failure due to assertion timeout
+ *       SAP SE - refactoring
  *
  */
 
@@ -19,6 +20,7 @@
 import org.eclipse.edc.catalog.spi.DataService;
 import org.eclipse.edc.connector.core.event.EventExecutorServiceContainer;
 import org.eclipse.edc.connector.dataplane.selector.spi.store.DataPlaneInstanceStore;
+import org.eclipse.edc.connector.policy.spi.store.PolicyArchive;
 import org.eclipse.edc.connector.spi.transferprocess.TransferProcessProtocolService;
 import org.eclipse.edc.connector.spi.transferprocess.TransferProcessService;
 import org.eclipse.edc.connector.transfer.spi.retry.TransferWaitStrategy;
@@ -28,6 +30,7 @@
 import org.eclipse.edc.connector.transfer.spi.types.TransferRequest;
 import org.eclipse.edc.connector.transfer.spi.types.protocol.TransferStartMessage;
 import org.eclipse.edc.junit.extensions.EdcExtension;
+import org.eclipse.edc.policy.model.Policy;
 import org.eclipse.edc.spi.event.EventEnvelope;
 import org.eclipse.edc.spi.event.EventRouter;
 import org.eclipse.edc.spi.event.EventSubscriber;
@@ -59,6 +62,7 @@
 import static org.eclipse.edc.junit.testfixtures.TestUtils.getFreePort;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.argThat;
+import static org.mockito.ArgumentMatchers.matches;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -83,15 +87,20 @@ void setUp(EdcExtension extension) {
         extension.registerServiceMock(EventExecutorServiceContainer.class, new EventExecutorServiceContainer(newSingleThreadExecutor()));
         extension.registerServiceMock(DataService.class, mock(DataService.class));
         extension.registerServiceMock(DataPlaneInstanceStore.class, mock(DataPlaneInstanceStore.class));
+        extension.registerServiceMock(PolicyArchive.class, mock(PolicyArchive.class));
     }
 
     @Test
-    void shouldDispatchEventsOnTransferProcessStateChanges(TransferProcessService service, TransferProcessProtocolService protocolService,
-                                                           EventRouter eventRouter, RemoteMessageDispatcherRegistry dispatcherRegistry,
-                                                           StatusCheckerRegistry statusCheckerRegistry) {
+    void shouldDispatchEventsOnTransferProcessStateChanges(TransferProcessService service,
+                                                           TransferProcessProtocolService protocolService,
+                                                           EventRouter eventRouter,
+                                                           RemoteMessageDispatcherRegistry dispatcherRegistry,
+                                                           StatusCheckerRegistry statusCheckerRegistry,
+                                                           PolicyArchive policyArchive) {
         var testDispatcher = mock(RemoteMessageDispatcher.class);
         when(testDispatcher.protocol()).thenReturn("test");
         when(testDispatcher.send(any(), any())).thenReturn(CompletableFuture.completedFuture("any"));
+        when(policyArchive.findPolicyForContract(matches("contractId"))).thenReturn(mock(Policy.class));
         dispatcherRegistry.register(testDispatcher);
         eventRouter.register(TransferProcessEvent.class, eventSubscriber);
         var statusCheck = mock(StatusChecker.class);
@@ -107,6 +116,7 @@ void shouldDispatchEventsOnTransferProcessStateChanges(TransferProcessService se
                 .protocol("test")
                 .managedResources(false)
                 .connectorAddress("http://an/address")
+                .contractId("contractId")
                 .build();
 
         var transferRequest = TransferRequest.Builder.newInstance()
@@ -155,6 +165,36 @@ void shouldDispatchEventsOnTransferProcessStateChanges(TransferProcessService se
         });
     }
 
+    @Test
+    void shouldTerminateOnInvalidPolicy(TransferProcessService service, EventRouter eventRouter, RemoteMessageDispatcherRegistry dispatcherRegistry) {
+        var testDispatcher = mock(RemoteMessageDispatcher.class);
+        when(testDispatcher.protocol()).thenReturn("test");
+        when(testDispatcher.send(any(), any())).thenReturn(CompletableFuture.completedFuture("any"));
+        dispatcherRegistry.register(testDispatcher);
+        eventRouter.register(TransferProcessEvent.class, eventSubscriber);
+
+        var dataRequest = DataRequest.Builder.newInstance()
+                .id("dataRequestId")
+                .assetId("assetId")
+                .destinationType("any")
+                .protocol("test")
+                .managedResources(false)
+                .connectorAddress("http://an/address")
+                .contractId("contractId")
+                .build();
+
+        var transferRequest = TransferRequest.Builder.newInstance()
+                .dataRequest(dataRequest)
+                .build();
+
+        service.initiateTransfer(transferRequest);
+
+        await().untilAsserted(() -> {
+            verify(eventSubscriber).on(argThat(isEnvelopeOf(TransferProcessInitiated.class)));
+            verify(eventSubscriber).on(argThat(isEnvelopeOf(TransferProcessTerminated.class)));
+        });
+    }
+
     @Test
     void shouldDispatchEventOnTransferProcessTerminated(TransferProcessService service, EventRouter eventRouter, RemoteMessageDispatcherRegistry dispatcherRegistry) {
         var testDispatcher = mock(RemoteMessageDispatcher.class);

core/control-plane/transfer-core/src/main/java/org/eclipse/edc/connector/transfer/process/TransferProcessManagerImpl.java

@@ -10,6 +10,7 @@
  *  Contributors:
  *       Microsoft Corporation - initial API and implementation
  *       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - improvements
+ *       SAP SE - refactoring
  *
  */
 
@@ -291,7 +292,13 @@ private StatusResult<String> initiateRequest(TransferProcess.Type type, Transfer
     private boolean processInitial(TransferProcess process) {
         var dataRequest = process.getDataRequest();
 
-        var policy = policyArchive.findPolicyForContract(dataRequest.getContractId());
+        var contractId = dataRequest.getContractId();
+        var policy = policyArchive.findPolicyForContract(contractId);
+
+        if (policy == null) {
+            transitionToTerminating(process, "Policy not found for contract: " + contractId);
+            return true;
+        }
 
         ResourceManifest manifest;
         if (process.getType() == CONSUMER) {

core/control-plane/transfer-core/src/test/java/org/eclipse/edc/connector/transfer/transfer/TransferProcessManagerImplTest.java

@@ -10,6 +10,7 @@
  *  Contributors:
  *       Microsoft Corporation - initial API and implementation
  *       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - improvements
+ *       SAP SE - refactoring
  *
  */
 
@@ -231,6 +232,22 @@ void initial_manifestEvaluationFailed_shouldTransitionToTerminating() {
         });
     }
 
+    @Test
+    void initial_noPolicyFound_shouldTransitionToTerminating() {
+        when(transferProcessStore.nextForState(eq(INITIAL.code()), anyInt()))
+                .thenReturn(List.of(createTransferProcess(INITIAL)))
+                .thenReturn(emptyList());
+        when(policyArchive.findPolicyForContract(anyString())).thenReturn(null);
+
+        manager.start();
+
+        await().untilAsserted(() -> {
+            verify(policyArchive, atLeastOnce()).findPolicyForContract(anyString());
+            verifyNoInteractions(provisionManager);
+            verify(transferProcessStore).save(argThat(p -> p.getState() == TERMINATING.code()));
+        });
+    }
+
     @Test
     void provisioning_shouldTransitionToProvisionedOnDataDestination() {
         var process = createTransferProcess(PROVISIONING).toBuilder()

extensions/control-plane/provision/provision-aws-s3/src/main/java/org/eclipse/edc/connector/provision/aws/s3/S3ConsumerResourceDefinitionGenerator.java

@@ -10,6 +10,7 @@
  *  Contributors:
  *       Microsoft Corporation - initial API and implementation
  *       ZF Friedrichshafen AG - improvements (refactoring of generate method)
+ *       SAP SE - refactoring
  *
  */
 
@@ -22,8 +23,6 @@
 import org.eclipse.edc.policy.model.Policy;
 import software.amazon.awssdk.regions.Region;
 
-import java.util.Objects;
-
 import static java.util.UUID.randomUUID;
 
 /**
@@ -33,9 +32,6 @@ public class S3ConsumerResourceDefinitionGenerator implements ConsumerResourceDe
 
     @Override
     public ResourceDefinition generate(DataRequest dataRequest, Policy policy) {
-        Objects.requireNonNull(dataRequest, "dataRequest must always be provided");
-        Objects.requireNonNull(policy, "policy must always be provided");
-
         if (dataRequest.getDataDestination().getProperty(S3BucketSchema.REGION) == null) {
             // FIXME generate region from policy engine
             return S3BucketResourceDefinition.Builder.newInstance().id(randomUUID().toString()).bucketName(dataRequest.getDataDestination().getProperty(S3BucketSchema.BUCKET_NAME)).regionId(Region.US_EAST_1.id()).build();
@@ -48,9 +44,6 @@ public ResourceDefinition generate(DataRequest dataRequest, Policy policy) {
 
     @Override
     public boolean canGenerate(DataRequest dataRequest, Policy policy) {
-        Objects.requireNonNull(dataRequest, "dataRequest must always be provided");
-        Objects.requireNonNull(policy, "policy must always be provided");
-
         return S3BucketSchema.TYPE.equals(dataRequest.getDestinationType());
     }
 }

extensions/control-plane/provision/provision-aws-s3/src/test/java/org/eclipse/edc/connector/provision/aws/s3/S3ConsumerResourceDefinitionGeneratorTest.java

@@ -9,6 +9,7 @@
  *
  *  Contributors:
  *       ZF Friedrichshafen AG - Initial API and Implementation
+ *       SAP SE - refactoring
  *
  */
 
@@ -57,17 +58,6 @@ void generate() {
         assertThat(objectDef.getId()).satisfies(UUID::fromString);
     }
 
-    @Test
-    void generate_noPolicyAsParameter() {
-        var destination = DataAddress.Builder.newInstance().type(S3BucketSchema.TYPE)
-                .property(S3BucketSchema.BUCKET_NAME, "test-name")
-                .property(S3BucketSchema.REGION, Region.EU_WEST_2.id())
-                .build();
-        var asset = Asset.Builder.newInstance().build();
-        var dr = DataRequest.Builder.newInstance().dataDestination(destination).assetId(asset.getId()).processId("process-id").build();
-        assertThatExceptionOfType(NullPointerException.class).isThrownBy(() -> generator.generate(dr, null));
-    }
-
     @Test
     void generate_noDataRequestAsParameter() {
         var policy = Policy.Builder.newInstance().build();

extensions/control-plane/provision/provision-blob/src/main/java/org/eclipse/edc/connector/provision/azure/blob/ObjectStorageConsumerResourceDefinitionGenerator.java

@@ -10,6 +10,7 @@
  *  Contributors:
  *       Microsoft Corporation - initial API and implementation
  *       ZF Friedrichshafen AG - improvements (refactoring of generate method)
+ *       SAP SE - refactoring
  *
  */
 
@@ -22,17 +23,12 @@
 import org.eclipse.edc.policy.model.Policy;
 import org.jetbrains.annotations.Nullable;
 
-import java.util.Objects;
-
 import static java.util.UUID.randomUUID;
 
 public class ObjectStorageConsumerResourceDefinitionGenerator implements ConsumerResourceDefinitionGenerator {
 
     @Override
     public @Nullable ResourceDefinition generate(DataRequest dataRequest, Policy policy) {
-        Objects.requireNonNull(dataRequest, "dataRequest must always be provided");
-        Objects.requireNonNull(policy, "policy must always be provided");
-
         var destination = dataRequest.getDataDestination();
         var id = randomUUID().toString();
         var account = destination.getProperty(AzureBlobStoreSchema.ACCOUNT_NAME);
@@ -46,9 +42,6 @@ public class ObjectStorageConsumerResourceDefinitionGenerator implements Consume
 
     @Override
     public boolean canGenerate(DataRequest dataRequest, Policy policy) {
-        Objects.requireNonNull(dataRequest, "dataRequest must always be provided");
-        Objects.requireNonNull(policy, "policy must always be provided");
-
         return AzureBlobStoreSchema.TYPE.equals(dataRequest.getDestinationType());
     }
 }

extensions/control-plane/provision/provision-blob/src/test/java/org/eclipse/edc/connector/provision/azure/blob/ObjectStorageConsumerResourceDefinitionGeneratorTest.java

@@ -10,6 +10,7 @@
  *  Contributors:
  *       Microsoft Corporation - initial API and implementation
  *       ZF Friedrichshafen AG - Addition of new tests
+ *       SAP SE - refactoring
  *
  */
 
@@ -62,19 +63,6 @@ void generate_noDataRequestAsParameter() {
         assertThatExceptionOfType(NullPointerException.class).isThrownBy(() -> generator.generate(null, policy));
     }
 
-    @Test
-    void generate_noPolicyAsParameter() {
-        var destination = DataAddress.Builder.newInstance().type(AzureBlobStoreSchema.TYPE)
-                .property(AzureBlobStoreSchema.CONTAINER_NAME, "test-container")
-                .property(AzureBlobStoreSchema.ACCOUNT_NAME, "test-account")
-                .build();
-        var asset = Asset.Builder.newInstance().build();
-        var dr = DataRequest.Builder.newInstance().dataDestination(destination).assetId(asset.getId()).build();
-
-        assertThatExceptionOfType(NullPointerException.class).isThrownBy(() -> generator.generate(dr, null));
-    }
-
-
     @Test
     void generate_withoutContainerName() {
         var destination = DataAddress.Builder.newInstance().type(AzureBlobStoreSchema.TYPE)
@@ -93,7 +81,6 @@ void generate_withoutContainerName() {
         assertThat(objectDef.getId()).satisfies(UUID::fromString);
     }
 
-
     @Test
     void canGenerate() {
         var destination = DataAddress.Builder.newInstance().type(AzureBlobStoreSchema.TYPE)

extensions/control-plane/provision/provision-gcs/src/main/java/org/eclipse/edc/connector/provision/gcp/GcsConsumerResourceDefinitionGenerator.java

@@ -10,6 +10,7 @@
  *  Contributors:
  *       Google LCC - Initial implementation
  *       ZF Friedrichshafen AG - improvements (refactoring of generate method)
+ *       SAP SE - refactoring
  *
  */
 
@@ -22,18 +23,13 @@
 import org.eclipse.edc.policy.model.Policy;
 import org.jetbrains.annotations.Nullable;
 
-import java.util.Objects;
-
 import static java.util.UUID.randomUUID;
 
 public class GcsConsumerResourceDefinitionGenerator implements ConsumerResourceDefinitionGenerator {
 
     @Override
     public @Nullable
     ResourceDefinition generate(DataRequest dataRequest, Policy policy) {
-        Objects.requireNonNull(dataRequest, "dataRequest must always be provided");
-        Objects.requireNonNull(policy, "policy must always be provided");
-
         var destination = dataRequest.getDataDestination();
         var id = randomUUID().toString();
         var location = destination.getProperty(GcsStoreSchema.LOCATION);
@@ -45,9 +41,6 @@ ResourceDefinition generate(DataRequest dataRequest, Policy policy) {
 
     @Override
     public boolean canGenerate(DataRequest dataRequest, Policy policy) {
-        Objects.requireNonNull(dataRequest, "dataRequest must always be provided");
-        Objects.requireNonNull(policy, "policy must always be provided");
-
         return GcsStoreSchema.TYPE.equals(dataRequest.getDestinationType());
     }
 }

extensions/control-plane/provision/provision-gcs/src/test/java/org/eclipse/edc/connector/provision/gcp/GcsConsumerResourceDefinitionGeneratorTest.java

@@ -9,6 +9,7 @@
  *
  *  Contributors:
  *       ZF Friedrichshafen AG - Initial API and Implementation
+ *       SAP SE - refactoring
  *
  */
 
@@ -61,18 +62,6 @@ void generate_noDataRequestAsParameter() {
         assertThatExceptionOfType(NullPointerException.class).isThrownBy(() -> generator.generate(null, policy));
     }
 
-    @Test
-    void generate_noPolicyAsParameter() {
-        var destination = DataAddress.Builder.newInstance().type(GcsStoreSchema.TYPE)
-                .property(GcsStoreSchema.LOCATION, "test-location")
-                .property(GcsStoreSchema.STORAGE_CLASS, "test-storage-class")
-                .build();
-        var asset = Asset.Builder.newInstance().build();
-        var dr = DataRequest.Builder.newInstance().dataDestination(destination).assetId(asset.getId()).build();
-
-        assertThatExceptionOfType(NullPointerException.class).isThrownBy(() -> generator.generate(dr, null));
-    }
-
     @Test
     void canGenerate() {
         var destination = DataAddress.Builder.newInstance().type(GcsStoreSchema.TYPE)

extensions/control-plane/provision/provision-http/src/main/java/org/eclipse/edc/connector/provision/http/impl/HttpProviderResourceDefinitionGenerator.java

@@ -10,6 +10,7 @@
  *  Contributors:
  *       Microsoft Corporation - initial API and implementation
  *       ZF Friedrichshafen AG - improvements (refactoring of canGenerate method)
+ *       SAP SE - refactoring
  *
  */
 
@@ -23,7 +24,6 @@
 import org.eclipse.edc.spi.types.domain.DataAddress;
 import org.jetbrains.annotations.Nullable;
 
-import java.util.Objects;
 import java.util.UUID;
 
 import static java.util.Objects.requireNonNull;
@@ -40,10 +40,6 @@ public HttpProviderResourceDefinitionGenerator(String dataAddressType) {
 
     @Override
     public @Nullable ResourceDefinition generate(DataRequest dataRequest, DataAddress assetAddress, Policy policy) {
-        Objects.requireNonNull(dataRequest, "dataRequest must always be provided");
-        Objects.requireNonNull(assetAddress, "assetAddress must always be provided");
-        Objects.requireNonNull(policy, "policy must always be provided");
-
         var assetId = dataRequest.getAssetId();
 
         if (assetId == null) {
@@ -60,10 +56,6 @@ public HttpProviderResourceDefinitionGenerator(String dataAddressType) {
 
     @Override
     public boolean canGenerate(DataRequest dataRequest, DataAddress assetAddress, Policy policy) {
-        Objects.requireNonNull(dataRequest, "dataRequest must always be provided");
-        Objects.requireNonNull(assetAddress, "assetAddress must always be provided");
-        Objects.requireNonNull(policy, "policy must always be provided");
-
         return dataAddressType.equals(assetAddress.getType());
     }
 }