test_class.declarator: connectionConfig
test_class.modifier: private
test_class.original_string: private VersionedConnection connectionConfig;
test_class.type: VersionedConnection
test_class.var_name: connectionConfig
test_class.declarator: connectionService
test_class.modifier: private
test_class.original_string: private FDConnectionService connectionService;
test_class.type: FDConnectionService
test_class.var_name: connectionService
test_class.file: efm/efm-flow-designer/efm-flow-designer-framework/src/test/java/com/cloudera/cem/efm/service/component/TestStandardFDConnectionService.java
test_class.identifier: TestStandardFDConnectionService
test_class.superclass: extends BaseFDServiceTest
test_case.body: @Test
      public void testCreateConnectionToInputPortInChildGroup() {
          final String flowId = flow.getFlowMetadata().getIdentifier();
          final String groupId = group2.getIdentifier();
  
          when(flowManager.getFlow(flowId)).thenReturn(Optional.of(flow));
  
          // Make the dest the input port of group 3 which is in the same group as the source
          final ConnectableComponent newDest = ComponentUtils.createConnectableComponent(group3InputPort);
          connectionConfig.setDestination(newDest);
  
          connectionService.create(flowId, groupId, connectionConfig);
      }
test_case.class_method_signature: TestStandardFDConnectionService.testCreateConnectionToInputPortInChildGroup()
test_case.full_signature: @Test public void testCreateConnectionToInputPortInChildGroup()
test_case.identifier: testCreateConnectionToInputPortInChildGroup
test_case.invocations: getIdentifier getFlowMetadata getIdentifier thenReturn when getFlow of createConnectableComponent setDestination create
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testCreateConnectionToInputPortInChildGroup()
focal_class.declarator: DEFAULT_BACK_PRESSURE_DATA_SIZE_THRESHOLD = "10000 B"
focal_class.modifier: public static final
focal_class.original_string: public static final String DEFAULT_BACK_PRESSURE_DATA_SIZE_THRESHOLD = "10000 B";
focal_class.type: String
focal_class.var_name: DEFAULT_BACK_PRESSURE_DATA_SIZE_THRESHOLD
focal_class.declarator: DEFAULT_BACK_PRESSURE_OBJECT_SIZE_THRESHOLD = 0
focal_class.modifier: public static final
focal_class.original_string: public static final long DEFAULT_BACK_PRESSURE_OBJECT_SIZE_THRESHOLD = 0;
focal_class.type: long
focal_class.var_name: DEFAULT_BACK_PRESSURE_OBJECT_SIZE_THRESHOLD
focal_class.declarator: DEFAULT_FLOW_FILE_EXPIRATION = "60 seconds"
focal_class.modifier: public static final
focal_class.original_string: public static final String DEFAULT_FLOW_FILE_EXPIRATION = "60 seconds";
focal_class.type: String
focal_class.var_name: DEFAULT_FLOW_FILE_EXPIRATION
focal_class.declarator: componentTypeFunctionMap
focal_class.modifier: private static
focal_class.original_string: private static Map<ConnectableComponentType, Function<VersionedProcessGroup, Set<? extends VersionedComponent>>> componentTypeFunctionMap;
focal_class.type: Map<ConnectableComponentType, Function<VersionedProcessGroup, Set<? extends VersionedComponent>>>
focal_class.var_name: componentTypeFunctionMap
focal_class.file: efm/efm-flow-designer/efm-flow-designer-framework/src/main/java/com/cloudera/cem/efm/service/component/StandardFDConnectionService.java
focal_class.identifier: StandardFDConnectionService
focal_class.interfaces: implements com.cloudera.cem.efm.service.component.FDConnectionService
focal_class.class_method_signature: StandardFDConnectionService.StandardFDConnectionService(final FDFlowManager flowManager, final ValidationService validationService)
focal_class.full_signature: public  StandardFDConnectionService(final FDFlowManager flowManager, final ValidationService validationService)
focal_class.identifier: StandardFDConnectionService
focal_class.modifiers: public
focal_class.parameters: (final FDFlowManager flowManager, final ValidationService validationService)
focal_class.signature:  StandardFDConnectionService(final FDFlowManager flowManager, final ValidationService validationService)
focal_class.class_method_signature: StandardFDConnectionService.create(final String flowId, final String processGroupId, final VersionedConnection requestComponentConfig)
focal_class.full_signature: @Override public FDConnection create(final String flowId, final String processGroupId, final VersionedConnection requestComponentConfig)
focal_class.identifier: create
focal_class.modifiers: @Override public
focal_class.parameters: (final String flowId, final String processGroupId, final VersionedConnection requestComponentConfig)
focal_class.return: FDConnection
focal_class.signature: FDConnection create(final String flowId, final String processGroupId, final VersionedConnection requestComponentConfig)
focal_class.class_method_signature: StandardFDConnectionService.get(final String flowId, final String componentId)
focal_class.full_signature: @Override public FDConnection get(final String flowId, final String componentId)
focal_class.identifier: get
focal_class.modifiers: @Override public
focal_class.parameters: (final String flowId, final String componentId)
focal_class.return: FDConnection
focal_class.signature: FDConnection get(final String flowId, final String componentId)
focal_class.class_method_signature: StandardFDConnectionService.update(final String flowId, final VersionedConnection requestComponentConfig)
focal_class.full_signature: @Override public FDConnection update(final String flowId, final VersionedConnection requestComponentConfig)
focal_class.identifier: update
focal_class.modifiers: @Override public
focal_class.parameters: (final String flowId, final VersionedConnection requestComponentConfig)
focal_class.return: FDConnection
focal_class.signature: FDConnection update(final String flowId, final VersionedConnection requestComponentConfig)
focal_class.class_method_signature: StandardFDConnectionService.delete(final String flowId, final String componentId)
focal_class.full_signature: @Override public void delete(final String flowId, final String componentId)
focal_class.identifier: delete
focal_class.modifiers: @Override public
focal_class.parameters: (final String flowId, final String componentId)
focal_class.return: void
focal_class.signature: void delete(final String flowId, final String componentId)
focal_class.class_method_signature: StandardFDConnectionService.instantiateComponent(final VersionedConnection componentConfig)
focal_class.full_signature: @Override protected FDConnection instantiateComponent(final VersionedConnection componentConfig)
focal_class.identifier: instantiateComponent
focal_class.modifiers: @Override protected
focal_class.parameters: (final VersionedConnection componentConfig)
focal_class.return: FDConnection
focal_class.signature: FDConnection instantiateComponent(final VersionedConnection componentConfig)
focal_class.class_method_signature: StandardFDConnectionService.configureComponentSpecifics(final VersionedConnection requestComponent,
                                                 final VersionedConnection resultComponent)
