feat: support String-Array query parameter in DP public API (#4051)

* feat: support String-Array query parameter in DP public API

* checkstyle

* DEPENDENCIES

* improve verification

DEPENDENCIES

@@ -224,7 +224,7 @@ maven/mavencentral/org.apache.velocity/velocity-engine-scripting/2.3, Apache-2.0
 maven/mavencentral/org.apache.xbean/xbean-reflect/3.7, Apache-2.0, approved, clearlydefined
 maven/mavencentral/org.apiguardian/apiguardian-api/1.1.2, Apache-2.0, approved, clearlydefined
 maven/mavencentral/org.assertj/assertj-core/3.25.3, Apache-2.0, approved, #12585
-maven/mavencentral/org.awaitility/awaitility/4.2.0, Apache-2.0, approved, clearlydefined
+maven/mavencentral/org.awaitility/awaitility/4.2.0, Apache-2.0, approved, #14178
 maven/mavencentral/org.bouncycastle/bcpkix-jdk18on/1.72, MIT, approved, #3789
 maven/mavencentral/org.bouncycastle/bcpkix-jdk18on/1.77, MIT, approved, #11593
 maven/mavencentral/org.bouncycastle/bcprov-jdk18on/1.72, MIT AND CC0-1.0, approved, #3538
@@ -347,7 +347,7 @@ maven/mavencentral/org.slf4j/slf4j-api/2.0.9, MIT, approved, #5915
 maven/mavencentral/org.testcontainers/database-commons/1.19.7, Apache-2.0, approved, #10345
 maven/mavencentral/org.testcontainers/jdbc/1.19.7, Apache-2.0, approved, #10348
 maven/mavencentral/org.testcontainers/junit-jupiter/1.19.7, MIT, approved, #10344
-maven/mavencentral/org.testcontainers/kafka/1.19.7, , restricted, clearlydefined
+maven/mavencentral/org.testcontainers/kafka/1.19.7, None, restricted, #14177
 maven/mavencentral/org.testcontainers/postgresql/1.19.7, MIT, approved, #10350
 maven/mavencentral/org.testcontainers/testcontainers/1.19.7, Apache-2.0 AND MIT, approved, #10347
 maven/mavencentral/org.testcontainers/vault/1.19.7, MIT, approved, #10852

extensions/data-plane/data-plane-public-api-v2/src/main/java/org/eclipse/edc/connector/dataplane/api/controller/ContainerRequestContextApiImpl.java

@@ -9,6 +9,7 @@
  *
  *  Contributors:
  *       Amadeus - initial API and implementation
+ *       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - allow multiple identical query params #4022
  *
  */
 
@@ -21,7 +22,6 @@
 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStreamReader;
-import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
@@ -51,8 +51,7 @@ public Map<String, String> headers() {
     public String queryParams() {
         return context.getUriInfo().getQueryParameters().entrySet()
                 .stream()
-                .map(entry -> new QueryParam(entry.getKey(), entry.getValue()))
-                .filter(QueryParam::isValid)
+                .flatMap(entry -> entry.getValue().stream().map(val -> new QueryParam(entry.getKey(), val)))
                 .map(QueryParam::toString)
                 .collect(Collectors.joining(QUERY_PARAM_SEPARATOR));
     }
@@ -66,19 +65,19 @@ public String body() {
         }
     }
 
-    @Override
-    public String path() {
-        var pathInfo = context.getUriInfo().getPath();
-        return pathInfo.startsWith("/") ? pathInfo.substring(1) : pathInfo;
-    }
-
     @Override
     public String mediaType() {
         return Optional.ofNullable(context.getMediaType())
                 .map(MediaType::toString)
                 .orElse(null);
     }
 
+    @Override
+    public String path() {
+        var pathInfo = context.getUriInfo().getPath();
+        return pathInfo.startsWith("/") ? pathInfo.substring(1) : pathInfo;
+    }
+
     @Override
     public String method() {
         return context.getMethod();
@@ -87,10 +86,10 @@ public String method() {
     private static final class QueryParam {
 
         private final String key;
-        private final List<String> values;
+        private final String values;
         private final boolean valid;
 
-        private QueryParam(String key, List<String> values) {
+        private QueryParam(String key, String values) {
             this.key = key;
             this.values = values;
             this.valid = key != null && values != null && !values.isEmpty();
@@ -102,7 +101,7 @@ public boolean isValid() {
 
         @Override
         public String toString() {
-            return valid ? key + "=" + values.get(0) : "";
+            return valid ? key + "=" + values : "";
         }
     }
 }

extensions/data-plane/data-plane-public-api-v2/src/main/java/org/eclipse/edc/connector/dataplane/api/controller/DataFlowRequestSupplier.java

@@ -9,6 +9,7 @@
  *
  *  Contributors:
  *       Amadeus - initial API and implementation
+ *       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - make it a PULL request by default
  *
  */
 
@@ -17,6 +18,7 @@
 import org.eclipse.edc.connector.dataplane.util.sink.AsyncStreamingDataSink;
 import org.eclipse.edc.spi.types.domain.DataAddress;
 import org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage;
+import org.eclipse.edc.spi.types.domain.transfer.FlowType;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -61,6 +63,7 @@ public DataFlowStartMessage apply(ContainerRequestContextApi contextApi, DataAdd
         return DataFlowStartMessage.Builder.newInstance()
                 .processId(UUID.randomUUID().toString())
                 .sourceDataAddress(dataAddress)
+                .flowType(FlowType.PULL) // if a request hits the public DP API, we can assume a PULL transfer
                 .destinationDataAddress(DataAddress.Builder.newInstance()
                         .type(AsyncStreamingDataSink.TYPE)
                         .build())

extensions/data-plane/data-plane-public-api-v2/src/main/java/org/eclipse/edc/connector/dataplane/api/controller/DataPlanePublicApiV2.java

@@ -9,6 +9,7 @@
  *
  *  Contributors:
  *       Amadeus - Initial implementation
+ *       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - handle HEAD requests
  *
  */
 
@@ -43,6 +44,15 @@ public interface DataPlanePublicApiV2 {
     )
     void get(ContainerRequestContext context, AsyncResponse response);
 
+    @Operation(description = "Send `HEAD` data query to the Data Plane.",
+            responses = {
+                    @ApiResponse(responseCode = "400", description = "Missing access token"),
+                    @ApiResponse(responseCode = "403", description = "Access token is expired or invalid"),
+                    @ApiResponse(responseCode = "500", description = "Failed to transfer data")
+            }
+    )
+    void head(ContainerRequestContext context, AsyncResponse response);
+
     @Operation(description = "Send `POST` data query to the Data Plane.",
             responses = {
                     @ApiResponse(responseCode = "400", description = "Missing access token"),

extensions/data-plane/data-plane-public-api-v2/src/main/java/org/eclipse/edc/connector/dataplane/api/controller/DataPlanePublicApiV2Controller.java

@@ -9,13 +9,15 @@
  *
  *  Contributors:
  *       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - initial API and implementation
+ *       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - handle HEAD requests
  *
  */
 
 package org.eclipse.edc.connector.dataplane.api.controller;
 
 import jakarta.ws.rs.DELETE;
 import jakarta.ws.rs.GET;
+import jakarta.ws.rs.HEAD;
 import jakarta.ws.rs.PATCH;
 import jakarta.ws.rs.POST;
 import jakarta.ws.rs.PUT;
@@ -74,6 +76,12 @@ public void get(@Context ContainerRequestContext requestContext, @Suspended Asyn
         handle(requestContext, response);
     }
 
+    @HEAD
+    @Override
+    public void head(@Context ContainerRequestContext requestContext, @Suspended AsyncResponse response) {
+        handle(requestContext, response);
+    }
+
     /**
      * Sends a {@link POST} request to the data source and returns data.
      *

system-tests/e2e-dataplane-tests/tests/src/test/java/org/eclipse/edc/test/e2e/DataPlanePublicApiEndToEndTest.java

@@ -29,9 +29,15 @@
 import org.eclipse.edc.spi.security.Vault;
 import org.eclipse.edc.spi.types.domain.DataAddress;
 import org.hamcrest.Matchers;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
+import org.mockserver.integration.ClientAndServer;
+import org.mockserver.model.HttpResponse;
+import org.mockserver.model.Parameter;
+import org.mockserver.verify.VerificationTimes;
 
 import java.security.Key;
 import java.security.PrivateKey;
@@ -40,17 +46,39 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.eclipse.edc.spi.CoreConstants.EDC_NAMESPACE;
+import static org.eclipse.edc.util.io.Ports.getFreePort;
 import static org.mockito.Mockito.mock;
+import static org.mockserver.model.HttpRequest.request;
+import static org.mockserver.model.JsonBody.json;
+import static org.mockserver.verify.VerificationTimes.exactly;
 
 public class DataPlanePublicApiEndToEndTest extends AbstractDataPlaneTest {
 
     public static final String PUBLIC_KEY_ALIAS = "public-key";
     public static final String PRIVATE_KEY_ALIAS = "1";
     // this is a data address representing the private backend for an HTTP pull transfer
-    public static final DataAddress BACKEND_API_DATAADDRESS = DataAddress.Builder.newInstance()
-            .type("HttpData")
-            .property(EDC_NAMESPACE + "baseUrl", "https://jsonplaceholder.typicode.com/todos")
-            .build();
+    public static DataAddress backendDataAddress;
+
+    private ClientAndServer backendServer;
+
+    @BeforeEach
+    void setup() {
+        backendServer = ClientAndServer.startClientAndServer(getFreePort());
+        backendDataAddress = DataAddress.Builder.newInstance()
+                .type("HttpData")
+                .property(EDC_NAMESPACE + "baseUrl", "http://localhost:%d/foo".formatted(backendServer.getPort()))
+                .property(EDC_NAMESPACE + "proxyQueryParams", "true")
+                .property(EDC_NAMESPACE + "proxyPath", "true")
+                .property(EDC_NAMESPACE + "proxyMethod", "true")
+                .build();
+
+        backendServer.when(request()).respond(HttpResponse.response().withBody("""
+                {
+                   "foo": "bar",
+                   "fizz": "buzz",
+                }
+                """).withStatusCode(200));
+    }
 
     @Test
     void httpPull_missingToken_expect401() {
@@ -67,6 +95,8 @@ void httpPull_missingToken_expect401() {
                 .then()
                 .statusCode(401)
                 .body(Matchers.containsString("Missing Authorization Header"));
+
+        backendServer.verify(request().withMethod("POST"), VerificationTimes.never());
     }
 
     @Test
@@ -86,20 +116,76 @@ void httpPull_invalidToken_expect403() {
                 .statusCode(403);
     }
 
+    @DisplayName("Test methods with request body")
     @ParameterizedTest(name = "Method = {0}")
-    @ValueSource(strings = { "GET", "POST", "PUT", "PATCH", "DELETE", "HEAD" })
-    void request_expect200(String method) {
+    @ValueSource(strings = { "POST", "PUT", "PATCH" })
+    void request_withBody_expect200(String method) {
+        backendDataAddress.getProperties().put(EDC_NAMESPACE + "proxyBody", "true");
+        backendDataAddress.getProperties().put(EDC_NAMESPACE + "mediaType", "application/json");
+
         var token = createEdr();
+        var jsonBody = """
+                { 
+                   "quizz": "quzz"
+                }
+                """;
         var body = DATAPLANE.getDataPlanePublicEndpoint()
                 .baseRequest()
                 .contentType(ContentType.JSON)
                 .header(HttpHeaders.AUTHORIZATION, token)
+                .body(jsonBody)
                 .request(method, "/v2/bar/baz")
                 .then()
                 .log().ifError()
                 .statusCode(200)
                 .extract().body().asString();
         assertThat(body).isNotNull();
+
+        backendServer.verify(request()
+                .withMethod(method)
+                .withPath("/foo/v2/bar/baz")
+                .withBody(json(jsonBody)), exactly(1));
+    }
+
+    @DisplayName("Test methods without request body")
+    @ParameterizedTest(name = "Method = {0}")
+    @ValueSource(strings = { "GET", "DELETE", "HEAD" })
+    void request_noBody_expect200(String method) {
+        var token = createEdr();
+        var body = DATAPLANE.getDataPlanePublicEndpoint()
+                .baseRequest()
+                .contentType(ContentType.JSON)
+                .header(HttpHeaders.AUTHORIZATION, token)
+                .request(method, "/v2/bar/baz")
+                .then()
+                .log().ifError()
+                .statusCode(200)
+                .extract().body().asString();
+        assertThat(body).isNotNull();
+
+        backendServer.verify(request().withMethod(method).withPath("/foo/v2/bar/baz"), exactly(1));
+    }
+
+    @Test
+    void request_getMultipleIdenticalQuery() {
+        var token = createEdr();
+        var body = DATAPLANE.getDataPlanePublicEndpoint()
+                .baseRequest()
+                .contentType(ContentType.JSON)
+                .header(HttpHeaders.AUTHORIZATION, token)
+                .request("GET", "/v2/bar/baz?foo=bar&foo=fizz&foo=buzz")
+                .then()
+                .log().ifError()
+                .statusCode(200)
+                .extract().body().asString();
+        assertThat(body).isNotNull();
+
+        backendServer.verify(request()
+                .withPath("/foo/v2/bar/baz")
+                .withQueryStringParameters(new Parameter("foo", "bar"),
+                        new Parameter("foo", "fizz"),
+                        new Parameter("foo", "buzz")
+                ), exactly(1));
     }
 
     private Key resolvePrivateKey() {
@@ -120,7 +206,7 @@ private String createEdr() {
 
         // store the EDR
         var accessTokenStore = runtime.getService(AccessTokenDataStore.class);
-        accessTokenStore.store(new AccessTokenData(tokenId, ClaimToken.Builder.newInstance().build(), BACKEND_API_DATAADDRESS));
+        accessTokenStore.store(new AccessTokenData(tokenId, ClaimToken.Builder.newInstance().build(), backendDataAddress));
         return jwt;
     }