private void startComponents(final String processGroupId, final Map<String, Revision> componentRevisions, final Map<String, AffectedComponentEntity> affectedComponents, final Pause pause,
                             final InvalidComponentAction invalidComponentAction) throws LifecycleManagementException {

    if (componentRevisions.isEmpty()) {
        return;
    }

    logger.debug("Starting components with ID's {} from Process Group {}", componentRevisions.keySet(), processGroupId);

    // Wait for all affected processors to be either VALID or INVALID
    waitForProcessorValidation(processGroupId, affectedComponents, pause);

    serviceFacade.verifyScheduleComponents(processGroupId, ScheduledState.RUNNING, componentRevisions.keySet());
    serviceFacade.scheduleComponents(processGroupId, ScheduledState.RUNNING, componentRevisions);

    // wait for all of the Processors to reach the desired state. We don't have to wait for other components because
    // Local and Remote Ports as well as funnels start immediately.
    waitForProcessorState(processGroupId, affectedComponents, ScheduledState.RUNNING, pause, invalidComponentAction);
}
 

@Override
public Set<ConfiguredComponent> updateControllerServiceReferencingComponents(
        final String controllerServiceId, final ScheduledState scheduledState, final ControllerServiceState controllerServiceState) {
    // get the controller service
    final ControllerServiceNode controllerService = locateControllerService(controllerServiceId);

    // this request is either acting upon referencing services or schedulable components
    if (controllerServiceState != null) {
        if (ControllerServiceState.ENABLED.equals(controllerServiceState)) {
            return serviceProvider.enableReferencingServices(controllerService);
        } else {
            return serviceProvider.disableReferencingServices(controllerService);
        }
    } else if (scheduledState != null) {
        if (ScheduledState.RUNNING.equals(scheduledState)) {
            return serviceProvider.scheduleReferencingComponents(controllerService);
        } else {
            return serviceProvider.unscheduleReferencingComponents(controllerService);
        }
    }

    return Collections.emptySet();
}
 

@Override
public void verifyUpdateReferencingComponents(final String controllerServiceId, final ScheduledState scheduledState, final ControllerServiceState controllerServiceState) {
    final ControllerServiceNode controllerService = locateControllerService(controllerServiceId);

    if (controllerServiceState != null) {
        if (ControllerServiceState.ENABLED.equals(controllerServiceState)) {
            serviceProvider.verifyCanEnableReferencingServices(controllerService);
        } else {
            serviceProvider.verifyCanDisableReferencingServices(controllerService);
        }
    } else if (scheduledState != null) {
        if (ScheduledState.RUNNING.equals(scheduledState)) {
            serviceProvider.verifyCanScheduleReferencingComponents(controllerService);
        } else {
            serviceProvider.verifyCanStopReferencingComponents(controllerService);
        }
    }
}
 

@Override
protected void handleStateTransition(final Port port, final ScheduledState proposedScheduledState) throws IllegalStateException {
    final ProcessGroup processGroup = port.getProcessGroup();
    switch (proposedScheduledState) {
        case RUNNING:
            processGroup.startOutputPort(port);
            break;
        case STOPPED:
            switch (port.getScheduledState()) {
                case RUNNING:
                    processGroup.stopOutputPort(port);
                    break;
                case DISABLED:
                    processGroup.enableOutputPort(port);
                    break;
            }
            break;
        case DISABLED:
            processGroup.disableOutputPort(port);
            break;
    }
}
 

@Override
public void verifyScheduleComponents(final String groupId, final ScheduledState state,final Set<String> componentIds) {
    final ProcessGroup group = locateProcessGroup(flowController, groupId);

    final Set<Connectable> connectables = new HashSet<>(componentIds.size());
    for (final String componentId : componentIds) {
        final Connectable connectable = group.findLocalConnectable(componentId);
        if (connectable == null) {
            throw new ResourceNotFoundException("Unable to find component with id " + componentId);
        }

        connectables.add(connectable);
    }

    // verify as appropriate
    connectables.forEach(connectable -> {
        if (ScheduledState.RUNNING.equals(state)) {
            group.verifyCanStart(connectable);
        } else {
            group.verifyCanStop(connectable);
        }
    });
}
 