focal_class.full_signature: @Override protected void configureComponentSpecifics(final VersionedConnection requestComponent,
                                                 final VersionedConnection resultComponent)
focal_class.identifier: configureComponentSpecifics
focal_class.modifiers: @Override protected
focal_class.parameters: (final VersionedConnection requestComponent,
                                                 final VersionedConnection resultComponent)
focal_class.return: void
focal_class.signature: void configureComponentSpecifics(final VersionedConnection requestComponent,
                                                 final VersionedConnection resultComponent)
focal_class.class_method_signature: StandardFDConnectionService.containsChildGroup(final VersionedProcessGroup parentGroup, final String childGroupId)
focal_class.full_signature: private boolean containsChildGroup(final VersionedProcessGroup parentGroup, final String childGroupId)
focal_class.identifier: containsChildGroup
focal_class.modifiers: private
focal_class.parameters: (final VersionedProcessGroup parentGroup, final String childGroupId)
focal_class.return: boolean
focal_class.signature: boolean containsChildGroup(final VersionedProcessGroup parentGroup, final String childGroupId)
focal_class.class_method_signature: StandardFDConnectionService.containsConnectable(final VersionedProcessGroup sourceGroup, final ConnectableComponent source)
focal_class.full_signature: private boolean containsConnectable(final VersionedProcessGroup sourceGroup, final ConnectableComponent source)
focal_class.identifier: containsConnectable
focal_class.modifiers: private
focal_class.parameters: (final VersionedProcessGroup sourceGroup, final ConnectableComponent source)
focal_class.return: boolean
focal_class.signature: boolean containsConnectable(final VersionedProcessGroup sourceGroup, final ConnectableComponent source)
focal_class.class_method_signature: StandardFDConnectionService.getVersionedComponentConnectable(
              final VersionedProcessGroup sourceGroup, final ConnectableComponent source)
focal_class.full_signature: private Optional<VersionedComponent> getVersionedComponentConnectable(
              final VersionedProcessGroup sourceGroup, final ConnectableComponent source)
focal_class.identifier: getVersionedComponentConnectable
focal_class.modifiers: private
focal_class.parameters: (
              final VersionedProcessGroup sourceGroup, final ConnectableComponent source)
focal_class.return: Optional<VersionedComponent>
focal_class.signature: Optional<VersionedComponent> getVersionedComponentConnectable(
              final VersionedProcessGroup sourceGroup, final ConnectableComponent source)
focal_class.class_method_signature: StandardFDConnectionService.isDestStillUsed(final VersionedConnection existingConnection,
                                      final ConnectableComponent existingDestination,
                                      final VersionedProcessGroup connectionGroup)
focal_class.full_signature: private boolean isDestStillUsed(final VersionedConnection existingConnection,
                                      final ConnectableComponent existingDestination,
                                      final VersionedProcessGroup connectionGroup)
focal_class.identifier: isDestStillUsed
focal_class.modifiers: private
focal_class.parameters: (final VersionedConnection existingConnection,
                                      final ConnectableComponent existingDestination,
                                      final VersionedProcessGroup connectionGroup)
focal_class.return: boolean
focal_class.signature: boolean isDestStillUsed(final VersionedConnection existingConnection,
                                      final ConnectableComponent existingDestination,
                                      final VersionedProcessGroup connectionGroup)
