fix: backport of transfer process termination on EDR expired

DEPENDENCIES

@@ -130,7 +130,7 @@ maven/mavencentral/io.opentelemetry.instrumentation/opentelemetry-instrumentatio
 maven/mavencentral/io.opentelemetry/opentelemetry-api/1.29.0, Apache-2.0, approved, #10088
 maven/mavencentral/io.opentelemetry/opentelemetry-context/1.29.0, Apache-2.0, approved, #10090
 maven/mavencentral/io.projectreactor.netty/reactor-netty-core/1.0.33, Apache-2.0, approved, #9687
-maven/mavencentral/io.projectreactor.netty/reactor-netty-http/1.0.33, Apache-2.0, approved, clearlydefined
+maven/mavencentral/io.projectreactor.netty/reactor-netty-http/1.0.33, Apache-2.0, approved, #11661
 maven/mavencentral/io.projectreactor/reactor-core/3.4.30, Apache-2.0, approved, #7517
 maven/mavencentral/io.rest-assured/json-path/5.3.1, Apache-2.0, approved, #9261
 maven/mavencentral/io.rest-assured/rest-assured-common/5.3.1, Apache-2.0, approved, #9264

edc-extensions/edr/edr-callback/src/main/java/org/eclipse/tractusx/edc/callback/TransferProcessLocalCallback.java

@@ -105,6 +105,15 @@ private void cleanOldEdr(String assetId, String agreementId) {
             monitor.debug(format("Expiring EDR for transfer process %s", entry.getTransferProcessId()));
             entry.transitionToExpired();
             edrCache.update(entry);
+
+            var transferProcess = transferProcessStore.findById(entry.getTransferProcessId());
+
+            if (transferProcess != null && transferProcess.canBeCompleted()) {
+                transferProcess.transitionCompleting();
+                transferProcessStore.save(transferProcess);
+            } else {
+                monitor.info(format("Cannot terminate transfer process with id: %s", entry.getTransferProcessId()));
+            }
         }));
     }
 

edc-tests/e2e-tests/src/test/java/org/eclipse/tractusx/edc/lifecycle/Participant.java

@@ -283,6 +283,17 @@ public String getTransferProcessState(String id) {
                 .extract().body().jsonPath().getString("'edc:state'");
     }
 
+    public JsonArray getAllTransferProcess() {
+        return baseRequest()
+                .when()
+                .post("/v2/transferprocesses/request")
+                .then()
+                .statusCode(200)
+                .extract()
+                .body()
+                .as(JsonArray.class);
+    }
+
     public EndpointDataReference getDataReference(String dataRequestId) {
         var dataReference = new AtomicReference<EndpointDataReference>();
 

edc-tests/e2e-tests/src/test/java/org/eclipse/tractusx/edc/tests/edr/AbstractRenewalEdrTest.java

@@ -20,6 +20,7 @@
 import okhttp3.mockwebserver.MockWebServer;
 import org.assertj.core.api.Condition;
 import org.eclipse.edc.connector.transfer.spi.event.TransferProcessCompleted;
+import org.eclipse.edc.connector.transfer.spi.types.TransferProcessStates;
 import org.eclipse.edc.policy.model.Operator;
 import org.eclipse.tractusx.edc.lifecycle.Participant;
 import org.junit.jupiter.api.AfterEach;
@@ -38,6 +39,7 @@
 import static org.assertj.core.api.Assertions.anyOf;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
+import static org.awaitility.pollinterval.FibonacciPollInterval.fibonacci;
 import static org.eclipse.edc.spi.CoreConstants.EDC_NAMESPACE;
 import static org.eclipse.tractusx.edc.edr.spi.types.EndpointDataReferenceEntryStates.EXPIRED;
 import static org.eclipse.tractusx.edc.edr.spi.types.EndpointDataReferenceEntryStates.NEGOTIATED;
@@ -106,20 +108,48 @@ void negotiateEdr_shouldRenewTheEdr() throws IOException {
 
         assertThat(expectedEvents).usingRecursiveFieldByFieldElementComparator().containsAll(events);
 
-        JsonArrayBuilder edrCaches = Json.createArrayBuilder();
+        JsonArrayBuilder edrCachesBuilder = Json.createArrayBuilder();
 
         await().atMost(ASYNC_TIMEOUT)
                 .untilAsserted(() -> {
                     var localEdrCaches = SOKRATES.getEdrEntriesByAssetId(assetId);
                     assertThat(localEdrCaches).hasSizeGreaterThan(1);
-                    localEdrCaches.forEach(edrCaches::add);
+                    localEdrCaches.forEach(edrCachesBuilder::add);
                 });
 
+        var edrCaches = edrCachesBuilder.build();
 
-        assertThat(edrCaches.build())
+        assertThat(edrCaches)
                 .extracting(json -> json.asJsonObject().getJsonString("tx:edrState").getString())
                 .areAtMost(1, anyOf(stateCondition(NEGOTIATED.name(), "Negotiated"), stateCondition(REFRESHING.name(), "Refreshing")))
                 .areAtLeast(1, stateCondition(EXPIRED.name(), "Expired"));
+
+        var transferProcessId = edrCaches.stream()
+                .filter(json -> json.asJsonObject().getJsonString("tx:edrState").getString().equals(EXPIRED.name()))
+                .map(json -> json.asJsonObject().getJsonString("edc:transferProcessId").getString())
+                .findFirst()
+                .orElseThrow();
+
+        await().pollInterval(fibonacci())
+                .atMost(ASYNC_TIMEOUT)
+                .untilAsserted(() -> {
+                    var tpState = SOKRATES.getTransferProcessState(transferProcessId);
+                    assertThat(tpState).isNotNull().isEqualTo(TransferProcessStates.COMPLETED.toString());
+                });
+
+        await().pollInterval(fibonacci())
+                .atMost(ASYNC_TIMEOUT)
+                .untilAsserted(() -> {
+                    var tpState = PLATO.getAllTransferProcess()
+                            .stream()
+                            .filter(json -> json.asJsonObject().getJsonString("edc:correlationId").getString().equals(transferProcessId))
+                            .map(json -> json.asJsonObject().getJsonString("edc:state").getString())
+                            .findFirst();
+
+                    assertThat(tpState)
+                            .isPresent()
+                            .hasValue(TransferProcessStates.COMPLETED.toString());
+                });
     }