@Override
public void scheduleComponents(final String groupId, final ScheduledState state, final Set<String> componentIds) {
    final ProcessGroup group = locateProcessGroup(flowController, groupId);

    for (final String componentId : componentIds) {
        final Connectable connectable = group.findLocalConnectable(componentId);
        if (ScheduledState.RUNNING.equals(state)) {
            if (ConnectableType.PROCESSOR.equals(connectable.getConnectableType())) {
                connectable.getProcessGroup().startProcessor((ProcessorNode) connectable);
            } else if (ConnectableType.INPUT_PORT.equals(connectable.getConnectableType())) {
                connectable.getProcessGroup().startInputPort((Port) connectable);
            } else if (ConnectableType.OUTPUT_PORT.equals(connectable.getConnectableType())) {
                connectable.getProcessGroup().startOutputPort((Port) connectable);
            }
        } else {
            if (ConnectableType.PROCESSOR.equals(connectable.getConnectableType())) {
                connectable.getProcessGroup().stopProcessor((ProcessorNode) connectable);
            } else if (ConnectableType.INPUT_PORT.equals(connectable.getConnectableType())) {
                connectable.getProcessGroup().stopInputPort((Port) connectable);
            } else if (ConnectableType.OUTPUT_PORT.equals(connectable.getConnectableType())) {
                connectable.getProcessGroup().stopOutputPort((Port) connectable);
            }
        }
    }
}
 

@Override
public void startProcessor(final ProcessorNode processor) {
    readLock.lock();
    try {
        if (getProcessor(processor.getIdentifier()) == null) {
            throw new IllegalStateException("Processor is not a member of this Process Group");
        }

        final ScheduledState state = processor.getScheduledState();
        if (state == ScheduledState.DISABLED) {
            throw new IllegalStateException("Processor is disabled");
        } else if (state == ScheduledState.RUNNING) {
            return;
        }

        scheduler.startProcessor(processor);
    } finally {
        readLock.unlock();
    }
}
 

@Override
public void startInputPort(final Port port) {
    readLock.lock();
    try {
        if (getInputPort(port.getIdentifier()) == null) {
            throw new IllegalStateException("Port " + port.getIdentifier() + " is not a member of this Process Group");
        }

        final ScheduledState state = port.getScheduledState();
        if (state == ScheduledState.DISABLED) {
            throw new IllegalStateException("InputPort " + port.getIdentifier() + " is disabled");
        } else if (state == ScheduledState.RUNNING) {
            return;
        }

        scheduler.startPort(port);
    } finally {
        readLock.unlock();
    }
}
 

/**
 * Validates that the Processor can be stopped when @OnScheduled blocks
 * indefinitely but written to react to thread interrupts
 */
@Test
public void validateProcessorCanBeStoppedWhenOnScheduledBlocksIndefinitelyInterruptable() throws Exception {
    final FlowManagerAndSystemBundle fcsb = this.buildFlowControllerForTest(NiFiProperties.PROCESSOR_SCHEDULING_TIMEOUT, "5 sec");
    flowManager = fcsb.getFlowManager();

    ProcessGroup testGroup = flowManager.createProcessGroup(UUID.randomUUID().toString());
    ProcessorNode testProcNode = flowManager.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString(),
            fcsb.getSystemBundle().getBundleDetails().getCoordinate());
    testProcNode.setProperties(properties);
    TestProcessor testProcessor = (TestProcessor) testProcNode.getProcessor();
    // sets the scenario for the processor to run
    this.blockingInterruptableOnUnschedule(testProcessor);

    testProcNode.performValidation();
    processScheduler.startProcessor(testProcNode, true);
    assertCondition(() -> ScheduledState.RUNNING == testProcNode.getScheduledState(), SHORT_DELAY_TOLERANCE);
    processScheduler.stopProcessor(testProcNode);
    assertCondition(() -> ScheduledState.STOPPED == testProcNode.getScheduledState(), MEDIUM_DELAY_TOLERANCE);
}
 

