class 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);
        }

        // Ensure that processors specify the selected relationships
        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();

        // VALIDATE SOURCE

        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");
            }

            // Locate the remote process group with the group id specified in the source, or throw exception
            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"));

            // Find the output port in the remote process group with the target id of the specified source
            final Optional<VersionedRemoteGroupPort> remoteGroupPort = sourceRemoteProcessGroup.getOutputPorts().stream()
                    .filter(p -> p.getTargetId().equals(sourceId)).findFirst();

            // If we couldn't find a remote output port with that id then create a new one, but don't add it to the RPG until later
            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);
            }

            // set the remote input port name to the id
            sourceConnectable.setName(sourceId);
        } else {
            // Validate source group exists and that the source connectable exists in the source group
            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 the source is not located in the same group as the connection, then ensure the source group is a child group
            // of the connection group, and ensure the source is an output port
            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");
                }
            }
        }


        // VALIDATE DESTINATION

        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");
            }

            // Locate the remote process group with the group id specified in the dest, or throw exception
            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"));

            // Find the input port in the remote process group with the target id of the specified dest
            final Optional<VersionedRemoteGroupPort> remoteGroupPort = destRemoteProcessGroup.getInputPorts().stream()
                    .filter(p -> p.getTargetId().equals(destId)).findFirst();

            // If we couldn't find a remote input port with that target id then create a new one, but don't add it to the RPG until later
            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);
            }

            // set the remote output port name to the id
            destConnectable.setName(destId);
        } else {
            // Validate dest group exists and that the dest connectable exists in the source group
            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 the dest is not located in the same group as the connection, then ensure the dest group is a child group
            // of the connection group, and ensure the dest is an input port
            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");
                }
            }
        }

        // Now that both source and dest passed validation we can add them to the remote process group
        if (newRemoteOutputPort != null) {
            sourceRemoteProcessGroup.getOutputPorts().add(newRemoteOutputPort);
        }
        if (newRemoteInputPort != null) {
            destRemoteProcessGroup.getInputPorts().add(newRemoteInputPort);
        }

        // Continue on to normal create...

        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;
    }

    public  StandardFDConnectionService(final FDFlowManager flowManager, final ValidationService validationService);

}