Termination of pull transfer process from consumer side not success in real but stated as "TERMINATED".

[AndrYurk]

Bug Report

Describe the Bug

During termination PULL transfer process from consumer, system does not call method terminate() from DataFlowManager on provider side.
As a result process not terminated in real but transfer process had state "TERMINATED", and then it is not possible to initiate terminating process again not from consumer neither form provider side.

Expected Behavior

Terminating from consumer should invoke org.eclipse.edc.connector.controlplane.transfer.flow.DataFlowManagerImpl#terminate on provider side.

Observed Behavior

Terminating from consumer does not invoke org.eclipse.edc.connector.controlplane.transfer.flow.DataFlowManagerImpl#terminate on provider side.

Steps to Reproduce

End to end test for tractusx-edc/edc-tests/edc-controlplane/transfer-tests/TransferPullEndToEndTest

@Test
        void transferData_withTerminateByConsumer() {
            var assetId = "api-asset-1";

            var requestDefinition = request().withMethod("GET").withPath(MOCK_BACKEND_PATH);

            Map<String, Object> dataAddress = Map.of(
                    "baseUrl", privateBackendUrl,
                    "type", "HttpData",
                    "contentType", "application/json"
            );

            PROVIDER.createAsset(assetId, Map.of(), dataAddress);

            var accessPolicyId = PROVIDER.createPolicyDefinition(createAccessPolicy(CONSUMER.getBpn()));
            var contractPolicyId = PROVIDER.createPolicyDefinition(createContractPolicy(CONSUMER.getBpn()));
            PROVIDER.createContractDefinition(assetId, "def-1", accessPolicyId, contractPolicyId);
            var transferProcessId = CONSUMER.requestAssetFrom(assetId, PROVIDER).withTransferType("HttpData-PULL").execute();

            CONSUMER.waitForTransferProcess(transferProcessId, TransferProcessStates.STARTED);

            // wait until EDC is available on the consumer side
            server.when(requestDefinition).respond(response().withStatusCode(200).withBody("test response"));

            var edr = CONSUMER.edrs().waitForEdr(transferProcessId);

            // consumer can fetch data with a valid token
            var data = CONSUMER.data().pullData(edr, Map.of());
            assertThat(data).isNotNull().isEqualTo("test response");

            server.verify(requestDefinition, VerificationTimes.exactly(1));

            JsonObjectBuilder requestBodyBuilder = Json.createObjectBuilder().add("@context", Json.createObjectBuilder()
                    .add("@vocab", "https://w3id.org/edc/v0.0.1/ns/")).add("@type", "TerminateTransfer").add("reason", "reason");
            CONSUMER.getManagementEndpoint().baseRequest().contentType(ContentType.JSON).body(requestBodyBuilder.build())
                    .when().post("/v3/transferprocesses/{id}/terminate", new Object[]{transferProcessId}).then().log().ifError().statusCode(204);

            CONSUMER.waitForTransferProcess(transferProcessId, TransferProcessStates.TERMINATED);

            // consumer cannot fetch data with the prev token (suspended)
            CONSUMER.data().pullDataRequest(edr, Map.of()).statusCode(403);
            server.verify(requestDefinition, VerificationTimes.exactly(1));
        }

Manual scenario:

1. Create simple asset, policy and contract definition(asset, policy_definition, contract_definition) on provider EDC.
2. Get policy_ID and initiate cotract negotiation(query_catalog, init_contract_negotiation) on consumer EDC.
3. Get contractAgreementId(get_contract_negotiations) using ID from initiated transfer.
4. Initiate HTTP PULL transfer(pull_transfer) on consumer EDC using.
5. Set breakpiont in org.eclipse.edc.connector.controlplane.transfer.flow.DataFlowManagerImpl#terminate.
6. Terminate transfer process using pull_transfer_ID(terminate_process) on consumer EDC. Breakpiont was not reached.
7. Check status process and get correlationId (get_transfer_process) using pull_transfer_ID on consumer EDC. State is "TERMINATED".
8. Check status process(get_transfer_process) using correlationId on provider EDC. State is "TERMINATED".
   Termination failed request command.txt

Context Information

Reproduced on EDC 0.7.2 as well as on 0.10.0

Possible Implementation

In method org.eclipse.edc.connector.controlplane.services.transferprocess.TransferProcessProtocolServiceImpl#terminatedAction add call to org.eclipse.edc.connector.controlplane.transfer.flow.DataFlowManagerImpl#terminate for provider. The same as implemented in org.eclipse.edc.connector.controlplane.services.transferprocess.TransferProcessProtocolServiceImpl#suspendedAction

[jimmarino]

I added the triage label back as I think we need to discuss the proper way to handle situations where the data plane does not cleanly terminate. See my comments in the proposed PR.