@Override
public void stopOutputPort(final Port port) {
    readLock.lock();
    try {
        if (!outputPorts.containsKey(port.getIdentifier())) {
            throw new IllegalStateException("No Output Port with ID " + port.getIdentifier() + " belongs to this Process Group");
        }

        final ScheduledState state = port.getScheduledState();
        if (state == ScheduledState.DISABLED) {
            throw new IllegalStateException("OutputPort is disabled");
        } else if (state == ScheduledState.STOPPED) {
            return;
        }

        scheduler.stopPort(port);
    } finally {
        readLock.unlock();
    }
}
 

@Test
public void testSearchBasedOnProperty() {
    // given
    final Map<PropertyDescriptor, String> rawProperties = new HashMap<>();
    final PropertyDescriptor descriptor1 = new PropertyDescriptor.Builder().name("property1").displayName("property1display").description("property1 description").sensitive(false).build();
    final PropertyDescriptor descriptor2 = new PropertyDescriptor.Builder().name("property2").displayName("property2display").description("property2 description").sensitive(true).build();
    rawProperties.put(descriptor1, "property1value");
    rawProperties.put(descriptor2, "property2value");

    final ProcessorNode processorNode = getProcessorNode("processor1", "name1", "", Optional.empty(), SchedulingStrategy.TIMER_DRIVEN,
            ExecutionNode.ALL, ScheduledState.RUNNING, ValidationStatus.VALID, new HashSet<>(), "Processor", Mockito.mock(Processor.class),
            rawProperties, AUTHORIZED);

    givenRootProcessGroup()
            .withProcessor(processorNode);

    // when
    whenExecuteSearch("property");

    // then
    thenResultConsists()
            .ofProcessor(getSimpleResultFromRoot("processor1", "name1", "Property name: property1", "Property value: property1 - property1value",
                    "Property description: property1 description", "Property name: property2", "Property description: property2 description"))
            .validate(results);
}
 

@Override
public void enableInputPort(final Port port) {
    readLock.lock();
    try {
        if (!inputPorts.containsKey(port.getIdentifier())) {
            throw new IllegalStateException("No Input Port with ID " + port.getIdentifier() + " belongs to this Process Group");
        }

        final ScheduledState state = port.getScheduledState();
        if (state == ScheduledState.STOPPED) {
            return;
        } else if (state == ScheduledState.RUNNING) {
            throw new IllegalStateException("InputPort is currently running");
        }

        scheduler.enablePort(port);
    } finally {
        readLock.unlock();
    }
}
 

/**
 * Converts a set of ports into a set of remote process group ports.
 *
 * @param ports to convert
 * @return descriptors of ports
 */
private Set<RemoteProcessGroupPortDescriptor> convertRemotePort(final Set<PortDTO> ports) {
    Set<RemoteProcessGroupPortDescriptor> remotePorts = null;
    if (ports != null) {
        remotePorts = new LinkedHashSet<>(ports.size());
        for (final PortDTO port : ports) {
            final StandardRemoteProcessGroupPortDescriptor descriptor = new StandardRemoteProcessGroupPortDescriptor();
            final ScheduledState scheduledState = ScheduledState.valueOf(port.getState());
            descriptor.setId(generatePortId(port.getId()));
            descriptor.setTargetId(port.getId());
            descriptor.setName(port.getName());
            descriptor.setComments(port.getComments());
            descriptor.setTargetRunning(ScheduledState.RUNNING.equals(scheduledState));
            remotePorts.add(descriptor);
        }
    }
    return remotePorts;
}
 

@Override
public void disableOutputPort(final Port port) {
    readLock.lock();
    try {
        if (!outputPorts.containsKey(port.getIdentifier())) {
            throw new IllegalStateException("No OutputPort with ID " + port.getIdentifier() + " belongs to this Process Group");
        }

        final ScheduledState state = port.getScheduledState();
        if (state == ScheduledState.DISABLED) {
            return;
        } else if (state == ScheduledState.RUNNING) {
            throw new IllegalStateException("OutputPort is currently running");
        }

        scheduler.disablePort(port);
    } finally {
        readLock.unlock();
    }
}
 

