StandardFDConnectionService extends BaseComponentService<VersionedConnection, FDConnection> implements com.cloudera.cem.efm.service.component.FDConnectionService { @Override public FDConnection update(final String flowId, final VersionedConnection requestComponentConfig) { final FDFlow currentFlow = getFlowOrNotFound(flowId); final VersionedProcessGroup flowContents = currentFlow.getFlowContent(); final String componentId = requestComponentConfig.getIdentifier(); final VersionedConnection existingConnection = ComponentUtils.getComponentOrNotFound( componentId, flowContents, VersionedProcessGroup::getConnections); final String connectionGroupId = existingConnection.getGroupIdentifier(); final VersionedProcessGroup connectionGroup = ComponentUtils.getProcessGroupOrIllegalArgument(connectionGroupId, flowContents); final ConnectableComponent requestSource = requestComponentConfig.getSource(); final ConnectableComponent existingSource = existingConnection.getSource(); if (isNotNull(requestSource) && isNotNull(requestSource.getId()) && !requestSource.getId().equals(existingSource.getId())) { throw new IllegalStateException("Connection with id " + existingConnection.getIdentifier() + " has conflicting source id"); } if (existingSource.getType() == ConnectableComponentType.PROCESSOR && isNotNull(requestComponentConfig.getSelectedRelationships()) && requestComponentConfig.getSelectedRelationships().isEmpty()) { throw new IllegalArgumentException("Selected relationships cannot be empty"); } validateRequestConfig(requestComponentConfig); final ConnectableComponent proposedDestination = requestComponentConfig.getDestination(); if (proposedDestination != null) { if (proposedDestination.getId() == null) { throw new IllegalArgumentException("Proposed destination must contain an id"); } if (proposedDestination.getType() == null) { throw new IllegalArgumentException("Proposed destination must contain a type"); } if (proposedDestination.getType() == ConnectableComponentType.FUNNEL && proposedDestination.getId().equals(existingSource.getId())) { throw new IllegalArgumentException("Connecting a funnel to itself is not allowed"); } final ConnectableComponent currentDestination = existingConnection.getDestination(); if (ConnectableComponentType.REMOTE_OUTPUT_PORT == proposedDestination.getType()) { throw new IllegalArgumentException("Destination cannot be a remote output port"); } else if (ConnectableComponentType.REMOTE_INPUT_PORT == proposedDestination.getType()) { final String proposedDestinationId = proposedDestination.getId(); final String proposedDestinationGroupId = proposedDestination.getGroupId(); try { UUID.fromString(proposedDestinationId); } catch (Exception e) { throw new IllegalArgumentException("The ID of a remote input port must be a UUID"); } if (proposedDestinationGroupId == null) { throw new IllegalArgumentException("When the destination is a remote input port its group id is required."); } boolean isDifferentRemoteProcessGroup = false; if (currentDestination.getType() == ConnectableComponentType.REMOTE_INPUT_PORT) { if (!proposedDestinationGroupId.equals(currentDestination.getGroupId())) { isDifferentRemoteProcessGroup = true; } } if (!proposedDestinationId.equals(currentDestination.getId()) || isDifferentRemoteProcessGroup) { final VersionedRemoteProcessGroup proposedRemoteProcessGroup = connectionGroup.getRemoteProcessGroups().stream() .filter(rpg -> rpg.getIdentifier().equals(proposedDestinationGroupId)) .findFirst() .orElseThrow(() -> new IllegalArgumentException("Unable to find the specified remote process group for the destination")); final Optional<VersionedRemoteGroupPort> remoteInputPort = proposedRemoteProcessGroup.getInputPorts().stream() .filter(p -> p.getTargetId().equals(proposedDestinationId)) .findFirst(); final VersionedRemoteGroupPort newRemoteInputPort = new VersionedRemoteGroupPort(); if (!remoteInputPort.isPresent()) { newRemoteInputPort.setIdentifier(UUID.randomUUID().toString()); newRemoteInputPort.setGroupIdentifier(connectionGroup.getIdentifier()); newRemoteInputPort.setRemoteGroupId(proposedDestinationGroupId); newRemoteInputPort.setTargetId(proposedDestinationId); newRemoteInputPort.setName(proposedDestinationId); newRemoteInputPort.setComponentType(ComponentType.REMOTE_INPUT_PORT); } if (!isDestStillUsed(existingConnection, currentDestination, connectionGroup)) { removeRemoteInputPort(connectionGroup, currentDestination); } proposedRemoteProcessGroup.getInputPorts().add(newRemoteInputPort); proposedDestination.setName(proposedDestinationId); existingConnection.setDestination(proposedDestination); } } else { if (!proposedDestination.getId().equals(currentDestination.getId())) { if (proposedDestination.getGroupId() == null) { proposedDestination.setGroupId(existingConnection.getGroupIdentifier()); } final VersionedProcessGroup destinationGroup = ComponentUtils.getProcessGroupOrIllegalArgument( proposedDestination.getGroupId(), currentFlow.getFlowContent()); final Optional<VersionedComponent> newDestination = getVersionedComponentConnectable( destinationGroup, proposedDestination); if (!newDestination.isPresent()) { throw new IllegalArgumentException("Unable to find the specified destination."); } final String existingConnectionGroupId = existingConnection.getGroupIdentifier(); final String newDestinationGroupId = newDestination.get().getGroupIdentifier(); if (!existingConnectionGroupId.equals(newDestinationGroupId)) { final ComponentType newDestinationComponentType = newDestination.get().getComponentType(); if (newDestinationComponentType != ComponentType.INPUT_PORT) { throw new IllegalStateException("Cannot update connection because the destination is in " + "a different process group and is not an input port"); } final boolean connectionGroupContainsDestGroup = containsChildGroup(connectionGroup, newDestinationGroupId); if (!connectionGroupContainsDestGroup) { throw new IllegalStateException("Cannot update connection because the new destination " + "group is not a child group of the connection group"); } } existingConnection.setDestination(proposedDestination); } } } final FDConnection connection = updateComponent(currentFlow, requestComponentConfig, existingConnection, null); getValidationService().clearValidationErrors(connection.getComponentConfiguration().getSource().getId()); getValidationService().clearValidationErrors(connection.getComponentConfiguration().getDestination().getId()); return connection; } StandardFDConnectionService(final FDFlowManager flowManager, final ValidationService validationService); @Override FDConnection create(final String flowId, final String processGroupId, final VersionedConnection requestComponentConfig); @Override FDConnection get(final String flowId, final String componentId); @Override FDConnection update(final String flowId, final VersionedConnection requestComponentConfig); @Override void delete(final String flowId, final String componentId); static final String DEFAULT_BACK_PRESSURE_DATA_SIZE_THRESHOLD; static final long DEFAULT_BACK_PRESSURE_OBJECT_SIZE_THRESHOLD; static final String DEFAULT_FLOW_FILE_EXPIRATION; }
@Test public void testUpdateToValidDestination() { final String flowId = flow.getFlowMetadata().getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); final ConnectableComponent newDest = ComponentUtils.createConnectableComponent(group2Processor1); final VersionedConnection requestConnection = new VersionedConnection(); requestConnection.setIdentifier(connectionG2P2toG2P3.getIdentifier()); requestConnection.setSource(connectionG2P2toG2P3.getSource()); requestConnection.setDestination(newDest); final FDConnection updatedConnection = connectionService.update(flowId, requestConnection); assertNotNull(updatedConnection); }
@Test public void testUpdateToValidDestinationWhenNoSourceProvided() { final String flowId = flow.getFlowMetadata().getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); final ConnectableComponent newDest = ComponentUtils.createConnectableComponent(group2Processor1); final VersionedConnection requestConnection = new VersionedConnection(); requestConnection.setIdentifier(connectionG2P2toG2P3.getIdentifier()); requestConnection.setDestination(newDest); final FDConnection updatedConnection = connectionService.update(flowId, requestConnection); assertNotNull(updatedConnection); }
@Test(expected = IllegalStateException.class) public void testUpdatedWhenSourceChanges() { final String flowId = flow.getFlowMetadata().getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); final ConnectableComponent newSource = ComponentUtils.createConnectableComponent(group2Processor1); final VersionedConnection requestConnection = new VersionedConnection(); requestConnection.setIdentifier(connectionG2P2toG2P3.getIdentifier()); requestConnection.setSource(newSource); requestConnection.setDestination(connectionG2P2toG2P3.getDestination()); connectionService.update(flowId, requestConnection); }
@Test(expected = ResourceNotFoundException.class) public void testUpdateWhenConnectionDoesNotExist() { final String flowId = flow.getFlowMetadata().getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); final ConnectableComponent newDest = ComponentUtils.createConnectableComponent(group2Processor1); final VersionedConnection requestConnection = new VersionedConnection(); requestConnection.setIdentifier("DOES-NOT-EXIST"); requestConnection.setSource(connectionG2P2toG2P3.getSource()); requestConnection.setDestination(newDest); connectionService.update(flowId, requestConnection); }
@Test(expected = IllegalArgumentException.class) public void testUpdateWhenDestinationGroupDoesNotExist() { final String flowId = flow.getFlowMetadata().getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); final ConnectableComponent newDest = ComponentUtils.createConnectableComponent(group2Processor1); newDest.setGroupId("DOES-NOT-EXIST"); final VersionedConnection requestConnection = new VersionedConnection(); requestConnection.setIdentifier(connectionG2P2toG2P3.getIdentifier()); requestConnection.setSource(connectionG2P2toG2P3.getSource()); requestConnection.setDestination(newDest); connectionService.update(flowId, requestConnection); }
@Test(expected = IllegalArgumentException.class) public void testUpdateWhenDestinationComponentNotInDestinationGroup() { final String flowId = flow.getFlowMetadata().getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); final ConnectableComponent newDest = ComponentUtils.createConnectableComponent(group1Processor1); newDest.setGroupId(group2.getIdentifier()); final VersionedConnection requestConnection = new VersionedConnection(); requestConnection.setIdentifier(connectionG2P2toG2P3.getIdentifier()); requestConnection.setSource(connectionG2P2toG2P3.getSource()); requestConnection.setDestination(newDest); connectionService.update(flowId, requestConnection); }
@Test public void testUpdateDestToInputPortInChildGroup() { final String flowId = flow.getFlowMetadata().getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); final ConnectableComponent newDest = ComponentUtils.createConnectableComponent(group3InputPort); final VersionedConnection requestConnection = new VersionedConnection(); requestConnection.setIdentifier(connectionG2P2toG2P3.getIdentifier()); requestConnection.setDestination(newDest); final FDConnection updatedConnection = connectionService.update(flowId, requestConnection); assertNotNull(updatedConnection); }
@Test(expected = IllegalStateException.class) public void testUpdateDestToProcessorInChildGroup() { final String flowId = flow.getFlowMetadata().getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); final ConnectableComponent newDest = ComponentUtils.createConnectableComponent(group3Processor1); final VersionedConnection requestConnection = new VersionedConnection(); requestConnection.setIdentifier(connectionG2P2toG2P3.getIdentifier()); requestConnection.setDestination(newDest); connectionService.update(flowId, requestConnection); }
@Test(expected = IllegalStateException.class) public void testUpdateDestToInputPortInNonChildGroup() { final String flowId = flow.getFlowMetadata().getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); final ConnectableComponent newDest = ComponentUtils.createConnectableComponent(group1InputPort); final VersionedConnection requestConnection = new VersionedConnection(); requestConnection.setIdentifier(connectionG2P2toG2P3.getIdentifier()); requestConnection.setDestination(newDest); connectionService.update(flowId, requestConnection); }
@Test(expected = IllegalArgumentException.class) public void testUpdateWithInvalidBackPressureDataSize() { final String flowId = flow.getFlowMetadata().getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); final VersionedConnection requestConnection = new VersionedConnection(); requestConnection.setIdentifier(connectionG2P2toG2P3.getIdentifier()); requestConnection.setBackPressureDataSizeThreshold("INVALID"); connectionService.update(flowId, requestConnection); }
@Test(expected = IllegalArgumentException.class) public void testUpdateWithInvalidFlowFileExpiration() { final String flowId = flow.getFlowMetadata().getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); final VersionedConnection requestConnection = new VersionedConnection(); requestConnection.setIdentifier(connectionG2P2toG2P3.getIdentifier()); requestConnection.setFlowFileExpiration("INVALID"); connectionService.update(flowId, requestConnection); }
@Test(expected = IllegalArgumentException.class) public void testUpdateWithEmptySelectedRelationships() { final String flowId = flow.getFlowMetadata().getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); final VersionedConnection requestConnection = new VersionedConnection(); requestConnection.setIdentifier(connectionG2P2toG2P3.getIdentifier()); requestConnection.setSelectedRelationships(Collections.emptySet()); connectionService.update(flowId, requestConnection); }