Skip to content

Commit

Permalink
feat: stop data flow when policy is not valid anymore
Browse files Browse the repository at this point in the history
  • Loading branch information
@ndr-brt
ndr-brt committed Sep 27, 2023
1 parent 3fb0dc2 commit 890fe64
Show file tree
Hide file tree
Showing 41 changed files with 753 additions and 361 deletions.
25 changes: 18 additions & 7 deletions ...nsfer-core/src/main/java/org/eclipse/edc/connector/transfer/flow/DataFlowManagerImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.function.Function;

import static java.lang.String.format;
import static org.eclipse.edc.spi.response.ResponseStatus.FATAL_ERROR;
Expand All @@ -51,18 +52,28 @@ public void register(int priority, DataFlowController controller) {
@Override
public @NotNull StatusResult<DataFlowResponse> initiate(TransferProcess transferProcess, Policy policy) {
try {
return controllers.stream()
.sorted(Comparator.comparingInt(a -> -a.priority))
.map(PrioritizedDataFlowController::controller)
.filter(controller -> controller.canHandle(transferProcess))
.findFirst()
.map(controller -> controller.initiateFlow(transferProcess, policy))
.orElseGet(() -> StatusResult.failure(FATAL_ERROR, controllerNotFound(transferProcess.getId())));
return chooseControllerAndApply(transferProcess, controller -> controller.initiateFlow(transferProcess, policy));
} catch (Exception e) {
return StatusResult.failure(FATAL_ERROR, runtimeException(transferProcess.getId(), e.getLocalizedMessage()));
}
}

@Override
public @NotNull StatusResult<Void> terminate(TransferProcess transferProcess) {
return chooseControllerAndApply(transferProcess, controller -> controller.terminate(transferProcess));
}

@NotNull
private <T> StatusResult<T> chooseControllerAndApply(TransferProcess transferProcess, Function<DataFlowController, StatusResult<T>> function) {
return controllers.stream()
.sorted(Comparator.comparingInt(a -> -a.priority))
.map(PrioritizedDataFlowController::controller)
.filter(controller -> controller.canHandle(transferProcess))
.findFirst()
.map(function)
.orElseGet(() -> StatusResult.failure(FATAL_ERROR, controllerNotFound(transferProcess.getId())));
}

private String runtimeException(String id, String message) {
return format("Unable to process transfer %s. Data flow controller throws an exception: %s", id, message);
}
Expand Down
99 changes: 61 additions & 38 deletions .../src/main/java/org/eclipse/edc/connector/transfer/process/TransferProcessManagerImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -127,7 +127,8 @@ protected StateMachineManager.Builder configureStateMachineManager(StateMachineM
.processor(processConsumerTransfersInState(REQUESTING, this::processRequesting))
.processor(processProviderTransfersInState(STARTING, this::processStarting))
.processor(processConsumerTransfersInState(STARTED, this::processStarted))
.processor(processTransfersInState(COMPLETING, this::processCompleting))
.processor(processProviderTransfersInState(COMPLETING, this::processProviderCompleting))
.processor(processConsumerTransfersInState(COMPLETING, this::processConsumerCompleting))
.processor(processTransfersInState(TERMINATING, this::processTerminating))
.processor(processTransfersInState(DEPROVISIONING, this::processDeprovisioning));
}
Expand Down Expand Up @@ -324,27 +325,6 @@ private boolean processStarting(TransferProcess process) {
.execute(description);
}

@WithSpan
private void sendTransferStartMessage(TransferProcess process, DataFlowResponse dataFlowResponse, Policy policy) {
var message = TransferStartMessage.Builder.newInstance()
.processId(process.getCorrelationId())
.protocol(process.getProtocol())
.dataAddress(dataFlowResponse.getDataAddress())
.counterPartyAddress(process.getConnectorAddress())
.policy(policy)
.build();

var description = format("Send %s to %s", message.getClass().getSimpleName(), process.getConnectorAddress());

entityRetryProcessFactory.doAsyncStatusResultProcess(process, () -> dispatcherRegistry.dispatch(Object.class, message))
.entityRetrieve(id -> store.findById(id))
.onSuccess((t, content) -> transitionToStarted(t))
.onFailure((t, throwable) -> transitionToStarting(t))
.onFatalError((n, failure) -> transitionToTerminated(n, failure.getFailureDetail()))
.onRetryExhausted((t, throwable) -> transitionToTerminating(t, throwable.getMessage(), throwable))
.execute(description);
}

/**
* Process STARTED transfer<p> if is completed or there's no checker and it's not managed, set to COMPLETE,
* nothing otherwise.
Expand Down Expand Up @@ -377,28 +357,30 @@ private Boolean checkCompletion(TransferProcess transferProcess) {
}

/**
* Process COMPLETING transfer<p> Send COMPLETED message to counter-part
* Process COMPLETING transfer<p>. Terminate data flow and send COMPLETED message to counter-part
*
* @param process the COMPLETING transfer fetched
* @return if the transfer has been processed or not
*/
@WithSpan
private boolean processCompleting(TransferProcess process) {
var message = TransferCompletionMessage.Builder.newInstance()
.protocol(process.getProtocol())
.counterPartyAddress(process.getConnectorAddress())
.processId(process.getCorrelationId())
.policy(policyArchive.findPolicyForContract(process.getContractId()))
.build();
private boolean processProviderCompleting(TransferProcess process) {
return entityRetryProcessFactory.doSyncProcess(process, () -> dataFlowManager.terminate(process))
.onSuccess((p, c) -> sendTransferCompletionMessage(p))
.onFatalError((p, failure) -> transitionToTerminating(p, failure.getFailureDetail()))
.onFailure((t, failure) -> transitionToCompleting(t))
.onRetryExhausted((p, failure) -> transitionToTerminating(p, failure.getFailureDetail()))
.execute("Terminate data flow");
}

var description = format("Send %s to %s", message.getClass().getSimpleName(), process.getConnectorAddress());
return entityRetryProcessFactory.doAsyncStatusResultProcess(process, () -> dispatcherRegistry.dispatch(Object.class, message))
.entityRetrieve(id -> store.findById(id))
.onSuccess((t, content) -> transitionToCompleted(t))
.onFailure((t, throwable) -> transitionToCompleting(t))
.onFatalError((n, failure) -> transitionToTerminated(n, failure.getFailureDetail()))
.onRetryExhausted((t, throwable) -> transitionToTerminating(t, throwable.getMessage(), throwable))
.execute(description);
/**
* Process COMPLETING transfer<p> Send COMPLETED message to counter-part
*
* @param process the COMPLETING transfer fetched
* @return if the transfer has been processed or not
*/
@WithSpan
private boolean processConsumerCompleting(TransferProcess process) {
return sendTransferCompletionMessage(process);
}

/**
Expand Down Expand Up @@ -455,6 +437,47 @@ private boolean processDeprovisioning(TransferProcess process) {
.execute("deprovisioning");
}

@WithSpan
private void sendTransferStartMessage(TransferProcess process, DataFlowResponse dataFlowResponse, Policy policy) {
var message = TransferStartMessage.Builder.newInstance()
.processId(process.getCorrelationId())
.protocol(process.getProtocol())
.dataAddress(dataFlowResponse.getDataAddress())
.counterPartyAddress(process.getConnectorAddress())
.policy(policy)
.build();

var description = format("Send %s to %s", message.getClass().getSimpleName(), process.getConnectorAddress());

entityRetryProcessFactory.doAsyncStatusResultProcess(process, () -> dispatcherRegistry.dispatch(Object.class, message))
.entityRetrieve(id -> store.findById(id))
.onSuccess((t, content) -> transitionToStarted(t))
.onFailure((t, throwable) -> transitionToStarting(t))
.onFatalError((n, failure) -> transitionToTerminated(n, failure.getFailureDetail()))
.onRetryExhausted((t, throwable) -> transitionToTerminating(t, throwable.getMessage(), throwable))
.execute(description);
}

@WithSpan
private boolean sendTransferCompletionMessage(TransferProcess process) {
var message = TransferCompletionMessage.Builder.newInstance()
.protocol(process.getProtocol())
.counterPartyAddress(process.getConnectorAddress())
.processId(process.getCorrelationId())
.policy(policyArchive.findPolicyForContract(process.getContractId()))
.build();

var description = format("Send %s to %s", message.getClass().getSimpleName(), process.getConnectorAddress());

return entityRetryProcessFactory.doAsyncStatusResultProcess(process, () -> dispatcherRegistry.dispatch(Object.class, message))
.entityRetrieve(id -> store.findById(id))
.onSuccess((t, content) -> transitionToCompleted(t))
.onFailure((t, throwable) -> transitionToCompleting(t))
.onFatalError((n, failure) -> transitionToTerminated(n, failure.getFailureDetail()))
.onRetryExhausted((t, throwable) -> transitionToTerminating(t, throwable.getMessage(), throwable))
.execute(description);
}

private <T> void handleResult(TransferProcess transferProcess, List<StatusResult<T>> responses, ResponsesHandler<StatusResult<T>> handler) {
if (handler.handle(transferProcess, responses)) {
update(transferProcess);
Expand Down
26 changes: 22 additions & 4 deletions ...r-core/src/test/java/org/eclipse/edc/connector/transfer/flow/DataFlowManagerImplTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.junit.jupiter.api.Test;

import static org.assertj.core.api.Assertions.assertThat;
import static org.eclipse.edc.junit.assertions.AbstractResultAssert.assertThat;
import static org.eclipse.edc.spi.response.ResponseStatus.FATAL_ERROR;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
Expand All @@ -37,7 +38,7 @@ class DataFlowManagerImplTest {
private final DataFlowManagerImpl manager = new DataFlowManagerImpl();

@Test
void should_initiate_flow_on_correct_controller() {
void initiate_shouldInitiateFlowOnCorrectController() {
var controller = mock(DataFlowController.class);
var dataRequest = DataRequest.Builder.newInstance().destinationType("test-dest-type").build();
var policy = Policy.Builder.newInstance().build();
Expand All @@ -54,7 +55,7 @@ void should_initiate_flow_on_correct_controller() {
}

@Test
void should_return_fatal_error_if_no_controller_can_handle_the_request() {
void initiate_shouldReturnFatalError_whenNoControllerCanHandleTheRequest() {
var controller = mock(DataFlowController.class);
var dataRequest = DataRequest.Builder.newInstance().destinationType("test-dest-type").build();
var dataAddress = DataAddress.Builder.newInstance().type("test-type").build();
Expand All @@ -71,7 +72,7 @@ void should_return_fatal_error_if_no_controller_can_handle_the_request() {
}

@Test
void should_catch_exceptions_and_return_fatal_error() {
void initiate_shouldCatchExceptionsAndReturnFatalError() {
var controller = mock(DataFlowController.class);
var dataRequest = DataRequest.Builder.newInstance().destinationType("test-dest-type").build();
var dataAddress = DataAddress.Builder.newInstance().type("test-type").build();
Expand All @@ -91,7 +92,7 @@ void should_catch_exceptions_and_return_fatal_error() {
}

@Test
void shouldChooseHighestPriorityController() {
void initiate_shouldChooseHighestPriorityController() {
var highPriority = createDataFlowController();
var lowPriority = createDataFlowController();
manager.register(1, lowPriority);
Expand All @@ -103,6 +104,23 @@ void shouldChooseHighestPriorityController() {
verifyNoInteractions(lowPriority);
}

@Test
void terminate_shouldChooseControllerAndTerminate() {
var controller = mock(DataFlowController.class);
var dataRequest = DataRequest.Builder.newInstance().destinationType("test-dest-type").build();
var dataAddress = DataAddress.Builder.newInstance().type("test-type").build();
var transferProcess = TransferProcess.Builder.newInstance().dataRequest(dataRequest).contentDataAddress(dataAddress).build();

when(controller.canHandle(any())).thenReturn(true);
when(controller.terminate(any())).thenReturn(StatusResult.success());
manager.register(controller);

var result = manager.terminate(transferProcess);

assertThat(result).isSucceeded();
verify(controller).terminate(transferProcess);
}

private DataFlowController createDataFlowController() {
var dataFlowController = mock(DataFlowController.class);
when(dataFlowController.canHandle(any())).thenReturn(true);
Expand Down
44 changes: 41 additions & 3 deletions .../test/java/org/eclipse/edc/connector/transfer/process/TransferProcessManagerImplTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,6 @@ class TransferProcessManagerImplTest {
@BeforeEach
void setup() {
when(protocolWebhook.url()).thenReturn(protocolWebhookUrl);
when(dataFlowManager.initiate(any(), any())).thenReturn(StatusResult.success(createDataFlowResponse()));
var observable = new TransferProcessObservableImpl();
observable.registerListener(listener);
var entityRetryProcessConfiguration = new EntityRetryProcessConfiguration(RETRY_LIMIT, () -> new ExponentialWaitStrategy(0L));
Expand Down Expand Up @@ -625,15 +624,52 @@ void started_shouldBreakLeaseIfNotConsumer() {
}

@Test
void completing_shouldTransitionToCompleted_whenSendingMessageSucceed() {
void completing_provider_shouldTerminateDataTransferAndTransitionToCompleted() {
var process = createTransferProcessBuilder(COMPLETING).dataRequest(createDataRequestBuilder().id("correlationId").build()).build();
when(transferProcessStore.nextNotLeased(anyInt(), stateIs(COMPLETING.code()))).thenReturn(List.of(process)).thenReturn(emptyList());
when(transferProcessStore.nextNotLeased(anyInt(), providerStateIs(COMPLETING.code()))).thenReturn(List.of(process)).thenReturn(emptyList());
when(transferProcessStore.findById(process.getId())).thenReturn(process, process.toBuilder().state(COMPLETING.code()).build());
when(dataFlowManager.terminate(any())).thenReturn(StatusResult.success());
when(dispatcherRegistry.dispatch(any(), isA(TransferCompletionMessage.class))).thenReturn(completedFuture(StatusResult.success("any")));

manager.start();

await().untilAsserted(() -> {
verify(dataFlowManager).terminate(process);
var captor = ArgumentCaptor.forClass(TransferCompletionMessage.class);
verify(dispatcherRegistry).dispatch(eq(Object.class), captor.capture());
var message = captor.getValue();
assertThat(message.getProcessId()).isEqualTo("correlationId");
verify(transferProcessStore, times(RETRY_LIMIT)).save(argThat(p -> p.getState() == COMPLETED.code()));
verify(listener).completed(process);
});
}

@Test
void completing_provider_whenDataFlowTerminationFailsAndRetriesNotExhausted_updatesStateCountForRetry() {
var process = createTransferProcess(COMPLETING).toBuilder().type(PROVIDER).build();
when(dataFlowManager.terminate(any())).thenReturn(StatusResult.failure(ResponseStatus.ERROR_RETRY));
when(transferProcessStore.nextNotLeased(anyInt(), providerStateIs(COMPLETING.code()))).thenReturn(List.of(process)).thenReturn(emptyList());
when(transferProcessStore.findById(process.getId())).thenReturn(process, process.toBuilder().state(COMPLETING.code()).build());

manager.start();

await().untilAsserted(() -> {
verify(transferProcessStore, times(RETRY_LIMIT)).save(argThat(p -> p.getState() == COMPLETING.code()));
});
}

@Test
void completing_consumer_shouldTransitionToCompletedAndNotTerminateDataTransfer() {
var process = createTransferProcessBuilder(COMPLETING).dataRequest(createDataRequestBuilder().id("correlationId").build()).build();
when(transferProcessStore.nextNotLeased(anyInt(), consumerStateIs(COMPLETING.code()))).thenReturn(List.of(process)).thenReturn(emptyList());
when(transferProcessStore.findById(process.getId())).thenReturn(process, process.toBuilder().state(COMPLETING.code()).build());
when(dataFlowManager.terminate(any())).thenReturn(StatusResult.success());
when(dispatcherRegistry.dispatch(any(), isA(TransferCompletionMessage.class))).thenReturn(completedFuture(StatusResult.success("any")));

manager.start();

await().untilAsserted(() -> {
verifyNoInteractions(dataFlowManager);
var captor = ArgumentCaptor.forClass(TransferCompletionMessage.class);
verify(dispatcherRegistry).dispatch(eq(Object.class), captor.capture());
var message = captor.getValue();
Expand Down Expand Up @@ -764,6 +800,8 @@ void dispatchFailure(TransferProcessStates starting, TransferProcessStates endin
.thenReturn(List.of(transferProcess)).thenReturn(emptyList());
when(dispatcherRegistry.dispatch(any(), any())).thenReturn(result);
when(transferProcessStore.findById(transferProcess.getId())).thenReturn(transferProcess);
when(dataFlowManager.initiate(any(), any())).thenReturn(StatusResult.success(createDataFlowResponse()));
when(dataFlowManager.terminate(any())).thenReturn(StatusResult.success());

manager.start();

Expand Down
4 changes: 1 addition & 3 deletions .../main/java/org/eclipse/edc/connector/dataplane/framework/DataPlaneFrameworkExtension.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@
import org.eclipse.edc.connector.api.client.spi.transferprocess.TransferProcessApiClient;
import org.eclipse.edc.connector.dataplane.framework.manager.DataPlaneManagerImpl;
import org.eclipse.edc.connector.dataplane.framework.pipeline.PipelineServiceImpl;
import org.eclipse.edc.connector.dataplane.framework.pipeline.PipelineServiceTransferServiceImpl;
import org.eclipse.edc.connector.dataplane.framework.registry.TransferServiceRegistryImpl;
import org.eclipse.edc.connector.dataplane.framework.registry.TransferServiceSelectionStrategy;
import org.eclipse.edc.connector.dataplane.spi.manager.DataPlaneManager;
Expand Down Expand Up @@ -100,10 +99,9 @@ public void initialize(ServiceExtensionContext context) {
var pipelineService = new PipelineServiceImpl(monitor);
pipelineService.registerFactory(new OutputStreamDataSinkFactory()); // Added by default to support synchronous data transfer, i.e. pull data
context.registerService(PipelineService.class, pipelineService);
var transferService = new PipelineServiceTransferServiceImpl(pipelineService);

var transferServiceRegistry = new TransferServiceRegistryImpl(transferServiceSelectionStrategy);
transferServiceRegistry.registerTransferService(transferService);
transferServiceRegistry.registerTransferService(pipelineService);
context.registerService(TransferServiceRegistry.class, transferServiceRegistry);

var numThreads = context.getSetting(TRANSFER_THREADS, DEFAULT_TRANSFER_THREADS);
Expand Down
Loading

0 comments on commit 890fe64

Please sign in to comment.