@Override
public void disableProcessor(final ProcessorNode processor) {
    readLock.lock();
    try {
        if (!processors.containsKey(processor.getIdentifier())) {
            throw new IllegalStateException("No Processor with ID " + processor.getIdentifier() + " belongs to this Process Group");
        }

        final ScheduledState state = processor.getScheduledState();
        if (state == ScheduledState.DISABLED) {
            return;
        } else if (state == ScheduledState.RUNNING) {
            throw new IllegalStateException("Processor is currently running");
        }

        scheduler.disableProcessor(processor);
    } finally {
        readLock.unlock();
    }
}
 

/**
 * Validates that the Processor can be stopped when @OnScheduled blocks
 * indefinitely and written to ignore thread interrupts
 */
@Test
public void validateProcessorCanBeStoppedWhenOnScheduledBlocksIndefinitelyUninterruptable() throws Exception {
    final FlowManagerAndSystemBundle fcsb = this.buildFlowControllerForTest(NiFiProperties.PROCESSOR_SCHEDULING_TIMEOUT, "1 sec");
    flowManager = fcsb.getFlowManager();

    ProcessGroup testGroup = flowManager.createProcessGroup(UUID.randomUUID().toString());
    ProcessorNode testProcNode = flowManager.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString(),
            fcsb.getSystemBundle().getBundleDetails().getCoordinate());
    testProcNode.setProperties(properties);
    TestProcessor testProcessor = (TestProcessor) testProcNode.getProcessor();
    // sets the scenario for the processor to run
    this.blockingUninterruptableOnUnschedule(testProcessor);

    testProcNode.performValidation();
    processScheduler.startProcessor(testProcNode, true);
    assertCondition(() -> ScheduledState.RUNNING == testProcNode.getScheduledState(), MEDIUM_DELAY_TOLERANCE);
    processScheduler.stopProcessor(testProcNode);
    assertCondition(() -> ScheduledState.STOPPED == testProcNode.getScheduledState(), MEDIUM_DELAY_TOLERANCE);
}
 

/**
 * Audits the update of process group configuration.
 *
 * @param proceedingJoinPoint join point
 * @param groupId group id
 * @param state scheduled state
 * @throws Throwable ex
 */
@Around("within(org.apache.nifi.web.dao.ProcessGroupDAO+) && "
        + "execution(void enableComponents(String, org.apache.nifi.controller.ScheduledState, java.util.Set<String>)) && "
        + "args(groupId, state, componentIds)")
public void enableComponentsAdvice(ProceedingJoinPoint proceedingJoinPoint, String groupId, ScheduledState state, Set<String> componentIds) throws Throwable {
    final Operation operation;

    proceedingJoinPoint.proceed();

    // determine the running state
    if (ScheduledState.DISABLED.equals(state)) {
        operation = Operation.Disable;
    } else {
        operation = Operation.Enable;
    }

    saveUpdateAction(groupId, operation);
}
 

@Override
public void match(final ProcessorNode component, final SearchQuery query, final List<String> matches) {
    final String searchTerm = query.getTerm();

    if (ScheduledState.DISABLED.equals(component.getScheduledState())) {
        if (StringUtils.containsIgnoreCase(SEARCH_TERM_DISABLED, searchTerm)) {
            matches.add(MATCH_PREFIX + MATCH_DISABLED);
        }
    } else if (StringUtils.containsIgnoreCase(SEARCH_TERM_INVALID, searchTerm) && component.getValidationStatus() == ValidationStatus.INVALID) {
        matches.add(MATCH_PREFIX + MATCH_INVALID);
    } else if (StringUtils.containsIgnoreCase(SEARCH_TERM_VALIDATING, searchTerm) && component.getValidationStatus() == ValidationStatus.VALIDATING) {
        matches.add(MATCH_PREFIX + MATCH_VALIDATING);
    } else if (ScheduledState.RUNNING.equals(component.getScheduledState()) && StringUtils.containsIgnoreCase(SEARCH_TERM_RUNNING, searchTerm)) {
        matches.add(MATCH_PREFIX + MATCH_RUNNING);
    } else if (ScheduledState.STOPPED.equals(component.getScheduledState()) && StringUtils.containsIgnoreCase(SEARCH_TERM_STOPPED, searchTerm)) {
        matches.add(MATCH_PREFIX + MATCH_STOPPED);
    }
}
 

