eclipse-dataplane-core · ndr-brt · Apr 2, 2026 · Apr 1, 2026
diff --git a/build.gradle.kts b/build.gradle.kts
@@ -8,7 +8,7 @@ plugins {
 }
 
 group = "org.eclipse.dataplane-core"
-version = "0.0.9-SNAPSHOT"
+version = "0.0.10-SNAPSHOT"
 
 repositories {
     mavenCentral()

diff --git a/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowPrepareMessage.java b/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowPrepareMessage.java
@@ -28,6 +28,7 @@ public record DataFlowPrepareMessage(
         String datasetId,
         URI callbackAddress,
         String transferType,
+        Map<String, Object> claims,
         List<String> labels,
         Map<String, Object> metadata
 ) {

diff --git a/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowStartMessage.java b/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowStartMessage.java
@@ -31,6 +31,7 @@ public record DataFlowStartMessage(
         URI callbackAddress,
         String transferType,
         DataAddress dataAddress,
+        Map<String, Object> claims,
         List<String> labels,
         Map<String, Object> metadata
 ) {

diff --git a/src/test/java/org/eclipse/dataplane/DataplaneTest.java b/src/test/java/org/eclipse/dataplane/DataplaneTest.java
@@ -40,8 +40,6 @@
 import static com.github.tomakehurst.wiremock.client.WireMock.postRequestedFor;
 import static com.github.tomakehurst.wiremock.client.WireMock.urlPathEqualTo;
 import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.options;
-import static java.util.Collections.emptyList;
-import static java.util.Collections.emptyMap;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.eclipse.dataplane.domain.dataflow.DataFlow.State.COMPLETED;
@@ -198,7 +196,6 @@ void shouldFail_whenStatusIsNot200() {
     }
 
     private DataFlowPrepareMessage createPrepareMessage() {
-        return new DataFlowPrepareMessage("any", "any", "any", "any", "dataFlowId", "any", "any",
-                URI.create(controlPlane.baseUrl()), "Something-PUSH", emptyList(), emptyMap());
+        return MessageFactory.createPrepareMessage("dataFlowId", URI.create(controlPlane.baseUrl()), "Something-PUSH");
     }
 }
diff --git a/src/test/java/org/eclipse/dataplane/MessageFactory.java b/src/test/java/org/eclipse/dataplane/MessageFactory.java
@@ -0,0 +1,44 @@
+/*
+ *  Copyright (c) 2026 Think-it GmbH
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       Think-it GmbH - initial API and implementation
+ *
+ */
+
+package org.eclipse.dataplane;
+
+import org.eclipse.dataplane.domain.DataAddress;
+import org.eclipse.dataplane.domain.dataflow.DataFlowPrepareMessage;
+import org.eclipse.dataplane.domain.dataflow.DataFlowStartMessage;
+import org.jspecify.annotations.NonNull;
+
+import java.net.URI;
+
+import static java.util.Collections.emptyList;
+import static java.util.Collections.emptyMap;
+
+public interface MessageFactory {
+
+    static @NonNull DataFlowPrepareMessage createPrepareMessage(String consumerProcessId, URI callbackAddress, String transferType) {
+        return new DataFlowPrepareMessage("theMessageId", "theParticipantId", "theCounterPartyId",
+                "theDataspaceContext", consumerProcessId, "theAgreementId", "theDatasetId", callbackAddress,
+                transferType, emptyMap(), emptyList(), emptyMap());
+    }
+
+    static @NonNull DataFlowStartMessage createStartMessage(String providerProcessId, URI callbackAddress, String transferType) {
+        return createStartMessage(providerProcessId, callbackAddress, transferType, null);
+    }
+
+    static @NonNull DataFlowStartMessage createStartMessage(String providerProcessId, URI callbackAddress, String transferType, DataAddress destinationDataAddress) {
+        return new DataFlowStartMessage("theMessageId", "theParticipantId", "theCounterPartyId",
+                "theDataspaceContext", providerProcessId, "theAgreementId", "theDatasetId", callbackAddress,
+                transferType, destinationDataAddress, emptyMap(), emptyList(), emptyMap());
+    }
+}
diff --git a/src/test/java/org/eclipse/dataplane/scenario/AuthorizationOauth2Test.java b/src/test/java/org/eclipse/dataplane/scenario/AuthorizationOauth2Test.java
@@ -31,27 +31,23 @@
 import org.eclipse.dataplane.Dataplane;
 import org.eclipse.dataplane.HttpServer;
 import org.eclipse.dataplane.domain.Result;
-import org.eclipse.dataplane.domain.dataflow.DataFlowPrepareMessage;
 import org.eclipse.dataplane.domain.dataflow.DataFlowStatusMessage;
 import org.eclipse.dataplane.domain.registration.AuthorizationProfile;
 import org.eclipse.dataplane.domain.registration.ControlPlaneRegistrationMessage;
 import org.eclipse.dataplane.domain.registration.Oauth2ClientCredentialsAuthorization;
-import org.jspecify.annotations.NonNull;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
-import java.net.URI;
 import java.util.Date;
 import java.util.Map;
 import java.util.Objects;
 import java.util.UUID;
 
 import static jakarta.ws.rs.core.MediaType.APPLICATION_FORM_URLENCODED;
 import static jakarta.ws.rs.core.MediaType.APPLICATION_JSON;
-import static java.util.Collections.emptyList;
-import static java.util.Collections.emptyMap;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.eclipse.dataplane.MessageFactory.createPrepareMessage;
 
 public class AuthorizationOauth2Test {
 
@@ -119,12 +115,6 @@ void shouldCommunicateWithControlPlaneUsingOauth2Authorization() {
         assertThat(notifyPreparedResult.succeeded()).isTrue();
     }
 
-    private @NonNull DataFlowPrepareMessage createPrepareMessage(String consumerProcessId, URI callbackAddress, String transferType) {
-        return new DataFlowPrepareMessage("theMessageId", "theParticipantId", "theCounterPartyId",
-                "theDataspaceContext", consumerProcessId, "theAgreementId", "theDatasetId", callbackAddress,
-                transferType, emptyList(), emptyMap());
-    }
-
     private AuthorizationProfile oauth2AuthorizationProfile() {
         return new AuthorizationProfile("oauth2_client_credentials")
                 .withAttribute("tokenEndpoint", "http://localhost:" + httpServer.port() + "/oauth2/token")

diff --git a/src/test/java/org/eclipse/dataplane/scenario/AuthorizationTest.java b/src/test/java/org/eclipse/dataplane/scenario/AuthorizationTest.java
@@ -19,21 +19,17 @@
 import org.eclipse.dataplane.HttpServer;
 import org.eclipse.dataplane.authorization.TestAuthorization;
 import org.eclipse.dataplane.domain.Result;
-import org.eclipse.dataplane.domain.dataflow.DataFlowPrepareMessage;
 import org.eclipse.dataplane.domain.dataflow.DataFlowStatusMessage;
 import org.eclipse.dataplane.domain.registration.ControlPlaneRegistrationMessage;
 import org.eclipse.dataplane.port.exception.DataFlowNotifyControlPlaneFailed;
-import org.jspecify.annotations.NonNull;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
-import java.net.URI;
 import java.util.UUID;
 
-import static java.util.Collections.emptyList;
-import static java.util.Collections.emptyMap;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.eclipse.dataplane.MessageFactory.createPrepareMessage;
 import static org.eclipse.dataplane.authorization.TestAuthorization.TOKEN_GENERATOR;
 
 public class AuthorizationTest {
@@ -130,10 +126,4 @@ void shouldGetUnauthorized_withDataPlaneIsNotAuthenticated() {
         });
     }
 
-    private @NonNull DataFlowPrepareMessage createPrepareMessage(String consumerProcessId, URI callbackAddress, String transferType) {
-        return new DataFlowPrepareMessage("theMessageId", "theParticipantId", "theCounterPartyId",
-                "theDataspaceContext", consumerProcessId, "theAgreementId", "theDatasetId", callbackAddress,
-                transferType, emptyList(), emptyMap());
-    }
-
 }
diff --git a/src/test/java/org/eclipse/dataplane/scenario/ConsumerPullTest.java b/src/test/java/org/eclipse/dataplane/scenario/ConsumerPullTest.java
@@ -21,13 +21,10 @@
 import org.eclipse.dataplane.domain.DataAddress;
 import org.eclipse.dataplane.domain.Result;
 import org.eclipse.dataplane.domain.dataflow.DataFlow;
-import org.eclipse.dataplane.domain.dataflow.DataFlowPrepareMessage;
-import org.eclipse.dataplane.domain.dataflow.DataFlowStartMessage;
 import org.eclipse.dataplane.domain.dataflow.DataFlowStartedNotificationMessage;
 import org.eclipse.dataplane.domain.dataflow.DataFlowStatusMessage;
 import org.eclipse.dataplane.domain.dataflow.DataFlowStatusResponseMessage;
 import org.eclipse.dataplane.domain.registration.ControlPlaneRegistrationMessage;
-import org.jspecify.annotations.NonNull;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -41,9 +38,10 @@
 import java.util.UUID;
 
 import static java.util.Collections.emptyList;
-import static java.util.Collections.emptyMap;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
+import static org.eclipse.dataplane.MessageFactory.createPrepareMessage;
+import static org.eclipse.dataplane.MessageFactory.createStartMessage;
 import static org.eclipse.dataplane.authorization.TestAuthorization.TOKEN_GENERATOR;
 import static org.eclipse.dataplane.authorization.TestAuthorization.createAuthorizationProfile;
 import static org.eclipse.dataplane.domain.dataflow.DataFlow.State.PREPARED;
@@ -80,14 +78,14 @@ void shouldPullDataFromProvider() {
         var transferType = "FileSystem-PULL";
         var processId = UUID.randomUUID().toString();
         var consumerProcessId = "consumer_" + processId;
-        var prepareMessage = createPrepareMessage(consumerProcessId, transferType);
+        var prepareMessage = createPrepareMessage(consumerProcessId, controlPlane.consumerCallbackAddress(), transferType);
 
         var prepareResponse = controlPlane.consumerPrepare(prepareMessage).statusCode(200).extract().as(DataFlowStatusMessage.class);
         assertThat(prepareResponse.state()).isEqualTo(PREPARED.name());
         assertThat(prepareResponse.dataAddress()).isNull();
 
         var providerProcessId = "provider_" + processId;
-        var startMessage = createStartMessage(providerProcessId, transferType);
+        var startMessage = createStartMessage(providerProcessId, controlPlane.providerCallbackAddress(), transferType);
         var startResponse = controlPlane.providerStart(startMessage).statusCode(200).extract().as(DataFlowStatusMessage.class);
         assertThat(startResponse.state()).isEqualTo(STARTED.name());
         assertThat(startResponse.dataAddress()).isNotNull();
@@ -104,11 +102,11 @@ void shouldPermitAsyncStartup() {
         var transferType = "FileSystemAsync-PULL";
         var processId = UUID.randomUUID().toString();
         var consumerProcessId = "consumer_" + processId;
-        var prepareMessage = createPrepareMessage(consumerProcessId, transferType);
+        var prepareMessage = createPrepareMessage(consumerProcessId, controlPlane.consumerCallbackAddress(), transferType);
         controlPlane.consumerPrepare(prepareMessage).statusCode(200).extract().as(DataFlowStatusMessage.class);
 
         var providerProcessId = "provider_" + processId;
-        var startMessage = createStartMessage(providerProcessId, transferType);
+        var startMessage = createStartMessage(providerProcessId, controlPlane.providerCallbackAddress(), transferType);
         var startResponse = controlPlane.providerStart(startMessage).statusCode(202).extract().as(DataFlowStatusMessage.class);
         assertThat(startResponse.state()).isEqualTo(STARTING.name());
         assertThat(startResponse.dataAddress()).isNull();
@@ -119,18 +117,6 @@ void shouldPermitAsyncStartup() {
                 .isEqualTo(STARTED.name());
     }
 
-    private @NonNull DataFlowStartMessage createStartMessage(String providerProcessId, String transferType) {
-        return new DataFlowStartMessage("theMessageId", "theParticipantId", "theCounterPartyId",
-                "theDataspaceContext", providerProcessId, "theAgreementId", "theDatasetId", controlPlane.providerCallbackAddress(),
-                transferType, null, emptyList(), emptyMap());
-    }
-
-    private @NonNull DataFlowPrepareMessage createPrepareMessage(String consumerProcessId, String transferType) {
-        return new DataFlowPrepareMessage("theMessageId", "theParticipantId", "theCounterPartyId",
-                "theDataspaceContext", consumerProcessId, "theAgreementId", "theDatasetId", controlPlane.consumerCallbackAddress(),
-                transferType, emptyList(), emptyMap());
-    }
-
     private class ConsumerDataPlane {
 
         private final Path storage;

diff --git a/src/test/java/org/eclipse/dataplane/scenario/ProviderPushTest.java b/src/test/java/org/eclipse/dataplane/scenario/ProviderPushTest.java
@@ -21,8 +21,6 @@
 import org.eclipse.dataplane.domain.DataAddress;
 import org.eclipse.dataplane.domain.Result;
 import org.eclipse.dataplane.domain.dataflow.DataFlow;
-import org.eclipse.dataplane.domain.dataflow.DataFlowPrepareMessage;
-import org.eclipse.dataplane.domain.dataflow.DataFlowStartMessage;
 import org.eclipse.dataplane.domain.dataflow.DataFlowStatusMessage;
 import org.eclipse.dataplane.domain.dataflow.DataFlowStatusResponseMessage;
 import org.eclipse.dataplane.domain.registration.ControlPlaneRegistrationMessage;
@@ -41,9 +39,10 @@
 import java.util.concurrent.Executors;
 
 import static java.util.Collections.emptyList;
-import static java.util.Collections.emptyMap;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
+import static org.eclipse.dataplane.MessageFactory.createPrepareMessage;
+import static org.eclipse.dataplane.MessageFactory.createStartMessage;
 import static org.eclipse.dataplane.authorization.TestAuthorization.TOKEN_GENERATOR;
 import static org.eclipse.dataplane.authorization.TestAuthorization.createAuthorizationProfile;
 import static org.eclipse.dataplane.domain.dataflow.DataFlow.State.COMPLETED;
@@ -148,18 +147,6 @@ void shouldPermitAsyncPreparation() {
                 .isEqualTo(PREPARED.name());
     }
 
-    private @NonNull DataFlowStartMessage createStartMessage(String providerProcessId, URI callbackAddress, String transferType, DataAddress destinationDataAddress) {
-        return new DataFlowStartMessage("theMessageId", "theParticipantId", "theCounterPartyId",
-                "theDataspaceContext", providerProcessId, "theAgreementId", "theDatasetId", callbackAddress,
-                transferType, destinationDataAddress, emptyList(), emptyMap());
-    }
-
-    private @NonNull DataFlowPrepareMessage createPrepareMessage(String consumerProcessId, URI callbackAddress, String transferType) {
-        return new DataFlowPrepareMessage("theMessageId", "theParticipantId", "theCounterPartyId",
-                "theDataspaceContext", consumerProcessId, "theAgreementId", "theDatasetId", callbackAddress,
-                transferType, emptyList(), emptyMap());
-    }
-
     private static class ProviderDataPlane {
 
         private final ExecutorService executor = Executors.newCachedThreadPool();

diff --git a/src/test/java/org/eclipse/dataplane/scenario/StreamingPullTest.java b/src/test/java/org/eclipse/dataplane/scenario/StreamingPullTest.java
@@ -22,9 +22,7 @@
 import org.eclipse.dataplane.domain.DataAddress;
 import org.eclipse.dataplane.domain.Result;
 import org.eclipse.dataplane.domain.dataflow.DataFlow;
-import org.eclipse.dataplane.domain.dataflow.DataFlowPrepareMessage;
 import org.eclipse.dataplane.domain.dataflow.DataFlowResumeMessage;
-import org.eclipse.dataplane.domain.dataflow.DataFlowStartMessage;
 import org.eclipse.dataplane.domain.dataflow.DataFlowStartedNotificationMessage;
 import org.eclipse.dataplane.domain.dataflow.DataFlowStatusMessage;
 import org.eclipse.dataplane.domain.dataflow.DataFlowSuspendMessage;
@@ -49,9 +47,10 @@
 import java.util.concurrent.TimeUnit;
 
 import static java.util.Collections.emptyList;
-import static java.util.Collections.emptyMap;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
+import static org.eclipse.dataplane.MessageFactory.createPrepareMessage;
+import static org.eclipse.dataplane.MessageFactory.createStartMessage;
 import static org.eclipse.dataplane.authorization.TestAuthorization.TOKEN_GENERATOR;
 import static org.eclipse.dataplane.authorization.TestAuthorization.createAuthorizationProfile;
 import static org.eclipse.dataplane.domain.dataflow.DataFlow.State.PREPARED;
@@ -86,14 +85,14 @@ void shouldPullDataFromProvider_thenProviderTerminatesIt() {
         var transferType = "FileSystemStreaming-PULL";
         var processId = UUID.randomUUID().toString();
         var consumerProcessId = "consumer_" + processId;
-        var prepareMessage = prepareMessage(consumerProcessId, transferType);
+        var prepareMessage = createPrepareMessage(consumerProcessId, controlPlane.consumerCallbackAddress(), transferType);
 
         var prepareResponse = controlPlane.consumerPrepare(prepareMessage).statusCode(200).extract().as(DataFlowStatusMessage.class);
         assertThat(prepareResponse.state()).isEqualTo(PREPARED.name());
         assertThat(prepareResponse.dataAddress()).isNull();
 
         var providerProcessId = "provider_" + processId;
-        var startMessage = startMessage(providerProcessId, transferType);
+        var startMessage = createStartMessage(providerProcessId, controlPlane.providerCallbackAddress(), transferType);
         var startResponse = controlPlane.providerStart(startMessage).statusCode(200).extract().as(DataFlowStatusMessage.class);
         assertThat(startResponse.state()).isEqualTo(STARTED.name());
         assertThat(startResponse.dataAddress()).isNotNull();
@@ -114,14 +113,14 @@ void shouldSuspendAndResumeOnProvider() {
         var transferType = "FileSystemStreaming-PULL";
         var processId = UUID.randomUUID().toString();
         var consumerProcessId = "consumer_" + processId;
-        var prepareMessage = prepareMessage(consumerProcessId, transferType);
+        var prepareMessage = createPrepareMessage(consumerProcessId, controlPlane.consumerCallbackAddress(), transferType);
 
         var prepareResponse = controlPlane.consumerPrepare(prepareMessage).statusCode(200).extract().as(DataFlowStatusMessage.class);
         assertThat(prepareResponse.state()).isEqualTo(PREPARED.name());
         assertThat(prepareResponse.dataAddress()).isNull();
 
         var providerProcessId = "provider_" + processId;
-        var startMessage = startMessage(providerProcessId, transferType);
+        var startMessage = createStartMessage(providerProcessId, controlPlane.providerCallbackAddress(), transferType);
         var startResponse = controlPlane.providerStart(startMessage).statusCode(200).extract().as(DataFlowStatusMessage.class);
         assertThat(startResponse.state()).isEqualTo(STARTED.name());
         assertThat(startResponse.dataAddress()).isNotNull();
@@ -141,18 +140,6 @@ void shouldSuspendAndResumeOnProvider() {
         consumerDataPlane.assertDataIsFlowing();
     }
 
-    private DataFlowPrepareMessage prepareMessage(String consumerProcessId, String transferType) {
-        return new DataFlowPrepareMessage("theMessageId", "theParticipantId", "theCounterPartyId",
-                "theDataspaceContext", consumerProcessId, "theAgreementId", "theDatasetId", controlPlane.consumerCallbackAddress(),
-                transferType, emptyList(), emptyMap());
-    }
-
-    private DataFlowStartMessage startMessage(String providerProcessId, String transferType) {
-        return new DataFlowStartMessage("theMessageId", "theParticipantId", "theCounterPartyId",
-                "theDataspaceContext", providerProcessId, "theAgreementId", "theDatasetId", controlPlane.providerCallbackAddress(),
-                transferType, null, emptyList(), emptyMap());
-    }
-
     private DataFlowResumeMessage resumeMessage(String providerProcessId) {
         return new DataFlowResumeMessage("theMessageId", providerProcessId, null);
     }