focal_class.class_method_signature: StandardFDConnectionService.isSourceStillUsed(final VersionedConnection existingConnection,
                                      final ConnectableComponent existingSource,
                                      final VersionedProcessGroup connectionGroup)
focal_class.full_signature: private boolean isSourceStillUsed(final VersionedConnection existingConnection,
                                      final ConnectableComponent existingSource,
                                      final VersionedProcessGroup connectionGroup)
focal_class.identifier: isSourceStillUsed
focal_class.modifiers: private
focal_class.parameters: (final VersionedConnection existingConnection,
                                      final ConnectableComponent existingSource,
                                      final VersionedProcessGroup connectionGroup)
focal_class.return: boolean
focal_class.signature: boolean isSourceStillUsed(final VersionedConnection existingConnection,
                                      final ConnectableComponent existingSource,
                                      final VersionedProcessGroup connectionGroup)
focal_class.class_method_signature: StandardFDConnectionService.removeRemoteInputPort(final VersionedProcessGroup connectionGroup, final ConnectableComponent remoteInputPort)
focal_class.full_signature: private void removeRemoteInputPort(final VersionedProcessGroup connectionGroup, final ConnectableComponent remoteInputPort)
focal_class.identifier: removeRemoteInputPort
focal_class.modifiers: private
focal_class.parameters: (final VersionedProcessGroup connectionGroup, final ConnectableComponent remoteInputPort)
focal_class.return: void
focal_class.signature: void removeRemoteInputPort(final VersionedProcessGroup connectionGroup, final ConnectableComponent remoteInputPort)
focal_class.class_method_signature: StandardFDConnectionService.removeRemoteOutputPort(VersionedProcessGroup connectionGroup, ConnectableComponent remoteOutputPort)
focal_class.full_signature: private void removeRemoteOutputPort(VersionedProcessGroup connectionGroup, ConnectableComponent remoteOutputPort)
focal_class.identifier: removeRemoteOutputPort
focal_class.modifiers: private
focal_class.parameters: (VersionedProcessGroup connectionGroup, ConnectableComponent remoteOutputPort)
focal_class.return: void
focal_class.signature: void removeRemoteOutputPort(VersionedProcessGroup connectionGroup, ConnectableComponent remoteOutputPort)
focal_class.class_method_signature: StandardFDConnectionService.validateRequestConfig(final VersionedConnection requestComponentConfig)
focal_class.full_signature: private void validateRequestConfig(final VersionedConnection requestComponentConfig)
focal_class.identifier: validateRequestConfig
focal_class.modifiers: private
focal_class.parameters: (final VersionedConnection requestComponentConfig)
focal_class.return: void
focal_class.signature: void validateRequestConfig(final VersionedConnection requestComponentConfig)
focal_class.superclass: extends BaseComponentService<VersionedConnection, FDConnection>
focal_method.body: @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;
      }
focal_method.class_method_signature: StandardFDConnectionService.create(final String flowId, final String processGroupId, final VersionedConnection requestComponentConfig)
focal_method.full_signature: @Override public FDConnection create(final String flowId, final String processGroupId, final VersionedConnection requestComponentConfig)
focal_method.identifier: create
focal_method.invocations: getSource getDestination getId getId getType getType getType getType equals getId getId getGroupId setGroupId getGroupId setGroupId getType getSelectedRelationships isEmpty getSelectedRelationships validateRequestConfig getFlowOrNotFound getFlowContent getProcessGroupOrNotFound getId getGroupId getId getGroupId getType fromString orElseThrow findFirst filter stream getRemoteProcessGroups equals getIdentifier findFirst filter stream getOutputPorts equals getTargetId isPresent setIdentifier toString randomUUID setGroupIdentifier getIdentifier setRemoteGroupId getIdentifier setTargetId setName setComponentType setName getProcessGroupOrIllegalArgument containsConnectable equals getType containsChildGroup getType fromString orElseThrow findFirst filter stream getRemoteProcessGroups equals getIdentifier findFirst filter stream getInputPorts equals getTargetId isPresent setIdentifier toString randomUUID setGroupIdentifier getIdentifier setRemoteGroupId getIdentifier setTargetId setName setComponentType setName getProcessGroupOrIllegalArgument containsConnectable equals getType containsChildGroup add getOutputPorts add getInputPorts createComponent setSource setDestination setSelectedRelationships setName setBends setPrioritizers setzIndex setLabelIndex setBackPressureDataSizeThreshold setBackPressureObjectThreshold setFlowFileExpiration setComponentConfiguration clearValidationErrors getValidationService getId clearValidationErrors getValidationService getId
focal_method.modifiers: @Override public
focal_method.parameters: (final String flowId, final String processGroupId, final VersionedConnection requestComponentConfig)
focal_method.return: FDConnection
focal_method.signature: FDConnection create(final String flowId, final String processGroupId, final VersionedConnection requestComponentConfig)