private void restartProcessors(final Set<AffectedComponentEntity> processors, final AsynchronousWebRequest<?, ?> asyncRequest, final ComponentLifecycle componentLifecycle, final URI uri)
    throws ResumeFlowException, LifecycleManagementException {

    if (logger.isDebugEnabled()) {
        logger.debug("Restarting {} Processors after having updated Parameter Context: {}", processors.size(), processors);
    } else {
        logger.info("Restarting {} Processors after having updated Parameter Context", processors.size());
    }

    // Step 14. Restart all components
    final Set<AffectedComponentEntity> componentsToStart = getUpdatedEntities(processors);

    final CancellableTimedPause startComponentsPause = new CancellableTimedPause(250, Long.MAX_VALUE, TimeUnit.MILLISECONDS);
    asyncRequest.setCancelCallback(startComponentsPause::cancel);

    try {
        componentLifecycle.scheduleComponents(uri, "root", componentsToStart, ScheduledState.RUNNING, startComponentsPause, InvalidComponentAction.SKIP);
        asyncRequest.markStepComplete();
    } catch (final IllegalStateException ise) {
        // Component Lifecycle will restart the Processors only if they are valid. If IllegalStateException gets thrown, we need to provide
        // a more intelligent error message as to exactly what happened, rather than indicate that the flow could not be updated.
        throw new ResumeFlowException("Failed to restart components because " + ise.getMessage(), ise);
    }
}
 

@Override
public void verifyUpdateReferencingComponents(final String controllerServiceId, final ScheduledState scheduledState, final ControllerServiceState controllerServiceState) {
    final ControllerServiceNode controllerService = locateControllerService(controllerServiceId);

    if (controllerServiceState != null) {
        if (ControllerServiceState.ENABLED.equals(controllerServiceState)) {
            serviceProvider.verifyCanEnableReferencingServices(controllerService);
        } else {
            serviceProvider.verifyCanDisableReferencingServices(controllerService);
        }
    } else if (scheduledState != null) {
        if (ScheduledState.RUNNING.equals(scheduledState)) {
            serviceProvider.verifyCanScheduleReferencingComponents(controllerService);
        } else {
            serviceProvider.verifyCanStopReferencingComponents(controllerService);
        }
    }
}
 

private void stopComponents(final String processGroupId, final Map<String, Revision> componentRevisions, final Map<String, AffectedComponentEntity> affectedComponents, final Pause pause,
                            final InvalidComponentAction invalidComponentAction) throws LifecycleManagementException {

    if (componentRevisions.isEmpty()) {
        return;
    }

    logger.debug("Stopping components with ID's {} from Process Group {}", componentRevisions.keySet(), processGroupId);

    serviceFacade.verifyScheduleComponents(processGroupId, ScheduledState.STOPPED, componentRevisions.keySet());
    serviceFacade.scheduleComponents(processGroupId, ScheduledState.STOPPED, componentRevisions);

    // wait for all of the Processors to reach the desired state. We don't have to wait for other components because
    // Local and Remote Ports as well as funnels stop immediately.
    waitForProcessorState(processGroupId, affectedComponents, ScheduledState.STOPPED, pause, invalidComponentAction);
}
 

/**
 * Converts a set of ports into a set of remote process group ports.
 *
 * @param ports to convert
 * @return descriptors of ports
 */
private Set<RemoteProcessGroupPortDescriptor> convertRemotePort(final Set<PortDTO> ports) {
    Set<RemoteProcessGroupPortDescriptor> remotePorts = null;
    if (ports != null) {
        remotePorts = new LinkedHashSet<>(ports.size());
        for (final PortDTO port : ports) {
            final StandardRemoteProcessGroupPortDescriptor descriptor = new StandardRemoteProcessGroupPortDescriptor();
            final ScheduledState scheduledState = ScheduledState.valueOf(port.getState());
            descriptor.setId(port.getId());
            descriptor.setName(port.getName());
            descriptor.setComments(port.getComments());
            descriptor.setTargetRunning(ScheduledState.RUNNING.equals(scheduledState));
            remotePorts.add(descriptor);
        }
    }
    return remotePorts;
}
 

