StandardFDConnectionService extends BaseComponentService<VersionedConnection, FDConnection> implements com.cloudera.cem.efm.service.component.FDConnectionService { @Override public FDConnection create(final String flowId, final String processGroupId, final VersionedConnection requestComponentConfig) { final ConnectableComponent sourceConnectable = requestComponentConfig.getSource(); final ConnectableComponent destConnectable = requestComponentConfig.getDestination(); if (sourceConnectable == null || destConnectable == null) { throw new IllegalArgumentException("Both source and destination must be specified"); } if (sourceConnectable.getId() == null || destConnectable.getId() == null) { throw new IllegalArgumentException("Both source and destination ids must be specified"); } if (sourceConnectable.getType() == null || destConnectable.getType() == null) { throw new IllegalArgumentException("Both source and destination types must be specified"); } if (sourceConnectable.getType() == ConnectableComponentType.FUNNEL && destConnectable.getType() == ConnectableComponentType.FUNNEL && sourceConnectable.getId().equals(destConnectable.getId())) { throw new IllegalArgumentException("A funnel cannot be connected to itself"); } if (sourceConnectable.getGroupId() == null) { sourceConnectable.setGroupId(processGroupId); } if (destConnectable.getGroupId() == null) { destConnectable.setGroupId(processGroupId); } if (sourceConnectable.getType() == ConnectableComponentType.PROCESSOR) { if (requestComponentConfig.getSelectedRelationships() == null || requestComponentConfig.getSelectedRelationships().isEmpty()) { throw new IllegalArgumentException("Selected relationships must be specified"); } } validateRequestConfig(requestComponentConfig); final FDFlow currentFlow = getFlowOrNotFound(flowId); final VersionedProcessGroup flowContent = currentFlow.getFlowContent(); final VersionedProcessGroup connectionGroup = ComponentUtils.getProcessGroupOrNotFound(processGroupId, flowContent); final String sourceId = sourceConnectable.getId(); final String sourceGroupId = sourceConnectable.getGroupId(); final String destId = destConnectable.getId(); final String destGroupId = destConnectable.getGroupId(); VersionedRemoteProcessGroup sourceRemoteProcessGroup = null; VersionedRemoteGroupPort newRemoteOutputPort = null; if (sourceConnectable.getType() == ConnectableComponentType.REMOTE_OUTPUT_PORT) { try { UUID.fromString(sourceId); } catch (Exception e) { throw new IllegalArgumentException("The ID of a remote output port must be a UUID"); } sourceRemoteProcessGroup = connectionGroup.getRemoteProcessGroups().stream() .filter(rpg -> rpg.getIdentifier().equals(sourceGroupId)) .findFirst() .orElseThrow(() -> new IllegalArgumentException("Unable to find the specified remote process group for connection source")); final Optional<VersionedRemoteGroupPort> remoteGroupPort = sourceRemoteProcessGroup.getOutputPorts().stream() .filter(p -> p.getTargetId().equals(sourceId)).findFirst(); if (!remoteGroupPort.isPresent()) { newRemoteOutputPort = new VersionedRemoteGroupPort(); newRemoteOutputPort.setIdentifier(UUID.randomUUID().toString()); newRemoteOutputPort.setGroupIdentifier(connectionGroup.getIdentifier()); newRemoteOutputPort.setRemoteGroupId(sourceRemoteProcessGroup.getIdentifier()); newRemoteOutputPort.setTargetId(sourceId); newRemoteOutputPort.setName(sourceId); newRemoteOutputPort.setComponentType(ComponentType.REMOTE_OUTPUT_PORT); } sourceConnectable.setName(sourceId); } else { final VersionedProcessGroup sourceGroup = ComponentUtils.getProcessGroupOrIllegalArgument(sourceGroupId, flowContent); final boolean sourceGroupContainsSource = containsConnectable(sourceGroup, sourceConnectable); if (!sourceGroupContainsSource) { throw new IllegalStateException("Cannot add connection because the source component was not found in the source group"); } if (!processGroupId.equals(sourceGroupId)) { if (sourceConnectable.getType() != ConnectableComponentType.OUTPUT_PORT) { throw new IllegalStateException("Cannot add connection between " + sourceId + " and " + destId + " because they are in different process groups and the source is not an output port"); } boolean connectionGroupContainsSourceGroup = containsChildGroup(connectionGroup, sourceGroupId); if (!connectionGroupContainsSourceGroup) { throw new IllegalStateException("Cannot add connection between " + sourceId + " and " + destId + " because the source group is not a child group of the connection group"); } } } VersionedRemoteProcessGroup destRemoteProcessGroup = null; VersionedRemoteGroupPort newRemoteInputPort = null; if (destConnectable.getType() == ConnectableComponentType.REMOTE_INPUT_PORT) { try { UUID.fromString(destId); } catch (Exception e) { throw new IllegalArgumentException("The ID of a remote input port must be a UUID"); } destRemoteProcessGroup = connectionGroup.getRemoteProcessGroups().stream() .filter(rpg -> rpg.getIdentifier().equals(destGroupId)) .findFirst() .orElseThrow(() -> new IllegalArgumentException("Unable to find the specified remote process group for connection destination")); final Optional<VersionedRemoteGroupPort> remoteGroupPort = destRemoteProcessGroup.getInputPorts().stream() .filter(p -> p.getTargetId().equals(destId)).findFirst(); if (!remoteGroupPort.isPresent()) { newRemoteInputPort = new VersionedRemoteGroupPort(); newRemoteInputPort.setIdentifier(UUID.randomUUID().toString()); newRemoteInputPort.setGroupIdentifier(connectionGroup.getIdentifier()); newRemoteInputPort.setRemoteGroupId(destRemoteProcessGroup.getIdentifier()); newRemoteInputPort.setTargetId(destId); newRemoteInputPort.setName(destId); newRemoteInputPort.setComponentType(ComponentType.REMOTE_INPUT_PORT); } destConnectable.setName(destId); } else { final VersionedProcessGroup destGroup = ComponentUtils.getProcessGroupOrIllegalArgument(destGroupId, flowContent); final boolean destGroupContainsSource = containsConnectable(destGroup, destConnectable); if (!destGroupContainsSource) { throw new IllegalStateException("Cannot add connection because the destination component was not found in the destination group"); } if (!processGroupId.equals(destGroupId)) { if (destConnectable.getType() != ConnectableComponentType.INPUT_PORT) { throw new IllegalStateException("Cannot add connection between " + sourceId + " and " + destId + " because they are in different process groups and the destination is not an input port"); } boolean connectionGroupContainsDestGroup = containsChildGroup(connectionGroup, destGroupId); if (!connectionGroupContainsDestGroup) { throw new IllegalStateException("Cannot add connection between " + sourceId + " and " + destId + " because the destination group is not a child group of the connection group"); } } } if (newRemoteOutputPort != null) { sourceRemoteProcessGroup.getOutputPorts().add(newRemoteOutputPort); } if (newRemoteInputPort != null) { destRemoteProcessGroup.getInputPorts().add(newRemoteInputPort); } final FDConnection created = createComponent( currentFlow, connectionGroup, null, requestComponentConfig, () -> { final VersionedConnection connectionConfig = new VersionedConnection(); connectionConfig.setSource(sourceConnectable); connectionConfig.setDestination(destConnectable); connectionConfig.setSelectedRelationships(new HashSet<>()); connectionConfig.setName(""); connectionConfig.setBends(new ArrayList<>()); connectionConfig.setPrioritizers(new ArrayList<>()); connectionConfig.setzIndex(new Long(0)); connectionConfig.setLabelIndex(new Integer(1)); connectionConfig.setBackPressureDataSizeThreshold(DEFAULT_BACK_PRESSURE_DATA_SIZE_THRESHOLD); connectionConfig.setBackPressureObjectThreshold(DEFAULT_BACK_PRESSURE_OBJECT_SIZE_THRESHOLD); connectionConfig.setFlowFileExpiration(DEFAULT_FLOW_FILE_EXPIRATION); final FDConnection connection = new FDConnection(); connection.setComponentConfiguration(connectionConfig); return connection; }, VersionedProcessGroup::getConnections, null); getValidationService().clearValidationErrors(sourceConnectable.getId()); getValidationService().clearValidationErrors(destConnectable.getId()); return created; } 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 testCreateConnectionSameGroup() { final String flowId = flow.getFlowMetadata().getIdentifier(); final String groupId = group2.getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); final FDConnection createdConnection = connectionService.create(flowId, groupId, connectionConfig); assertNotNull(createdConnection); assertNotNull(createdConnection.getComponentConfiguration()); assertEquals(connectionConfig.getIdentifier(), createdConnection.getComponentConfiguration().getIdentifier()); }
@Test(expected = IllegalArgumentException.class) public void testCreateConnectionInvalidBackPressureDataSize() { final String flowId = flow.getFlowMetadata().getIdentifier(); final String groupId = group2.getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); connectionConfig.setBackPressureDataSizeThreshold("INVALID"); connectionService.create(flowId, groupId, connectionConfig); }
@Test(expected = IllegalArgumentException.class) public void testCreateConnectionInvalidFlowFileExpiration() { final String flowId = flow.getFlowMetadata().getIdentifier(); final String groupId = group2.getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); connectionConfig.setFlowFileExpiration("INVALID"); connectionService.create(flowId, groupId, connectionConfig); }
@Test(expected = IllegalArgumentException.class) public void testCreateConnectionNoSourceRelationships() { final String flowId = flow.getFlowMetadata().getIdentifier(); final String groupId = group2.getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); connectionConfig.setSelectedRelationships(Collections.emptySet()); connectionService.create(flowId, groupId, connectionConfig); }
@Test(expected = ResourceNotFoundException.class) public void testCreateConnectionSameGroupFlowNotFound() { final String flowId = flow.getFlowMetadata().getIdentifier(); final String groupId = group2.getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.empty()); connectionService.create(flowId, groupId, connectionConfig); }
@Test(expected = ResourceNotFoundException.class) public void testCreateConnectionSameGroupConnectionGroupNotFound() { final String flowId = flow.getFlowMetadata().getIdentifier(); final String groupId = "DOES-NOT-EXIST"; when(flowManager.getFlow(flowId)).thenReturn(Optional.empty()); connectionService.create(flowId, groupId, connectionConfig); }
@Test(expected = IllegalArgumentException.class) public void testCreateConnectionSourceGroupNotFound() { final String flowId = flow.getFlowMetadata().getIdentifier(); final String groupId = group2.getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); connectionConfig.getSource().setGroupId("DOES-NOT-EXIST"); connectionService.create(flowId, groupId, connectionConfig); }
@Test(expected = IllegalArgumentException.class) public void testCreateConnectionDestGroupNotFound() { final String flowId = flow.getFlowMetadata().getIdentifier(); final String groupId = group2.getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); connectionConfig.getDestination().setGroupId("DOES-NOT-EXIST"); connectionService.create(flowId, groupId, connectionConfig); }
@Test(expected = IllegalStateException.class) public void testCreateConnectionWhenSourceGroupNotChildOfConnectionGroup() { final String flowId = flow.getFlowMetadata().getIdentifier(); final String groupId = group2.getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); final ConnectableComponent newSource = ComponentUtils.createConnectableComponent(group1OutputPort); connectionConfig.setSource(newSource); connectionService.create(flowId, groupId, connectionConfig); }
@Test(expected = IllegalStateException.class) public void testCreateConnectionWhenSourceGroupIsChildButSourceNotOutputPort() { final String flowId = flow.getFlowMetadata().getIdentifier(); final String groupId = group2.getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); final ConnectableComponent newSource = ComponentUtils.createConnectableComponent(group3Processor1); connectionConfig.setSource(newSource); connectionService.create(flowId, groupId, connectionConfig); }
@Test public void testCreateConnectionFromOutputPortInChildGroup() { final String flowId = flow.getFlowMetadata().getIdentifier(); final String groupId = group2.getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); final ConnectableComponent newSource = ComponentUtils.createConnectableComponent(group3OutputPort); connectionConfig.setSource(newSource); connectionService.create(flowId, groupId, connectionConfig); }
@Test(expected = IllegalStateException.class) public void testCreateConnectionWhenDestGroupNotChildOfConnectionGroup() { final String flowId = flow.getFlowMetadata().getIdentifier(); final String groupId = group2.getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); final ConnectableComponent newDest = ComponentUtils.createConnectableComponent(group1InputPort); connectionConfig.setDestination(newDest); connectionService.create(flowId, groupId, connectionConfig); }
@Test(expected = IllegalStateException.class) public void testCreateConnectionWhenDestGroupIsChildButDestNotInputPort() { final String flowId = flow.getFlowMetadata().getIdentifier(); final String groupId = group2.getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); final ConnectableComponent newDest = ComponentUtils.createConnectableComponent(group3OutputPort); connectionConfig.setDestination(newDest); connectionService.create(flowId, groupId, connectionConfig); }
@Test public void testCreateConnectionToInputPortInChildGroup() { final String flowId = flow.getFlowMetadata().getIdentifier(); final String groupId = group2.getIdentifier(); when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow)); final ConnectableComponent newDest = ComponentUtils.createConnectableComponent(group3InputPort); connectionConfig.setDestination(newDest); connectionService.create(flowId, groupId, connectionConfig); }