feat: add filtering for participant context in selectFor (#5669)

wolf4ood · web-flow · commit bd79c143f9cd · 2026-04-15T15:27:00.000+02:00
diff --git a/core/data-plane-selector/data-plane-selector-core/src/main/java/org/eclipse/edc/connector/dataplane/selector/service/EmbeddedDataPlaneSelectorService.java b/core/data-plane-selector/data-plane-selector-core/src/main/java/org/eclipse/edc/connector/dataplane/selector/service/EmbeddedDataPlaneSelectorService.java
@@ -29,6 +29,7 @@
 import java.util.function.Predicate;
 
 import static org.eclipse.edc.connector.dataplane.selector.spi.instance.DataPlaneInstanceStates.UNREGISTERED;
+import static org.eclipse.edc.participantcontext.spi.types.ParticipantResource.queryByParticipantContextId;
 
 public class EmbeddedDataPlaneSelectorService implements DataPlaneSelectorService {
 
@@ -85,7 +86,7 @@ public ServiceResult<DataPlaneInstance> selectFor(TransferProcess transferProces
             return ServiceResult.badRequest("Strategy " + selectionStrategy + " was not found");
         }
         return transactionContext.execute(() -> {
-            try (var stream = store.getAll()) {
+            try (var stream = store.query(queryByParticipantContextId(transferProcess.getParticipantContextId()).build())) {
                 var dataPlanes = stream
                         .filter(it -> it.getState() != UNREGISTERED.code())
                         .filter(it -> it.getAllowedTransferTypes().contains(transferProcess.getTransferType()))
diff --git a/core/data-plane-selector/data-plane-selector-core/src/main/java/org/eclipse/edc/connector/dataplane/selector/store/InMemoryDataPlaneInstanceStore.java b/core/data-plane-selector/data-plane-selector-core/src/main/java/org/eclipse/edc/connector/dataplane/selector/store/InMemoryDataPlaneInstanceStore.java
@@ -18,6 +18,7 @@
 import org.eclipse.edc.connector.dataplane.selector.spi.instance.DataPlaneInstanceStates;
 import org.eclipse.edc.connector.dataplane.selector.spi.store.DataPlaneInstanceStore;
 import org.eclipse.edc.spi.query.CriterionOperatorRegistry;
+import org.eclipse.edc.spi.query.QuerySpec;
 import org.eclipse.edc.spi.result.StoreResult;
 import org.eclipse.edc.store.InMemoryStatefulEntityStore;
 
@@ -52,4 +53,9 @@ public StoreResult<DataPlaneInstance> deleteById(String instanceId) {
     public Stream<DataPlaneInstance> getAll() {
         return findAll();
     }
+
+    @Override
+    public Stream<DataPlaneInstance> query(QuerySpec querySpec) {
+        return findAll(querySpec);
+    }
 }
diff --git a/core/data-plane-selector/data-plane-selector-core/src/test/java/org/eclipse/edc/connector/dataplane/selector/service/EmbeddedDataPlaneSelectorServiceTest.java b/core/data-plane-selector/data-plane-selector-core/src/test/java/org/eclipse/edc/connector/dataplane/selector/service/EmbeddedDataPlaneSelectorServiceTest.java
@@ -54,6 +54,20 @@ public class EmbeddedDataPlaneSelectorServiceTest {
     private final String configuredSelectionStrategy = "strategy";
     private final DataPlaneSelectorService service = new EmbeddedDataPlaneSelectorService(store, selectionStrategyRegistry, new NoopTransactionContext(), configuredSelectionStrategy);
 
+    private DataPlaneInstance.Builder createInstanceBuilder(String id) {
+        return DataPlaneInstance.Builder.newInstance()
+                .id(id)
+                .url("http://any");
+    }
+
+    private DataAddress createAddress(String type) {
+        return DataAddress.Builder.newInstance()
+                .type(type)
+                .keyName("key-name")
+                .property("someprop", "someval")
+                .build();
+    }
+
     @Nested
     class SelectFor {
 
@@ -64,7 +78,7 @@ void shouldUseConfiguredSelector() {
                     createInstanceBuilder("ignoredInstance").allowedTransferType("anotherTransferType").state(REGISTERED.code()).build(),
                     createInstanceBuilder("expectedInstance").allowedTransferType("chosenTransferType").state(REGISTERED.code()).build()
             );
-            when(store.getAll()).thenAnswer(i -> instances.stream());
+            when(store.query(any())).thenAnswer(i -> instances.stream());
             var strategy = testRandomSelectionStrategy();
             when(selectionStrategyRegistry.find(any())).thenReturn(strategy);
 
@@ -81,7 +95,7 @@ void shouldFilterOutUnregisteredInstances() {
                     .allowedSourceType("srcTestType").allowedTransferType("chosenTransferType").build();
             var unregisteredInstance = createInstanceBuilder("unregistered").state(UNREGISTERED.code())
                     .allowedSourceType("srcTestType").allowedTransferType("chosenTransferType").build();
-            when(store.getAll()).thenReturn(Stream.of(unregisteredInstance, registeredInstance));
+            when(store.query(any())).thenReturn(Stream.of(unregisteredInstance, registeredInstance));
             var strategy = testRandomSelectionStrategy();
             when(selectionStrategyRegistry.find(any())).thenReturn(strategy);
 
@@ -101,7 +115,7 @@ void shouldFilterInstancesWithLabels_whenLabelsAreDefined() {
                     .build();
             var expectedInstance = createInstanceBuilder("expectedInstance").state(REGISTERED.code()).allowedTransferType("chosenTransferType")
                     .label("gold").label("blue").build();
-            when(store.getAll()).thenReturn(Stream.of(ignoredInstance, expectedInstance));
+            when(store.query(any())).thenReturn(Stream.of(ignoredInstance, expectedInstance));
             var strategy = testRandomSelectionStrategy();
             when(selectionStrategyRegistry.find(any())).thenReturn(strategy);
 
@@ -320,18 +334,4 @@ void shouldFail_whenLeaseFails() {
             verify(store, never()).save(any());
         }
     }
-
-    private DataPlaneInstance.Builder createInstanceBuilder(String id) {
-        return DataPlaneInstance.Builder.newInstance()
-                .id(id)
-                .url("http://any");
-    }
-
-    private DataAddress createAddress(String type) {
-        return DataAddress.Builder.newInstance()
-                .type(type)
-                .keyName("key-name")
-                .property("someprop", "someval")
-                .build();
-    }
 }
diff --git a/data-protocols/data-plane-signaling/data-plane-signaling-core/build.gradle.kts b/data-protocols/data-plane-signaling/data-plane-signaling-core/build.gradle.kts
@@ -23,6 +23,7 @@ dependencies {
     api(project(":spi:common:json-ld-spi"))
     api(project(":spi:common:transform-spi"))
     api(project(":spi:common:web-spi"))
+    api(project(":spi:common:participant-context-single-spi"))
     api(project(":spi:control-plane:contract-spi"))
     api(project(":spi:control-plane:control-plane-spi"))
     api(project(":spi:control-plane:transfer-spi"))
diff --git a/data-protocols/data-plane-signaling/data-plane-signaling-core/src/main/java/org/eclipse/edc/signaling/DataPlaneSignalingApiExtension.java b/data-protocols/data-plane-signaling/data-plane-signaling-core/src/main/java/org/eclipse/edc/signaling/DataPlaneSignalingApiExtension.java
@@ -16,6 +16,7 @@
 
 import org.eclipse.edc.connector.controlplane.services.spi.transferprocess.TransferProcessService;
 import org.eclipse.edc.connector.dataplane.selector.spi.DataPlaneSelectorService;
+import org.eclipse.edc.participantcontext.single.spi.SingleParticipantContextSupplier;
 import org.eclipse.edc.runtime.metamodel.annotation.Configuration;
 import org.eclipse.edc.runtime.metamodel.annotation.Extension;
 import org.eclipse.edc.runtime.metamodel.annotation.Inject;
@@ -26,6 +27,7 @@
 import org.eclipse.edc.signaling.port.transformer.DataFlowStatusMessageToDataFlowResponseTransformer;
 import org.eclipse.edc.signaling.port.transformer.DspDataAddressToDataAddressTransformer;
 import org.eclipse.edc.signaling.spi.authorization.SignalingAuthorizationRegistry;
+import org.eclipse.edc.spi.monitor.Monitor;
 import org.eclipse.edc.spi.system.ServiceExtension;
 import org.eclipse.edc.spi.system.ServiceExtensionContext;
 import org.eclipse.edc.spi.system.apiversion.ApiVersionService;
@@ -64,6 +66,12 @@ public class DataPlaneSignalingApiExtension implements ServiceExtension {
     @Inject
     private SignalingAuthorizationRegistry signalingAuthorizationRegistry;
 
+    @Inject(required = false)
+    private SingleParticipantContextSupplier participantContextSupplier;
+
+    @Inject
+    private Monitor monitor;
+
     @Override
     public String name() {
         return NAME;
@@ -79,7 +87,12 @@ public void initialize(ServiceExtensionContext context) {
         typeTransformerRegistry.register(new DataFlowStatusMessageToDataFlowResponseTransformer());
         typeTransformerRegistry.register(new DspDataAddressToDataAddressTransformer());
 
-        webService.registerResource(ApiContext.MANAGEMENT, new DataPlaneRegistrationApiV4Controller(dataPlaneSelectorService));
+        if (participantContextSupplier != null) {
+            webService.registerResource(ApiContext.MANAGEMENT, new DataPlaneRegistrationApiV4Controller(dataPlaneSelectorService, participantContextSupplier));
+        } else {
+            monitor.debug("Running in virtual mode registration of DataPlaneRegistrationApiV4Controller will be skipped");
+        }
+
         webService.registerResource(ApiContext.SIGNALING, new DataPlaneTransferAuthorizationFilter(signalingAuthorizationRegistry, transferProcessService, dataPlaneSelectorService));
         webService.registerResource(ApiContext.SIGNALING, new DataPlaneTransferApiController(transferProcessService, typeTransformerRegistry));
 
diff --git a/data-protocols/data-plane-signaling/data-plane-signaling-core/src/main/java/org/eclipse/edc/signaling/port/api/DataPlaneRegistrationApiV4Controller.java b/data-protocols/data-plane-signaling/data-plane-signaling-core/src/main/java/org/eclipse/edc/signaling/port/api/DataPlaneRegistrationApiV4Controller.java
@@ -24,6 +24,7 @@
 import org.eclipse.edc.connector.dataplane.selector.spi.DataPlaneSelectorService;
 import org.eclipse.edc.connector.dataplane.selector.spi.instance.AuthorizationProfile;
 import org.eclipse.edc.connector.dataplane.selector.spi.instance.DataPlaneInstance;
+import org.eclipse.edc.participantcontext.single.spi.SingleParticipantContextSupplier;
 import org.eclipse.edc.signaling.domain.DataPlaneRegistrationMessage;
 
 import java.util.Map;
@@ -37,23 +38,26 @@
 public class DataPlaneRegistrationApiV4Controller implements DataPlaneRegistrationApiV4 {
 
     private final DataPlaneSelectorService dataPlaneSelectorService;
+    private final SingleParticipantContextSupplier participantContextSupplier;
 
-    public DataPlaneRegistrationApiV4Controller(DataPlaneSelectorService dataPlaneSelectorService) {
+    public DataPlaneRegistrationApiV4Controller(DataPlaneSelectorService dataPlaneSelectorService, SingleParticipantContextSupplier participantContextSupplier) {
         this.dataPlaneSelectorService = dataPlaneSelectorService;
+        this.participantContextSupplier = participantContextSupplier;
     }
 
     @PUT
     @Override
     public Response register(DataPlaneRegistrationMessage registration) {
         toAuthorizationProfile(registration.authorization());
-        var dataPlaneInstance = DataPlaneInstance.Builder.newInstance()
-                .id(registration.dataplaneId())
-                .url(registration.endpoint())
-                .allowedTransferType(registration.transferTypes())
-                .authorizationProfile(toAuthorizationProfile(registration.authorization()))
-                .build();
-
-        dataPlaneSelectorService.register(dataPlaneInstance)
+        
+        participantContextSupplier.get().map(participantContext -> DataPlaneInstance.Builder.newInstance()
+                        .id(registration.dataplaneId())
+                        .url(registration.endpoint())
+                        .allowedTransferType(registration.transferTypes())
+                        .authorizationProfile(toAuthorizationProfile(registration.authorization()))
+                        .participantContextId(participantContext.getParticipantContextId())
+                        .build())
+                .compose(dataPlaneSelectorService::register)
                 .orElseThrow(it -> mapToException(it, DataPlaneInstance.class, registration.dataplaneId()));
 
         return Response.ok().build();
diff --git a/data-protocols/data-plane-signaling/data-plane-signaling-core/src/test/java/org/eclipse/edc/signaling/DataPlaneRegistrationApiV4ControllerTest.java b/data-protocols/data-plane-signaling/data-plane-signaling-core/src/test/java/org/eclipse/edc/signaling/DataPlaneRegistrationApiV4ControllerTest.java
@@ -18,6 +18,8 @@
 import io.restassured.specification.RequestSpecification;
 import org.eclipse.edc.connector.dataplane.selector.spi.DataPlaneSelectorService;
 import org.eclipse.edc.connector.dataplane.selector.spi.instance.DataPlaneInstance;
+import org.eclipse.edc.participantcontext.single.spi.SingleParticipantContextSupplier;
+import org.eclipse.edc.participantcontext.spi.types.ParticipantContext;
 import org.eclipse.edc.signaling.port.api.DataPlaneRegistrationApiV4Controller;
 import org.eclipse.edc.spi.result.ServiceResult;
 import org.eclipse.edc.web.jersey.testfixtures.RestControllerTestBase;
@@ -38,6 +40,20 @@
 public class DataPlaneRegistrationApiV4ControllerTest extends RestControllerTestBase {
 
     private final DataPlaneSelectorService dataPlaneSelectorService = mock();
+    private final SingleParticipantContextSupplier participantContextSupplier = () ->
+            ServiceResult.success(ParticipantContext.Builder.newInstance().participantContextId("participant-context-id").identity("identity").build());
+
+    @Override
+    protected Object controller() {
+        return new DataPlaneRegistrationApiV4Controller(dataPlaneSelectorService, participantContextSupplier);
+    }
+
+    private RequestSpecification baseRequest() {
+        return given()
+                .baseUri("http://localhost:" + port)
+                .basePath("/v4beta")
+                .when();
+    }
 
     @Nested
     class Register {
@@ -112,16 +128,4 @@ void shouldReturnError_whenDeletionFails() {
                     .statusCode(409);
         }
     }
-
-    @Override
-    protected Object controller() {
-        return new DataPlaneRegistrationApiV4Controller(dataPlaneSelectorService);
-    }
-
-    private RequestSpecification baseRequest() {
-        return given()
-                .baseUri("http://localhost:" + port)
-                .basePath("/v4beta")
-                .when();
-    }
 }
diff --git a/data-protocols/data-plane-signaling/data-plane-signaling-core/src/test/java/org/eclipse/edc/signaling/port/api/DataPlaneRegistrationApiV4ControllerTest.java b/data-protocols/data-plane-signaling/data-plane-signaling-core/src/test/java/org/eclipse/edc/signaling/port/api/DataPlaneRegistrationApiV4ControllerTest.java
@@ -16,6 +16,8 @@
 
 import io.restassured.http.ContentType;
 import org.eclipse.edc.connector.dataplane.selector.spi.DataPlaneSelectorService;
+import org.eclipse.edc.participantcontext.single.spi.SingleParticipantContextSupplier;
+import org.eclipse.edc.participantcontext.spi.types.ParticipantContext;
 import org.eclipse.edc.signaling.domain.DataPlaneRegistrationMessage;
 import org.eclipse.edc.spi.result.ServiceResult;
 import org.eclipse.edc.web.jersey.testfixtures.RestControllerTestBase;
@@ -36,6 +38,14 @@ class DataPlaneRegistrationApiV4ControllerTest extends RestControllerTestBase {
 
     private final DataPlaneSelectorService dataPlaneSelectorService = mock();
 
+    private final SingleParticipantContextSupplier participantContextSupplier = () ->
+            ServiceResult.success(ParticipantContext.Builder.newInstance().participantContextId("participant-context-id").identity("identity").build());
+
+    @Override
+    protected Object controller() {
+        return new DataPlaneRegistrationApiV4Controller(dataPlaneSelectorService, participantContextSupplier);
+    }
+
     @Nested
     class Register {
 
@@ -55,9 +65,9 @@ void shouldRegisterDataPlane() {
 
             verify(dataPlaneSelectorService).register(argThat(instance ->
                     instance.getId().equals("dp-id") &&
-                    instance.getUrl().toString().equals("http://dataplane/endpoint") &&
-                    instance.getAllowedTransferTypes().contains("HttpData-PUSH") &&
-                    instance.getAuthorizationProfile() == null
+                            instance.getUrl().toString().equals("http://dataplane/endpoint") &&
+                            instance.getAllowedTransferTypes().contains("HttpData-PUSH") &&
+                            instance.getAuthorizationProfile() == null
             ));
         }
 
@@ -79,7 +89,7 @@ void shouldRegisterDataPlane_withAuthorizationProfile() {
 
             verify(dataPlaneSelectorService).register(argThat(instance ->
                     instance.getAuthorizationProfile() != null &&
-                    instance.getAuthorizationProfile().type().equals("oauth2")
+                            instance.getAuthorizationProfile().type().equals("oauth2")
             ));
         }
 
@@ -130,9 +140,4 @@ void shouldReturnNotFound_whenDataPlaneDoesNotExist() {
                     .statusCode(404);
         }
     }
-
-    @Override
-    protected Object controller() {
-        return new DataPlaneRegistrationApiV4Controller(dataPlaneSelectorService);
-    }
 }
diff --git a/extensions/data-plane-selector/store/sql/data-plane-instance-store-sql/src/main/java/org/eclipse/edc/connector/dataplane/selector/store/sql/SqlDataPlaneInstanceStore.java b/extensions/data-plane-selector/store/sql/data-plane-instance-store-sql/src/main/java/org/eclipse/edc/connector/dataplane/selector/store/sql/SqlDataPlaneInstanceStore.java
@@ -162,6 +162,18 @@ public Stream<DataPlaneInstance> getAll() {
         });
     }
 
+    @Override
+    public Stream<DataPlaneInstance> query(QuerySpec querySpec) {
+        return transactionContext.execute(() -> {
+            try {
+                var statement = statements.createQuery(querySpec);
+                return queryExecutor.query(getConnection(), true, this::mapResultSet, statement.getQueryAsString(), statement.getParameters());
+            } catch (SQLException e) {
+                throw new EdcPersistenceException(e);
+            }
+        });
+    }
+
     private DataPlaneInstance findByIdInternal(Connection connection, String id) {
         var sql = statements.getFindByIdTemplate();
         return queryExecutor.single(connection, false, this::mapResultSet, sql, id);
diff --git a/extensions/data-plane-selector/store/sql/data-plane-instance-store-sql/src/main/java/org/eclipse/edc/connector/dataplane/selector/store/sql/schema/DataPlaneInstanceMapping.java b/extensions/data-plane-selector/store/sql/data-plane-instance-store-sql/src/main/java/org/eclipse/edc/connector/dataplane/selector/store/sql/schema/DataPlaneInstanceMapping.java
@@ -31,6 +31,7 @@ public DataPlaneInstanceMapping(DataPlaneInstanceStatements statements) {
         add("stateCount", data);
         add("stateTimestamp", data);
         add("createdAt", data);
+        add("participantContextId", data);
         add("traceContext", new JsonFieldTranslator(statements.getTraceContextColumn()));
         add("errorDetail", data);
     }
diff --git a/spi/data-plane-selector/data-plane-selector-spi/src/main/java/org/eclipse/edc/connector/dataplane/selector/spi/store/DataPlaneInstanceStore.java b/spi/data-plane-selector/data-plane-selector-spi/src/main/java/org/eclipse/edc/connector/dataplane/selector/spi/store/DataPlaneInstanceStore.java
@@ -16,6 +16,7 @@
 
 import org.eclipse.edc.connector.dataplane.selector.spi.instance.DataPlaneInstance;
 import org.eclipse.edc.spi.persistence.StateEntityStore;
+import org.eclipse.edc.spi.query.QuerySpec;
 import org.eclipse.edc.spi.result.StoreResult;
 
 import java.util.stream.Stream;
@@ -36,4 +37,7 @@ public interface DataPlaneInstanceStore extends StateEntityStore<DataPlaneInstan
 
     Stream<DataPlaneInstance> getAll();
 
+    Stream<DataPlaneInstance> query(QuerySpec querySpec);
+
+
 }
diff --git a/spi/data-plane-selector/data-plane-selector-spi/src/testFixtures/java/org/eclipse/edc/connector/dataplane/selector/spi/testfixtures/store/DataPlaneInstanceStoreTestBase.java b/spi/data-plane-selector/data-plane-selector-spi/src/testFixtures/java/org/eclipse/edc/connector/dataplane/selector/spi/testfixtures/store/DataPlaneInstanceStoreTestBase.java
@@ -33,6 +33,7 @@
 import static org.eclipse.edc.connector.dataplane.selector.spi.instance.DataPlaneInstanceStates.REGISTERED;
 import static org.eclipse.edc.connector.dataplane.selector.spi.instance.DataPlaneInstanceStates.UNREGISTERED;
 import static org.eclipse.edc.junit.assertions.AbstractResultAssert.assertThat;
+import static org.eclipse.edc.participantcontext.spi.types.ParticipantResource.queryByParticipantContextId;
 import static org.eclipse.edc.spi.persistence.StateEntityStore.hasState;
 import static org.eclipse.edc.spi.result.StoreFailure.Reason.ALREADY_LEASED;
 import static org.eclipse.edc.spi.result.StoreFailure.Reason.NOT_FOUND;
@@ -109,6 +110,24 @@ void getAll() {
         }
     }
 
+    @Nested
+    class Query {
+        @Test
+        void query_participantContext() {
+            var doc1 = createInstanceBuilder("test-id").participantContextId("participant1").build();
+            var doc2 = createInstanceBuilder("test-id-2").participantContextId("participant2").build();
+
+            var store = getStore();
+
+            store.save(doc1);
+            store.save(doc2);
+
+            var foundItems = store.query(queryByParticipantContextId("participant1").build());
+
+            assertThat(foundItems).isNotNull().hasSize(1);
+        }
+    }
+
     @Nested
     class Create {
 
diff --git a/system-tests/e2e-transfer-test/runner/src/test/java/org/eclipse/edc/test/e2e/transfer/VirtualTransferEndToEndTestBase.java b/system-tests/e2e-transfer-test/runner/src/test/java/org/eclipse/edc/test/e2e/transfer/VirtualTransferEndToEndTestBase.java
@@ -73,6 +73,7 @@ static void beforeAll(ManagementApiClientV5 connectorClient,
                 .url(consumerDataPlane.getDataFlowsEndpoint())
                 .allowedTransferType("NonFinite-PULL")
                 .authorizationProfile(toAuthorizationProfile(consumerControlPlaneOauth2Profile))
+                .participantContextId(participants.consumer().contextId())
                 .build();
 
         dataPlaneSelectorService.register(consumerDp)
@@ -83,6 +84,7 @@ static void beforeAll(ManagementApiClientV5 connectorClient,
                 .url(providerDataPlane.getDataFlowsEndpoint())
                 .allowedTransferType("NonFinite-PULL")
                 .authorizationProfile(toAuthorizationProfile(providerControlPlaneOauth2Profile))
+                .participantContextId(participants.provider().contextId())
                 .build();
 
         dataPlaneSelectorService.register(providerDp)

Original file line number	Diff line number	Diff line change
`@@ -31,6 +31,7 @@ public DataPlaneInstanceMapping(DataPlaneInstanceStatements statements) {`
`31`	`31`	`add("stateCount", data);`
`32`	`32`	`add("stateTimestamp", data);`
`33`	`33`	`add("createdAt", data);`
	`34`	`+ add("participantContextId", data);`
`34`	`35`	`add("traceContext", new JsonFieldTranslator(statements.getTraceContextColumn()));`
`35`	`36`	`add("errorDetail", data);`
`36`	`37`	`}`