@Test
public void validateDisableOperation() throws Exception {
    final FlowManagerAndSystemBundle fcsb = this.buildFlowControllerForTest();
    flowManager = fcsb.getFlowManager();

    ProcessGroup testGroup = flowManager.createProcessGroup(UUID.randomUUID().toString());
    final ProcessorNode testProcNode = flowManager.createProcessor(TestProcessor.class.getName(),
            UUID.randomUUID().toString(), fcsb.getSystemBundle().getBundleDetails().getCoordinate());
    testProcNode.setProperties(properties);
    assertCondition(() -> ScheduledState.STOPPED == testProcNode.getScheduledState());
    assertCondition(() -> ScheduledState.STOPPED == testProcNode.getPhysicalScheduledState());
    // validates idempotency
    for (int i = 0; i < 2; i++) {
        testProcNode.disable();
    }
    assertCondition(() -> ScheduledState.DISABLED == testProcNode.getScheduledState());
    assertCondition(() -> ScheduledState.DISABLED == testProcNode.getPhysicalScheduledState());

    testProcNode.performValidation();
    processScheduler.startProcessor(testProcNode, true);
    assertCondition(() -> ScheduledState.DISABLED == testProcNode.getPhysicalScheduledState());
}
 

@Override
public Set<ComponentNode> updateControllerServiceReferencingComponents(
        final String controllerServiceId, final ScheduledState scheduledState, final ControllerServiceState controllerServiceState) {
    // get the controller service
    final ControllerServiceNode controllerService = locateControllerService(controllerServiceId);

    // this request is either acting upon referencing services or schedulable components
    if (controllerServiceState != null) {
        if (ControllerServiceState.ENABLED.equals(controllerServiceState)) {
            return serviceProvider.enableReferencingServices(controllerService);
        } else {
            return serviceProvider.disableReferencingServices(controllerService);
        }
    } else if (scheduledState != null) {
        if (ScheduledState.RUNNING.equals(scheduledState)) {
            return serviceProvider.scheduleReferencingComponents(controllerService);
        } else {
            return serviceProvider.unscheduleReferencingComponents(controllerService);
        }
    }

    return Collections.emptySet();
}
 

/**
 * Validates that stop calls are harmless and idempotent if processor is not
 * in STARTING or RUNNING state.
 */
@Test
public void validateStopCallsAreMeaninglessIfProcessorNotStarted() throws Exception {
    final FlowManagerAndSystemBundle fcsb = this.buildFlowControllerForTest();
    flowManager = fcsb.getFlowManager();
    ProcessGroup testGroup = flowManager.createProcessGroup(UUID.randomUUID().toString());
    final ProcessorNode testProcNode = flowManager.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString(),
            fcsb.getSystemBundle().getBundleDetails().getCoordinate());
    testProcNode.setProperties(properties);
    TestProcessor testProcessor = (TestProcessor) testProcNode.getProcessor();
    assertCondition(() -> ScheduledState.STOPPED == testProcNode.getScheduledState());
    // sets the scenario for the processor to run
    int randomDelayLimit = 3000;
    this.randomOnTriggerDelay(testProcessor, randomDelayLimit);

    processScheduler.stopProcessor(testProcNode);
    assertCondition(() -> ScheduledState.STOPPED == testProcNode.getScheduledState());
    assertTrue(testProcessor.operationNames.isEmpty());
}
 

/**
 * Waits for all of the given Processors to reach the given Scheduled State.
 *
 * @return <code>true</code> if all processors have reached the desired state, false if the given {@link Pause} indicates
 *         to give up before all of the processors have reached the desired state
 */
private boolean waitForProcessorState(final String groupId, final Map<String, AffectedComponentEntity> affectedComponents,
    final ScheduledState desiredState, final Pause pause, final InvalidComponentAction invalidComponentAction) throws LifecycleManagementException {

    logger.debug("Waiting for {} processors to transition their states to {}", affectedComponents.size(), desiredState);

    boolean continuePolling = true;
    while (continuePolling) {
        final Set<ProcessorEntity> processorEntities = serviceFacade.getProcessors(groupId, true);

        if (isProcessorActionComplete(processorEntities, affectedComponents, desiredState, invalidComponentAction)) {
            logger.debug("All {} processors of interest now have the desired state of {}", affectedComponents.size(), desiredState);
            return true;
        }

        // Not all of the processors are in the desired state. Pause for a bit and poll again.
        continuePolling = pause.pause();
    }

    return false;
}
 

public static Port getPort(
        final String id,
        final String name,
        final String comments,
        final ScheduledState scheduledState,
        final boolean isValid,
        final boolean isAuthorized) {
    return getPort(Port.class, id, name, Optional.empty(), comments, scheduledState, isValid, isAuthorized);
}
 

@Override
public ScheduleComponentsEntity scheduleComponents(final String processGroupId, final ScheduledState state, final Map<String, Revision> componentRevisions) {
    final NiFiUser user = NiFiUserUtils.getNiFiUser();
    final RevisionUpdate<ScheduleComponentsEntity> updatedComponent = revisionManager.updateRevision(new StandardRevisionClaim(componentRevisions.values()), user, new
            UpdateRevisionTask<ScheduleComponentsEntity>() {
                @Override
                public RevisionUpdate<ScheduleComponentsEntity> update() {
                    // schedule the components
                    processGroupDAO.scheduleComponents(processGroupId, state, componentRevisions.keySet());

                    // update the revisions
                    final Map<String, Revision> updatedRevisions = new HashMap<>();
                    for (final Revision revision : componentRevisions.values()) {
                        final Revision currentRevision = revisionManager.getRevision(revision.getComponentId());
                        updatedRevisions.put(revision.getComponentId(), currentRevision.incrementRevision(revision.getClientId()));
                    }

                    // save
                    controllerFacade.save();

                    // gather details for response
                    final ScheduleComponentsEntity entity = new ScheduleComponentsEntity();
                    entity.setId(processGroupId);
                    entity.setState(state.name());
                    return new StandardRevisionUpdate<>(entity, null, new HashSet<>(updatedRevisions.values()));
                }
            });

    return updatedComponent.getComponent();
}
 

public synchronized void enableReportingTask(final ReportingTaskNode taskNode) {
    if (taskNode.getScheduledState() != ScheduledState.DISABLED) {
        throw new IllegalStateException("Reporting Task cannot be enabled because it is not disabled");
    }

    taskNode.setScheduledState(ScheduledState.STOPPED);
}
 

public static PortDTO getPort(final Element element) {
    final PortDTO portDTO = new PortDTO();
    portDTO.setId(getString(element, "id"));
    portDTO.setVersionedComponentId(getString(element, "versionedComponentId"));
    portDTO.setPosition(getPosition(DomUtils.getChild(element, "position")));
    portDTO.setName(getString(element, "name"));
    portDTO.setComments(getString(element, "comments"));
    portDTO.setAllowRemoteAccess(getBoolean(element, "allowRemoteAccess"));
    final ScheduledState scheduledState = getScheduledState(element);
    portDTO.setState(scheduledState.toString());

    final List<Element> maxTasksElements = getChildrenByTagName(element, "maxConcurrentTasks");
    if (!maxTasksElements.isEmpty()) {
        portDTO.setConcurrentlySchedulableTaskCount(Integer.parseInt(maxTasksElements.get(0).getTextContent()));
    }

    final List<Element> userAccessControls = getChildrenByTagName(element, "userAccessControl");
    if (userAccessControls != null && !userAccessControls.isEmpty()) {
        final Set<String> users = new HashSet<>();
        portDTO.setUserAccessControl(users);
        for (final Element userElement : userAccessControls) {
            users.add(userElement.getTextContent());
        }
    }

    final List<Element> groupAccessControls = getChildrenByTagName(element, "groupAccessControl");
    if (groupAccessControls != null && !groupAccessControls.isEmpty()) {
        final Set<String> groups = new HashSet<>();
        portDTO.setGroupAccessControl(groups);
        for (final Element groupElement : groupAccessControls) {
            groups.add(groupElement.getTextContent());
        }
    }

    return portDTO;
}