/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.nifi.controller;

import org.apache.commons.lang3.builder.EqualsBuilder;
import org.apache.commons.lang3.builder.HashCodeBuilder;
import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
import org.apache.nifi.annotation.behavior.Restricted;
import org.apache.nifi.annotation.behavior.SideEffectFree;
import org.apache.nifi.annotation.behavior.TriggerWhenAnyDestinationAvailable;
import org.apache.nifi.annotation.behavior.TriggerWhenEmpty;
import org.apache.nifi.annotation.configuration.DefaultSchedule;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.DeprecationNotice;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.annotation.lifecycle.OnStopped;
import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
import org.apache.nifi.authorization.Resource;
import org.apache.nifi.authorization.resource.Authorizable;
import org.apache.nifi.authorization.resource.ResourceFactory;
import org.apache.nifi.authorization.resource.ResourceType;
import org.apache.nifi.bundle.BundleCoordinate;
import org.apache.nifi.components.ConfigurableComponent;
import org.apache.nifi.components.ValidationContext;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.components.validation.ValidationState;
import org.apache.nifi.components.validation.ValidationStatus;
import org.apache.nifi.components.validation.ValidationTrigger;
import org.apache.nifi.connectable.Connectable;
import org.apache.nifi.connectable.ConnectableType;
import org.apache.nifi.connectable.Connection;
import org.apache.nifi.connectable.Position;
import org.apache.nifi.controller.exception.ProcessorInstantiationException;
import org.apache.nifi.controller.scheduling.LifecycleState;
import org.apache.nifi.controller.scheduling.SchedulingAgent;
import org.apache.nifi.controller.service.ControllerServiceNode;
import org.apache.nifi.controller.service.ControllerServiceProvider;
import org.apache.nifi.controller.tasks.ActiveTask;
import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.logging.LogLevel;
import org.apache.nifi.logging.LogRepositoryFactory;
import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.nar.NarCloseable;
import org.apache.nifi.parameter.ParameterContext;
import org.apache.nifi.parameter.ParameterLookup;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSessionFactory;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.SimpleProcessLogger;
import org.apache.nifi.registry.ComponentVariableRegistry;
import org.apache.nifi.scheduling.ExecutionNode;
import org.apache.nifi.scheduling.SchedulingStrategy;
import org.apache.nifi.util.CharacterFilterUtils;
import org.apache.nifi.util.FormatUtils;
import org.apache.nifi.util.ReflectionUtils;
import org.apache.nifi.util.ThreadUtils;
import org.apache.nifi.util.file.classloader.ClassLoaderUtils;
import org.quartz.CronExpression;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.management.ThreadInfo;
import java.lang.reflect.InvocationTargetException;
import java.net.URL;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static java.util.Objects.requireNonNull;

/**
 * ProcessorNode provides thread-safe access to a FlowFileProcessor as it exists
 * within a controlled flow. This node keeps track of the processor, its
 * scheduling information and its relationships to other processors and whatever
 * scheduled futures exist for it. Must be thread safe.
 *
 */
public class StandardProcessorNode extends ProcessorNode implements Connectable {

    private static final Logger LOG = LoggerFactory.getLogger(StandardProcessorNode.class);

    public static final String BULLETIN_OBSERVER_ID = "bulletin-observer";

    public static final TimeUnit DEFAULT_TIME_UNIT = TimeUnit.MILLISECONDS;
    public static final String DEFAULT_YIELD_PERIOD = "1 sec";
    public static final String DEFAULT_PENALIZATION_PERIOD = "30 sec";
    private final AtomicReference<ProcessGroup> processGroup;
    private final AtomicReference<ProcessorDetails> processorRef;
    private final AtomicReference<String> identifier;
    private final Map<Connection, Connectable> destinations;
    private final Map<Relationship, Set<Connection>> connections;
    private final AtomicReference<Set<Relationship>> undefinedRelationshipsToTerminate;
    private final AtomicReference<List<Connection>> incomingConnections;
    private final AtomicBoolean lossTolerant;
    private final AtomicReference<String> comments;
    private final AtomicReference<Position> position;
    private final AtomicReference<String> schedulingPeriod; // stored as string so it's presented to user as they entered it
    private final AtomicReference<String> yieldPeriod;
    private final AtomicReference<String> penalizationPeriod;
    private final AtomicReference<Map<String, String>> style;
    private final AtomicInteger concurrentTaskCount;
    private final AtomicLong yieldExpiration;
    private final AtomicLong schedulingNanos;
    private final AtomicReference<String> versionedComponentId = new AtomicReference<>();
    private final ProcessScheduler processScheduler;
    private long runNanos = 0L;
    private volatile long yieldNanos;
    private volatile ScheduledState desiredState = ScheduledState.STOPPED;
    private volatile LogLevel bulletinLevel = LogLevel.WARN;

    private SchedulingStrategy schedulingStrategy; // guarded by synchronized keyword
    private ExecutionNode executionNode;
    private final Map<Thread, ActiveTask> activeThreads = new HashMap<>(48);
    private final int hashCode;
    private volatile boolean hasActiveThreads = false;

    public StandardProcessorNode(final LoggableComponent<Processor> processor, final String uuid,
                                 final ValidationContextFactory validationContextFactory, final ProcessScheduler scheduler,
                                 final ControllerServiceProvider controllerServiceProvider, final ComponentVariableRegistry variableRegistry,
                                 final ReloadComponent reloadComponent, final ExtensionManager extensionManager, final ValidationTrigger validationTrigger) {

        this(processor, uuid, validationContextFactory, scheduler, controllerServiceProvider, processor.getComponent().getClass().getSimpleName(),
            processor.getComponent().getClass().getCanonicalName(), variableRegistry, reloadComponent, extensionManager, validationTrigger, false);
    }

    public StandardProcessorNode(final LoggableComponent<Processor> processor, final String uuid,
                                 final ValidationContextFactory validationContextFactory, final ProcessScheduler scheduler,
                                 final ControllerServiceProvider controllerServiceProvider,
                                 final String componentType, final String componentCanonicalClass, final ComponentVariableRegistry variableRegistry,
                                 final ReloadComponent reloadComponent, final ExtensionManager extensionManager, final ValidationTrigger validationTrigger,
                                 final boolean isExtensionMissing) {

        super(uuid, validationContextFactory, controllerServiceProvider, componentType, componentCanonicalClass, variableRegistry, reloadComponent,
                extensionManager, validationTrigger, isExtensionMissing);

        final ProcessorDetails processorDetails = new ProcessorDetails(processor);
        this.processorRef = new AtomicReference<>(processorDetails);

        identifier = new AtomicReference<>(uuid);
        destinations = new HashMap<>();
        connections = new HashMap<>();
        incomingConnections = new AtomicReference<>(new ArrayList<>());
        lossTolerant = new AtomicBoolean(false);
        final Set<Relationship> emptySetOfRelationships = new HashSet<>();
        undefinedRelationshipsToTerminate = new AtomicReference<>(emptySetOfRelationships);
        comments = new AtomicReference<>("");
        schedulingPeriod = new AtomicReference<>("0 sec");
        schedulingNanos = new AtomicLong(MINIMUM_SCHEDULING_NANOS);
        yieldPeriod = new AtomicReference<>(DEFAULT_YIELD_PERIOD);
        yieldExpiration = new AtomicLong(0L);
        concurrentTaskCount = new AtomicInteger(1);
        position = new AtomicReference<>(new Position(0D, 0D));
        style = new AtomicReference<>(Collections.unmodifiableMap(new HashMap<String, String>()));
        this.processGroup = new AtomicReference<>();
        processScheduler = scheduler;
        penalizationPeriod = new AtomicReference<>(DEFAULT_PENALIZATION_PERIOD);

        schedulingStrategy = SchedulingStrategy.TIMER_DRIVEN;
        executionNode = isExecutionNodeRestricted() ? ExecutionNode.PRIMARY : ExecutionNode.ALL;
        this.hashCode = new HashCodeBuilder(7, 67).append(identifier).toHashCode();

        try {
            if (processorDetails.getProcClass().isAnnotationPresent(DefaultSchedule.class)) {
                DefaultSchedule dsc = processorDetails.getProcClass().getAnnotation(DefaultSchedule.class);
                try {
                    this.setSchedulingStrategy(dsc.strategy());
                } catch (Throwable ex) {
                    LOG.error(String.format("Error while setting scheduling strategy from DefaultSchedule annotation: %s", ex.getMessage()), ex);
                }
                try {
                    this.setScheduldingPeriod(dsc.period());
                } catch (Throwable ex) {
                    this.setSchedulingStrategy(SchedulingStrategy.TIMER_DRIVEN);
                    LOG.error(String.format("Error while setting scheduling period from DefaultSchedule annotation: %s", ex.getMessage()), ex);
                }
                if (!processorDetails.isTriggeredSerially()) {
                    try {
                        setMaxConcurrentTasks(dsc.concurrentTasks());
                    } catch (Throwable ex) {
                        LOG.error(String.format("Error while setting max concurrent tasks from DefaultSchedule annotation: %s", ex.getMessage()), ex);
                    }
                }
            }
        } catch (Throwable ex) {
            LOG.error(String.format("Error while setting default schedule from DefaultSchedule annotation: %s",ex.getMessage()),ex);
        }
    }

    @Override
    public ConfigurableComponent getComponent() {
        return processorRef.get().getProcessor();
    }

    @Override
    public TerminationAwareLogger getLogger() {
        return processorRef.get().getComponentLog();
    }

    @Override
    public Object getRunnableComponent() {
        return getProcessor();
    }

    @Override
    public BundleCoordinate getBundleCoordinate() {
        return processorRef.get().getBundleCoordinate();
    }

    /**
     * @return comments about this specific processor instance
     */
    @Override
    public String getComments() {
        return comments.get();
    }

    @Override
    public Authorizable getParentAuthorizable() {
        return getProcessGroup();
    }

    @Override
    public Resource getResource() {
        return ResourceFactory.getComponentResource(ResourceType.Processor, getIdentifier(), getName());
    }

    @Override
    public boolean isRestricted() {
        return getProcessor().getClass().isAnnotationPresent(Restricted.class);
    }

    @Override
    public Class<?> getComponentClass() {
        return getProcessor().getClass();
    }

    @Override
    public boolean isDeprecated() {
        return getProcessor().getClass().isAnnotationPresent(DeprecationNotice.class);
    }


    /**
     * Provides and opportunity to retain information about this particular
     * processor instance
     *
     * @param comments
     *            new comments
     */
    @Override
    public synchronized void setComments(final String comments) {
        this.comments.set(CharacterFilterUtils.filterInvalidXmlCharacters(comments));
    }

    @Override
    public Position getPosition() {
        return position.get();
    }

    @Override
    public synchronized void setPosition(final Position position) {
        this.position.set(position);
    }

    @Override
    public Map<String, String> getStyle() {
        return style.get();
    }

    @Override
    public synchronized void setStyle(final Map<String, String> style) {
        if (style != null) {
            this.style.set(Collections.unmodifiableMap(new HashMap<>(style)));
        }
    }

    @Override
    public String getIdentifier() {
        return identifier.get();
    }

    /**
     * @return if true flow file content generated by this processor is
     *         considered loss tolerant
     */
    @Override
    public boolean isLossTolerant() {
        return lossTolerant.get();
    }

    @Override
    @SuppressWarnings("deprecation")
    public boolean isIsolated() {
        return schedulingStrategy == SchedulingStrategy.PRIMARY_NODE_ONLY || executionNode == ExecutionNode.PRIMARY;
    }

    /**
     * @return true if the processor has the {@link TriggerWhenEmpty}
     *         annotation, false otherwise.
     */
    @Override
    public boolean isTriggerWhenEmpty() {
        return processorRef.get().isTriggerWhenEmpty();
    }

    /**
     * @return true if the processor has the {@link SideEffectFree} annotation,
     *         false otherwise.
     */
    @Override
    public boolean isSideEffectFree() {
        return processorRef.get().isSideEffectFree();
    }

    @Override
    public boolean isSessionBatchingSupported() {
        return processorRef.get().isBatchSupported();
    }

    /**
     * @return true if the processor has the
     *         {@link TriggerWhenAnyDestinationAvailable} annotation, false
     *         otherwise.
     */
    @Override
    public boolean isTriggerWhenAnyDestinationAvailable() {
        return processorRef.get().isTriggerWhenAnyDestinationAvailable();
    }

    /**
     *  Indicates whether the processor's executionNode configuration is restricted to run only in primary node
     */
    @Override
    public boolean isExecutionNodeRestricted(){
        return processorRef.get().isExecutionNodeRestricted();
    }

    /**
     * Indicates whether flow file content made by this processor must be
     * persisted
     *
     * @param lossTolerant
     *            tolerant
     */
    @Override
    public synchronized void setLossTolerant(final boolean lossTolerant) {
        if (isRunning()) {
            throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
        }
        this.lossTolerant.set(lossTolerant);
    }

    @Override
    public boolean isAutoTerminated(final Relationship relationship) {
        if (relationship.isAutoTerminated() && getConnections(relationship).isEmpty()) {
            return true;
        }
        final Set<Relationship> terminatable = undefinedRelationshipsToTerminate.get();
        return terminatable == null ? false : terminatable.contains(relationship);
    }

    @Override
    public void setAutoTerminatedRelationships(final Set<Relationship> terminate) {
        if (isRunning()) {
            throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
        }

        for (final Relationship rel : terminate) {
            if (!getConnections(rel).isEmpty()) {
                throw new IllegalStateException("Cannot mark relationship '" + rel.getName()
                        + "' as auto-terminated because Connection already exists with this relationship");
            }
        }

        undefinedRelationshipsToTerminate.set(new HashSet<>(terminate));
        LOG.debug("Resetting Validation State of {} due to setting auto-terminated relationships", this);
        resetValidationState();
    }

    /**
     * @return an unmodifiable Set that contains all of the
     *         ProcessorRelationship objects that are configured to be
     *         auto-terminated
     */
    @Override
    public Set<Relationship> getAutoTerminatedRelationships() {
        Set<Relationship> relationships = undefinedRelationshipsToTerminate.get();
        if (relationships == null) {
            relationships = new HashSet<>();
        }
        return Collections.unmodifiableSet(relationships);
    }

    /**
     * @return the value of the processor's {@link CapabilityDescription}
     *         annotation, if one exists, else <code>null</code>.
     */
    public String getProcessorDescription() {
        final Processor processor = processorRef.get().getProcessor();
        final CapabilityDescription capDesc = processor.getClass().getAnnotation(CapabilityDescription.class);
        String description = null;
        if (capDesc != null) {
            description = capDesc.value();
        }
        return description;
    }

    @Override
    public synchronized void setName(final String name) {
        super.setName(name);
    }

    /**
     * @param timeUnit determines the unit of time to represent the scheduling period.
     * @return the schedule period that should elapse before subsequent cycles of this processor's tasks
     */
    @Override
    public long getSchedulingPeriod(final TimeUnit timeUnit) {
        return timeUnit.convert(schedulingNanos.get(), TimeUnit.NANOSECONDS);
    }

    @Override
    public boolean isEventDrivenSupported() {
        return processorRef.get().isEventDrivenSupported();
    }

    /**
     * Updates the Scheduling Strategy used for this Processor
     *
     * @param schedulingStrategy
     *            strategy
     *
     * @throws IllegalArgumentException
     *             if the SchedulingStrategy is not not applicable for this
     *             Processor
     */
    @Override
    public synchronized void setSchedulingStrategy(final SchedulingStrategy schedulingStrategy) {
        if (schedulingStrategy == SchedulingStrategy.EVENT_DRIVEN && !processorRef.get().isEventDrivenSupported()) {
            // not valid. Just ignore it. We don't throw an Exception because if
            // a developer changes a Processor so that
            // it no longer supports EventDriven mode, we don't want the app to
            // fail to startup if it was already in Event-Driven
            // Mode. Instead, we will simply leave it in Timer-Driven mode
            return;
        }

        this.schedulingStrategy = schedulingStrategy;
    }

    /**
     * @return the currently configured scheduling strategy
     */
    @Override
    public SchedulingStrategy getSchedulingStrategy() {
        return this.schedulingStrategy;
    }

    @Override
    public String getSchedulingPeriod() {
        return schedulingPeriod.get();
    }

    @Override
    @SuppressWarnings("deprecation")
    public synchronized void setScheduldingPeriod(final String schedulingPeriod) {
        if (isRunning()) {
            throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
        }

        switch (schedulingStrategy) {
        case CRON_DRIVEN: {
            try {
                new CronExpression(schedulingPeriod);
            } catch (final Exception e) {
                throw new IllegalArgumentException(
                        "Scheduling Period is not a valid cron expression: " + schedulingPeriod);
            }
        }
            break;
        case PRIMARY_NODE_ONLY:
        case TIMER_DRIVEN: {
            final long schedulingNanos = FormatUtils.getTimeDuration(requireNonNull(schedulingPeriod),
                    TimeUnit.NANOSECONDS);
            if (schedulingNanos < 0) {
                throw new IllegalArgumentException("Scheduling Period must be positive");
            }
            this.schedulingNanos.set(Math.max(MINIMUM_SCHEDULING_NANOS, schedulingNanos));
        }
            break;
        case EVENT_DRIVEN:
        default:
            return;
        }

        this.schedulingPeriod.set(schedulingPeriod);
    }

    @Override
    public synchronized void setExecutionNode(final ExecutionNode executionNode) {
        if (this.isExecutionNodeRestricted()) {
            this.executionNode = ExecutionNode.PRIMARY;
        } else {
            this.executionNode = executionNode;
        }
    }

    @Override
    public ExecutionNode getExecutionNode() {
        return this.executionNode;
    }

    @Override
    public long getRunDuration(final TimeUnit timeUnit) {
        return timeUnit.convert(this.runNanos, TimeUnit.NANOSECONDS);
    }

    @Override
    public synchronized void setRunDuration(final long duration, final TimeUnit timeUnit) {
        if (duration < 0) {
            throw new IllegalArgumentException("Run Duration must be non-negative value; cannot set to "
                    + timeUnit.toSeconds(duration) + " seconds");
        }

        this.runNanos = timeUnit.toNanos(duration);
    }

    @Override
    public long getYieldPeriod(final TimeUnit timeUnit) {
        final TimeUnit unit = (timeUnit == null ? DEFAULT_TIME_UNIT : timeUnit);
        return unit.convert(yieldNanos, TimeUnit.NANOSECONDS);
    }

    @Override
    public String getYieldPeriod() {
        return yieldPeriod.get();
    }

    @Override
    public synchronized void setYieldPeriod(final String yieldPeriod) {
        if (isRunning()) {
            throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
        }
        final long yieldNanos = FormatUtils.getTimeDuration(requireNonNull(yieldPeriod), TimeUnit.NANOSECONDS);
        if (yieldNanos < 0) {
            throw new IllegalArgumentException("Yield duration must be positive");
        }
        this.yieldPeriod.set(yieldPeriod);
        this.yieldNanos = yieldNanos;
    }

    /**
     * Causes the processor not to be scheduled for some period of time. This
     * duration can be obtained and set via the
     * {@link #getYieldPeriod(TimeUnit)} and
     * {@link #setYieldPeriod(String)}.
     */
    @Override
    public void yield() {
        final Processor processor = processorRef.get().getProcessor();
        final long yieldMillis = getYieldPeriod(TimeUnit.MILLISECONDS);
        yield(yieldMillis, TimeUnit.MILLISECONDS);

        final String yieldDuration = (yieldMillis > 1000) ? (yieldMillis / 1000) + " seconds"
                : yieldMillis + " milliseconds";
        LoggerFactory.getLogger(processor.getClass()).debug(
                "{} has chosen to yield its resources; will not be scheduled to run again for {}", processor,
                yieldDuration);
    }

    @Override
    public void yield(final long period, final TimeUnit timeUnit) {
        final long yieldMillis = TimeUnit.MILLISECONDS.convert(period, timeUnit);
        yieldExpiration.set(Math.max(yieldExpiration.get(), System.currentTimeMillis() + yieldMillis));

        processScheduler.yield(this);
    }

    /**
     * @return the number of milliseconds since Epoch at which time this
     *         processor is to once again be scheduled.
     */
    @Override
    public long getYieldExpiration() {
        return yieldExpiration.get();
    }

    @Override
    public long getPenalizationPeriod(final TimeUnit timeUnit) {
        return FormatUtils.getTimeDuration(getPenalizationPeriod(), timeUnit == null ? DEFAULT_TIME_UNIT : timeUnit);
    }

    @Override
    public String getPenalizationPeriod() {
        return penalizationPeriod.get();
    }

    @Override
    public synchronized void setPenalizationPeriod(final String penalizationPeriod) {
        if (isRunning()) {
            throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
        }

        final long penalizationMillis = FormatUtils.getTimeDuration(requireNonNull(penalizationPeriod), TimeUnit.MILLISECONDS);
        if (penalizationMillis < 0) {
            throw new IllegalArgumentException("Penalization duration must be positive");
        }

        this.penalizationPeriod.set(penalizationPeriod);
    }

    /**
     * Determines the number of concurrent tasks that may be running for this
     * processor.
     *
     * @param taskCount
     *            a number of concurrent tasks this processor may have running
     * @throws IllegalArgumentException
     *             if the given value is less than 1
     */
    @Override
    public synchronized void setMaxConcurrentTasks(final int taskCount) {
        if (isRunning()) {
            throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
        }

        if (taskCount < 1 && getSchedulingStrategy() != SchedulingStrategy.EVENT_DRIVEN) {
            throw new IllegalArgumentException("Cannot set Concurrent Tasks to " + taskCount + " for component "
                    + getIdentifier() + " because Scheduling Strategy is not Event Driven");
        }

        if (!isTriggeredSerially()) {
            concurrentTaskCount.set(taskCount);
        }
    }

    @Override
    public boolean isTriggeredSerially() {
        return processorRef.get().isTriggeredSerially();
    }

    /**
     * @return the number of tasks that may execute concurrently for this processor
     */
    @Override
    public int getMaxConcurrentTasks() {
        return concurrentTaskCount.get();
    }

    @Override
    public LogLevel getBulletinLevel() {
        return bulletinLevel;
    }

    @Override
    public synchronized void setBulletinLevel(final LogLevel level) {
        LogRepositoryFactory.getRepository(getIdentifier()).setObservationLevel(BULLETIN_OBSERVER_ID, level);
        this.bulletinLevel = level;
    }

    @Override
    public Set<Connection> getConnections() {
        final Set<Connection> allConnections = new HashSet<>();
        for (final Set<Connection> connectionSet : connections.values()) {
            allConnections.addAll(connectionSet);
        }

        return allConnections;
    }

    @Override
    public List<Connection> getIncomingConnections() {
        return incomingConnections.get();
    }

    @Override
    public Set<Connection> getConnections(final Relationship relationship) {
        final Set<Connection> applicableConnections = connections.get(relationship);
        return (applicableConnections == null) ? Collections.<Connection> emptySet()
                : Collections.unmodifiableSet(applicableConnections);
    }

    @Override
    public void addConnection(final Connection connection) {
        Objects.requireNonNull(connection, "connection cannot be null");

        if (!connection.getSource().equals(this) && !connection.getDestination().equals(this)) {
            throw new IllegalStateException(
                    "Cannot a connection to a ProcessorNode for which the ProcessorNode is neither the Source nor the Destination");
        }

        try {
            List<Connection> updatedIncoming = null;
            if (connection.getDestination().equals(this)) {
                // don't add the connection twice. This may occur if we have a
                // self-loop because we will be told
                // to add the connection once because we are the source and again
                // because we are the destination.
                final List<Connection> incomingConnections = getIncomingConnections();
                updatedIncoming = new ArrayList<>(incomingConnections);
                if (!updatedIncoming.contains(connection)) {
                    updatedIncoming.add(connection);
                }
            }

            if (connection.getSource().equals(this)) {
                // don't add the connection twice. This may occur if we have a
                // self-loop because we will be told
                // to add the connection once because we are the source and again
                // because we are the destination.
                if (!destinations.containsKey(connection)) {
                    for (final Relationship relationship : connection.getRelationships()) {
                        final Relationship rel = getRelationship(relationship.getName());
                        Set<Connection> set = connections.get(rel);
                        if (set == null) {
                            set = new HashSet<>();
                            connections.put(rel, set);
                        }

                        set.add(connection);

                        destinations.put(connection, connection.getDestination());
                    }

                    final Set<Relationship> autoTerminated = this.undefinedRelationshipsToTerminate.get();
                    if (autoTerminated != null) {
                        autoTerminated.removeAll(connection.getRelationships());
                        this.undefinedRelationshipsToTerminate.set(autoTerminated);
                    }
                }
            }

            if (updatedIncoming != null) {
                setIncomingConnections(Collections.unmodifiableList(updatedIncoming));
            }
        } finally {
            LOG.debug("Resetting Validation State of {} due to connection added", this);
            resetValidationState();
        }
    }

    @Override
    public boolean hasIncomingConnection() {
        return !getIncomingConnections().isEmpty();
    }

    @Override
    public void updateConnection(final Connection connection) throws IllegalStateException {
        try {
            if (requireNonNull(connection).getSource().equals(this)) {
                // update any relationships
                //
                // first check if any relations were removed.
                final List<Relationship> existingRelationships = new ArrayList<>();
                for (final Map.Entry<Relationship, Set<Connection>> entry : connections.entrySet()) {
                    if (entry.getValue().contains(connection)) {
                        existingRelationships.add(entry.getKey());
                    }
                }

                for (final Relationship rel : connection.getRelationships()) {
                    if (!existingRelationships.contains(rel)) {
                        // relationship was removed. Check if this is legal.
                        final Set<Connection> connectionsForRelationship = getConnections(rel);
                        if (connectionsForRelationship != null && connectionsForRelationship.size() == 1 && this.isRunning()
                            && !isAutoTerminated(rel) && getRelationships().contains(rel)) {
                            // if we are running and we do not terminate undefined
                            // relationships and this is the only
                            // connection that defines the given relationship, and
                            // that relationship is required,
                            // then it is not legal to remove this relationship from
                            // this connection.
                            throw new IllegalStateException("Cannot remove relationship " + rel.getName()
                                + " from Connection because doing so would invalidate Processor " + this
                                + ", which is currently running");
                        }
                    }
                }

                // remove the connection from any list that currently contains
                for (final Set<Connection> list : connections.values()) {
                    list.remove(connection);
                }

                // add the connection in for all relationships listed.
                for (final Relationship rel : connection.getRelationships()) {
                    Set<Connection> set = connections.get(rel);
                    if (set == null) {
                        set = new HashSet<>();
                        connections.put(rel, set);
                    }
                    set.add(connection);
                }

                // update to the new destination
                destinations.put(connection, connection.getDestination());

                final Set<Relationship> autoTerminated = this.undefinedRelationshipsToTerminate.get();
                if (autoTerminated != null) {
                    autoTerminated.removeAll(connection.getRelationships());
                    this.undefinedRelationshipsToTerminate.set(autoTerminated);
                }
            }

            if (connection.getDestination().equals(this)) {
                // update our incoming connections -- we can just remove & re-add
                // the connection to update the list.
                final List<Connection> incomingConnections = getIncomingConnections();
                final List<Connection> updatedIncoming = new ArrayList<>(incomingConnections);
                updatedIncoming.remove(connection);
                updatedIncoming.add(connection);
                setIncomingConnections(Collections.unmodifiableList(updatedIncoming));
            }
        } finally {
            // need to perform validation in case selected relationships were changed.
            LOG.debug("Resetting Validation State of {} due to updating connection", this);
            resetValidationState();
        }
    }

    @Override
    public void removeConnection(final Connection connection) {
        boolean connectionRemoved = false;

        if (requireNonNull(connection).getSource().equals(this)) {
            for (final Relationship relationship : connection.getRelationships()) {
                final Set<Connection> connectionsForRelationship = getConnections(relationship);
                if ((connectionsForRelationship == null || connectionsForRelationship.size() <= 1) && isRunning()) {
                    throw new IllegalStateException(
                            "This connection cannot be removed because its source is running and removing it will invalidate this processor");
                }
            }

            for (final Set<Connection> connectionList : this.connections.values()) {
                connectionList.remove(connection);
            }

            connectionRemoved = (destinations.remove(connection) != null);
        }

        if (connection.getDestination().equals(this)) {
            final List<Connection> incomingConnections = getIncomingConnections();
            if (incomingConnections.contains(connection)) {
                final List<Connection> updatedIncoming = new ArrayList<>(incomingConnections);
                updatedIncoming.remove(connection);
                setIncomingConnections(Collections.unmodifiableList(updatedIncoming));
                return;
            }
        }

        if (!connectionRemoved) {
            throw new IllegalArgumentException(
                    "Cannot remove a connection from a ProcessorNode for which the ProcessorNode is not the Source");
        }

        LOG.debug("Resetting Validation State of {} due to connection removed", this);
        resetValidationState();
    }

    private void setIncomingConnections(final List<Connection> incoming) {
        this.incomingConnections.set(incoming);
        LOG.debug("Resetting Validation State of {} due to setting incoming connections", this);
        resetValidationState();
    }

    /**
     * @param relationshipName
     *            name
     * @return the relationship for this nodes processor for the given name or
     *         creates a new relationship for the given name
     */
    @Override
    public Relationship getRelationship(final String relationshipName) {
        final Relationship specRel = new Relationship.Builder().name(relationshipName).build();
        Relationship returnRel = specRel;

        final Set<Relationship> relationships;
        final Processor processor = processorRef.get().getProcessor();
        try (final NarCloseable narCloseable = NarCloseable.withComponentNarLoader(getExtensionManager(), processor.getClass(), processor.getIdentifier())) {
            relationships = processor.getRelationships();
        }

        for (final Relationship rel : relationships) {
            if (rel.equals(specRel)) {
                returnRel = rel;
                break;
            }
        }
        return returnRel;
    }

    @Override
    public Processor getProcessor() {
        return processorRef.get().getProcessor();
    }

    @Override
    public synchronized void setProcessor(final LoggableComponent<Processor> processor) {
        if (isRunning()) {
            throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
        }

        final ProcessorDetails processorDetails = new ProcessorDetails(processor);
        processorRef.set(processorDetails);
    }

    @Override
    public synchronized void reload(final Set<URL> additionalUrls) throws ProcessorInstantiationException {
        if (isRunning()) {
            throw new IllegalStateException("Cannot reload Processor while the Processor is running");
        }
        String additionalResourcesFingerprint = ClassLoaderUtils.generateAdditionalUrlsFingerprint(additionalUrls);
        setAdditionalResourcesFingerprint(additionalResourcesFingerprint);
        getReloadComponent().reload(this, getCanonicalClassName(), getBundleCoordinate(), additionalUrls);
    }

    /**
     * @return the Set of destination processors for all relationships excluding
     *         any destinations that are this processor itself (self-loops)
     */
    public Set<Connectable> getDestinations() {
        final Set<Connectable> nonSelfDestinations = new HashSet<>();
        for (final Connectable connectable : destinations.values()) {
            if (connectable != this) {
                nonSelfDestinations.add(connectable);
            }
        }
        return nonSelfDestinations;
    }

    public Set<Connectable> getDestinations(final Relationship relationship) {
        final Set<Connectable> destinationSet = new HashSet<>();
        final Set<Connection> relationshipConnections = connections.get(relationship);
        if (relationshipConnections != null) {
            for (final Connection connection : relationshipConnections) {
                destinationSet.add(destinations.get(connection));
            }
        }
        return destinationSet;
    }

    public Set<Relationship> getUndefinedRelationships() {
        final Set<Relationship> undefined = new HashSet<>();
        final Set<Relationship> relationships;
        final Processor processor = processorRef.get().getProcessor();
        try (final NarCloseable narCloseable = NarCloseable.withComponentNarLoader(getExtensionManager(), processor.getClass(), processor.getIdentifier())) {
            relationships = processor.getRelationships();
        }

        if (relationships == null) {
            return undefined;
        }
        for (final Relationship relation : relationships) {
            final Set<Connection> connectionSet = this.connections.get(relation);
            if (connectionSet == null || connectionSet.isEmpty()) {
                undefined.add(relation);
            }
        }
        return undefined;
    }

    /**
     * Determines if the given node is a destination for this node
     *
     * @param node
     *            node
     * @return true if is a direct destination node; false otherwise
     */
    boolean isRelated(final ProcessorNode node) {
        return this.destinations.containsValue(node);
    }

    @Override
    public boolean isRunning() {
        return getScheduledState().equals(ScheduledState.RUNNING) || hasActiveThreads;
    }

    @Override
    public boolean isValidationNecessary() {
        switch (getPhysicalScheduledState()) {
            case STOPPED:
            case STOPPING:
            case STARTING:
                return true;
        }

        return false;
    }

    @Override
    public int getActiveThreadCount() {
        final int activeThreadCount = processScheduler.getActiveThreadCount(this);

        // When getScheduledState() is called, we map the 'physical' state of STOPPING to STOPPED. This is done in order to maintain
        // backward compatibility because the UI and other clients will not know of the (relatively newer) 'STOPPING' state.
        // Because of there previously was no STOPPING state, the way to determine of a processor had truly stopped was to check if its
        // Scheduled State was STOPPED AND it had no active threads.
        //
        // Also, we can have a situation in which a processor is started while invalid. Before the processor becomes valid, it can be stopped.
        // In this situation, the processor state will become STOPPING until the background thread checks the state, calls any necessary lifecycle methods,
        // and finally updates the state to STOPPED. In the interim, we have a situation where a call to getScheduledState() returns STOPPED and there are no
        // active threads, which the client will interpret as the processor being fully stopped. However, in this situation, an attempt to update the processor, etc.
        // will fail because the processor is not truly fully stopped.
        //
        // To prevent this situation, we return 1 for the number of active tasks when the processor is considered STOPPING. In doing this, we ensure that the condition
        // of (getScheduledState() == STOPPED and activeThreads == 0) never happens while the processor is still stopping.
        //
        // This probably is calling for a significant refactoring / rethinking of this class. It would make sense, for example, to extract some of the logic into a separate
        // StateTransition class as we've done with Controller Services. That would at least more cleanly encapsulate this logic. However, this is a simple enough work around for the time being.
        if (activeThreadCount == 0 && getPhysicalScheduledState() == ScheduledState.STOPPING) {
            return 1;
        }

        return activeThreadCount;
    }

    List<Connection> getIncomingNonLoopConnections() {
        final List<Connection> connections = getIncomingConnections();
        final List<Connection> nonLoopConnections = new ArrayList<>(connections.size());
        for (final Connection connection : connections) {
            if (!connection.getSource().equals(this)) {
                nonLoopConnections.add(connection);
            }
        }

        return nonLoopConnections;
    }


    @Override
    public Collection<ValidationResult> getValidationErrors() {
        final ValidationState validationState = getValidationState();
        return validationState.getValidationErrors();
    }

    @Override
    protected Collection<ValidationResult> computeValidationErrors(final ValidationContext validationContext) {
        final List<ValidationResult> results = new ArrayList<>();
        try {
            final Collection<ValidationResult> validationResults = super.computeValidationErrors(validationContext);

            validationResults.stream()
                .filter(result -> !result.isValid())
                .forEach(results::add);

            // Ensure that any relationships that don't have a connection defined are auto-terminated
            for (final Relationship relationship : getUndefinedRelationships()) {
                if (!isAutoTerminated(relationship)) {
                    final ValidationResult error = new ValidationResult.Builder()
                        .explanation("Relationship '" + relationship.getName()
                            + "' is not connected to any component and is not auto-terminated")
                        .subject("Relationship " + relationship.getName()).valid(false).build();
                    results.add(error);
                }
            }

            // Ensure that the requirements of the InputRequirement are met.
            switch (getInputRequirement()) {
                case INPUT_ALLOWED:
                    break;
                case INPUT_FORBIDDEN: {
                    final int incomingConnCount = getIncomingNonLoopConnections().size();
                    if (incomingConnCount != 0) {
                        results.add(new ValidationResult.Builder().explanation(
                            "Processor does not allow upstream connections but currently has " + incomingConnCount)
                            .subject("Upstream Connections").valid(false).build());
                    }
                    break;
                }
                case INPUT_REQUIRED: {
                    if (getIncomingNonLoopConnections().isEmpty()) {
                        results.add(new ValidationResult.Builder()
                            .explanation("Processor requires an upstream connection but currently has none")
                            .subject("Upstream Connections").valid(false).build());
                    }
                    break;
                }
            }

            // Ensure that execution node will not be misused
            if (getExecutionNode() == ExecutionNode.PRIMARY && hasIncomingConnection()) {
                results.add(new ValidationResult.Builder()
                        .explanation("Processors with incoming connections cannot be scheduled for Primary Node Only.")
                        .subject("Execution Node").valid(false).build());
            }
        } catch (final Throwable t) {
            LOG.error("Failed to perform validation", t);
            results.add(new ValidationResult.Builder().explanation("Failed to run validation due to " + t.toString())
                    .valid(false).build());
        }

        return results;
    }

    @Override
    public Requirement getInputRequirement() {
        return processorRef.get().getInputRequirement();
    }

    /**
     * Establishes node equality (based on the processor's identifier)
     *
     * @param other
     *            node
     * @return true if equal
     */
    @Override
    public boolean equals(final Object other) {
        if (!(other instanceof ProcessorNode)) {
            return false;
        }
        final ProcessorNode on = (ProcessorNode) other;
        return new EqualsBuilder().append(identifier.get(), on.getIdentifier()).isEquals();
    }

    @Override
    public int hashCode() {
        return hashCode;
    }

    @Override
    public Collection<Relationship> getRelationships() {
        final Processor processor = processorRef.get().getProcessor();
        try (final NarCloseable narCloseable = NarCloseable.withComponentNarLoader(getExtensionManager(), processor.getClass(), processor.getIdentifier())) {
            return getProcessor().getRelationships();
        }
    }

    @Override
    public String toString() {
        final Processor processor = processorRef.get().getProcessor();
        try (final NarCloseable narCloseable = NarCloseable.withComponentNarLoader(getExtensionManager(), processor.getClass(), processor.getIdentifier())) {
            return getProcessor().toString();
        }
    }

    @Override
    public ProcessGroup getProcessGroup() {
        return processGroup.get();
    }

    @Override
    protected ParameterContext getParameterContext() {
        final ProcessGroup processGroup = getProcessGroup();
        return processGroup == null ? null : processGroup.getParameterContext();
    }

    @Override
    public ParameterLookup getParameterLookup() {
        return getParameterContext();
    }

    @Override
    public synchronized void setProcessGroup(final ProcessGroup group) {
        this.processGroup.set(group);
        LOG.debug("Resetting Validation State of {} due to setting process group", this);
        resetValidationState();
    }

    @Override
    public void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory) {
        final Processor processor = processorRef.get().getProcessor();

        activateThread();
        try (final NarCloseable narCloseable = NarCloseable.withComponentNarLoader(getExtensionManager(), processor.getClass(), processor.getIdentifier())) {
            processor.onTrigger(context, sessionFactory);
        } finally {
            deactivateThread();
        }
    }

    @Override
    public ConnectableType getConnectableType() {
        return ConnectableType.PROCESSOR;
    }

    @Override
    public void setAnnotationData(final String data) {
        if (isRunning()) {
            throw new IllegalStateException("Cannot set AnnotationData while processor is running");
        }
        super.setAnnotationData(data);
    }

    @Override
    public void verifyCanDelete() throws IllegalStateException {
        verifyCanDelete(false);
    }

    @Override
    public void verifyCanDelete(final boolean ignoreConnections) {
        if (isRunning()) {
            throw new IllegalStateException(this.getIdentifier() + " is running");
        }

        if (!ignoreConnections) {
            for (final Set<Connection> connectionSet : connections.values()) {
                for (final Connection connection : connectionSet) {
                    connection.verifyCanDelete();
                }
            }

            for (final Connection connection : getIncomingConnections()) {
                if (connection.getSource().equals(this)) {
                    connection.verifyCanDelete();
                } else {
                    throw new IllegalStateException(this.getIdentifier() + " is the destination of another component");
                }
            }
        }
    }

    @Override
    public void verifyCanStart() {
        this.verifyCanStart(null);
    }

    @Override
    public void verifyCanStart(final Set<ControllerServiceNode> ignoredReferences) {
        final ScheduledState currentState = getPhysicalScheduledState();
        if (currentState != ScheduledState.STOPPED && currentState != ScheduledState.DISABLED) {
            throw new IllegalStateException(this.getIdentifier() + " cannot be started because it is not stopped. Current state is " + currentState.name());
        }

        verifyNoActiveThreads();

        switch (getValidationStatus()) {
            case VALID:
                return;
            case VALIDATING:
                throw new IllegalStateException("Processor with ID " + getIdentifier() + " cannot be started because its validation is still being performed");
        }

        final Collection<ValidationResult> validationErrors = getValidationErrors(ignoredReferences);
        if (ignoredReferences != null && !validationErrors.isEmpty()) {
            throw new IllegalStateException("Processor with ID " + getIdentifier() + " cannot be started because it is not currently valid");
        }
    }

    @Override
    public void verifyCanStop() {
        if (getScheduledState() != ScheduledState.RUNNING) {
            throw new IllegalStateException(this.getIdentifier() + " is not scheduled to run");
        }
    }

    @Override
    public void verifyCanUpdate() {
        if (isRunning()) {
            throw new IllegalStateException(this.getIdentifier() + " is not stopped");
        }
    }

    @Override
    public void verifyCanEnable() {
        if (getScheduledState() != ScheduledState.DISABLED) {
            throw new IllegalStateException(this.getIdentifier() + " is not disabled");
        }

        verifyNoActiveThreads();
    }

    @Override
    public void verifyCanDisable() {
        if (getScheduledState() != ScheduledState.STOPPED) {
            throw new IllegalStateException(this.getIdentifier() + " is not stopped");
        }
        verifyNoActiveThreads();
    }

    @Override
    public void verifyCanClearState() throws IllegalStateException {
        verifyCanUpdate();
    }

    private void verifyNoActiveThreads() throws IllegalStateException {
        if (hasActiveThreads) {
            final int threadCount = getActiveThreadCount();
            if (threadCount > 0) {
                throw new IllegalStateException(this.getIdentifier() + " has " + threadCount + " threads still active");
            }
        }
    }

    @Override
    public void verifyModifiable() throws IllegalStateException {
        if (isRunning()) {
            throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
        }
    }

    @Override
    public void enable() {
        desiredState = ScheduledState.STOPPED;
        final boolean updated = scheduledState.compareAndSet(ScheduledState.DISABLED, ScheduledState.STOPPED);

        if (updated) {
            LOG.info("{} enabled so ScheduledState transitioned from DISABLED to STOPPED", this);
        } else {
            LOG.info("{} enabled but not currently DISABLED so set desired state to STOPPED; current state is {}", this, scheduledState.get());
        }
    }

    @Override
    public void disable() {
        desiredState = ScheduledState.DISABLED;
        final boolean updated = scheduledState.compareAndSet(ScheduledState.STOPPED, ScheduledState.DISABLED);

        if (updated) {
            LOG.info("{} disabled so ScheduledState transitioned from STOPPED to DISABLED", this);
        } else {
            LOG.info("{} disabled but not currently STOPPED so set desired state to DISABLED; current state is {}", this, scheduledState.get());
        }
    }


    /**
     * Will idempotently start the processor using the following sequence: <i>
     * <ul>
     * <li>Validate Processor's state (e.g., PropertyDescriptors,
     * ControllerServices etc.)</li>
     * <li>Transition (atomically) Processor's scheduled state form STOPPED to
     * STARTING. If the above state transition succeeds, then execute the start
     * task (asynchronously) which will be re-tried until @OnScheduled is
     * executed successfully and "schedulingAgentCallback' is invoked, or until
     * STOP operation is initiated on this processor. If state transition fails
     * it means processor is already being started and WARN message will be
     * logged explaining it.</li>
     * </ul>
     * </i>
     * <p>
     * Any exception thrown while invoking operations annotated with @OnSchedule
     * will be caught and logged after which @OnUnscheduled operation will be
     * invoked (quietly) and the start sequence will be repeated (re-try) after
     * delay provided by 'administrativeYieldMillis'.
     * </p>
     * <p>
     * Upon successful completion of start sequence (@OnScheduled -&gt;
     * 'schedulingAgentCallback') the attempt will be made to transition
     * processor's scheduling state to RUNNING at which point processor is
     * considered to be fully started and functioning. If upon successful
     * invocation of @OnScheduled operation the processor can not be
     * transitioned to RUNNING state (e.g., STOP operation was invoked on the
     * processor while it's @OnScheduled operation was executing), the
     * processor's @OnUnscheduled operation will be invoked and its scheduling
     * state will be set to STOPPED at which point the processor is considered
     * to be fully stopped.
     * </p>
     */
    @Override
    public void start(final ScheduledExecutorService taskScheduler, final long administrativeYieldMillis, final long timeoutMillis, final Supplier<ProcessContext> processContextFactory,
            final SchedulingAgentCallback schedulingAgentCallback, final boolean failIfStopping) {

        run(taskScheduler, administrativeYieldMillis, timeoutMillis, processContextFactory, schedulingAgentCallback, failIfStopping, ScheduledState.RUNNING, ScheduledState.STARTING);
    }

    /**
     * Similar to {@link #start(ScheduledExecutorService, long, long, Supplier, SchedulingAgentCallback, boolean)}, except for the following:
     * <ul>
     *     <li>
     *         Once the {@link Processor#onTrigger(ProcessContext, ProcessSessionFactory)} method has been invoked successfully, the processor is scehduled to be stopped immediately.
     *         All appropriate lifecycle methods will be executed as well.
     *     </li>
     *     <li>
     *         The processor's desired state is going to be set to STOPPED right away. This usually doesn't prevent the processor to run once, unless NiFi is restarted before it can finish.
     *         In that case the processor will stay STOPPED after the restart.
     *     </li>
     * </ul>
     */
    @Override
    public void runOnce(final ScheduledExecutorService taskScheduler, final long administrativeYieldMillis, final long timeoutMillis, final Supplier<ProcessContext> processContextFactory,
                        final SchedulingAgentCallback schedulingAgentCallback) {

        run(taskScheduler, administrativeYieldMillis, timeoutMillis, processContextFactory, schedulingAgentCallback, true, ScheduledState.RUN_ONCE, ScheduledState.RUN_ONCE);
    }

    private void run(ScheduledExecutorService taskScheduler, long administrativeYieldMillis, long timeoutMillis, Supplier<ProcessContext> processContextFactory,
                     SchedulingAgentCallback schedulingAgentCallback, boolean failIfStopping, ScheduledState desiredState, ScheduledState scheduledState) {

        final Processor processor = processorRef.get().getProcessor();
        final ComponentLog procLog = new SimpleProcessLogger(StandardProcessorNode.this.getIdentifier(), processor);
        LOG.info("Starting {}", this);

        ScheduledState currentState;
        boolean starting;
        synchronized (this) {
            currentState = this.scheduledState.get();

            if (currentState == ScheduledState.STOPPED) {
                starting = this.scheduledState.compareAndSet(ScheduledState.STOPPED, scheduledState);
                if (starting) {
                    this.desiredState = desiredState;
                }
            } else if (currentState == ScheduledState.STOPPING && !failIfStopping) {
                this.desiredState = desiredState;
                return;
            } else {
                starting = false;
            }
        }

        if (starting) { // will ensure that the Processor represented by this node can only be started once
            initiateStart(taskScheduler, administrativeYieldMillis, timeoutMillis, processContextFactory, schedulingAgentCallback);
        } else {
            final String procName = processorRef.get().toString();
            LOG.warn("Cannot start {} because it is not currently stopped. Current state is {}", procName, currentState);
            procLog.warn("Cannot start {} because it is not currently stopped. Current state is {}", new Object[]{procName, currentState});
        }
    }

    private synchronized void activateThread() {
        final Thread thread = Thread.currentThread();
        final Long timestamp = System.currentTimeMillis();
        activeThreads.put(thread, new ActiveTask(timestamp));
    }

    private synchronized void deactivateThread() {
        activeThreads.remove(Thread.currentThread());
    }

    @Override
    public synchronized List<ActiveThreadInfo> getActiveThreads(final ThreadDetails threadDetails) {
        final long now = System.currentTimeMillis();

        final Map<Long, ThreadInfo> threadInfoMap = Stream.of(threadDetails.getThreadInfos())
            .collect(Collectors.toMap(ThreadInfo::getThreadId, Function.identity(), (a, b) -> a));

        final List<ActiveThreadInfo> threadList = new ArrayList<>(activeThreads.size());
        for (final Map.Entry<Thread, ActiveTask> entry : activeThreads.entrySet()) {
            final Thread thread = entry.getKey();
            final ActiveTask activeTask = entry.getValue();
            final Long timestamp = activeTask.getStartTime();
            final long activeMillis = now - timestamp;
            final ThreadInfo threadInfo = threadInfoMap.get(thread.getId());

            final String stackTrace = ThreadUtils.createStackTrace(thread, threadInfo, threadDetails.getDeadlockedThreadIds(), threadDetails.getMonitorDeadlockThreadIds(), activeMillis);

            final ActiveThreadInfo activeThreadInfo = new ActiveThreadInfo(thread.getName(), stackTrace, activeMillis, activeTask.isTerminated());
            threadList.add(activeThreadInfo);
        }

        return threadList;
    }

    @Override
    public synchronized int getTerminatedThreadCount() {
        int count = 0;
        for (final ActiveTask task : activeThreads.values()) {
            if (task.isTerminated()) {
                count++;
            }
        }

        return count;
    }


    @Override
    public int terminate() {
        verifyCanTerminate();

        int count = 0;
        for (final Map.Entry<Thread, ActiveTask> entry : activeThreads.entrySet()) {
            final Thread thread = entry.getKey();
            final ActiveTask activeTask = entry.getValue();

            if (!activeTask.isTerminated()) {
                activeTask.terminate();

                thread.setName(thread.getName() + " <Terminated Task>");
                count++;
            }

            thread.interrupt();
        }

        getLogger().terminate();
        scheduledState.set(ScheduledState.STOPPED);
        hasActiveThreads = false;

        return count;
    }

    @Override
    public boolean isTerminated(final Thread thread) {
        final ActiveTask activeTask = activeThreads.get(thread);
        if (activeTask == null) {
            return false;
        }

        return activeTask.isTerminated();
    }

    @Override
    public void verifyCanTerminate() {
        if (getScheduledState() != ScheduledState.STOPPED && getScheduledState() != ScheduledState.RUN_ONCE) {
            throw new IllegalStateException("Processor is not stopped");
        }
    }


    private void initiateStart(final ScheduledExecutorService taskScheduler, final long administrativeYieldMillis, final long timeoutMilis,
            final Supplier<ProcessContext> processContextFactory, final SchedulingAgentCallback schedulingAgentCallback) {

        final Processor processor = getProcessor();
        final ComponentLog procLog = new SimpleProcessLogger(StandardProcessorNode.this.getIdentifier(), processor);

        // Completion Timestamp is set to MAX_VALUE because we don't want to timeout until the task has a chance to run.
        final AtomicLong completionTimestampRef = new AtomicLong(Long.MAX_VALUE);

        // Create a task to invoke the @OnScheduled annotation of the processor
        final Callable<Void> startupTask = () -> {
            final ProcessContext processContext = processContextFactory.get();

            final ScheduledState currentScheduleState = scheduledState.get();
            if (currentScheduleState == ScheduledState.STOPPING || currentScheduleState == ScheduledState.STOPPED || getDesiredState() == ScheduledState.STOPPED) {
                LOG.debug("{} is stopped. Will not call @OnScheduled lifecycle methods or begin trigger onTrigger() method", StandardProcessorNode.this);
                schedulingAgentCallback.onTaskComplete();
                scheduledState.set(ScheduledState.STOPPED);
                return null;
            }

            final ValidationStatus validationStatus = getValidationStatus();
            if (validationStatus != ValidationStatus.VALID) {
                LOG.debug("Cannot start {} because Processor is currently not valid; will try again after 5 seconds", StandardProcessorNode.this);

                // re-initiate the entire process
                final Runnable initiateStartTask = () -> initiateStart(taskScheduler, administrativeYieldMillis, timeoutMilis, processContextFactory, schedulingAgentCallback);
                taskScheduler.schedule(initiateStartTask, 5, TimeUnit.SECONDS);

                schedulingAgentCallback.onTaskComplete();
                return null;
            }

            LOG.debug("Invoking @OnScheduled methods of {}", processor);

            // Now that the task has been scheduled, set the timeout
            completionTimestampRef.set(System.currentTimeMillis() + timeoutMilis);

            try (final NarCloseable nc = NarCloseable.withComponentNarLoader(getExtensionManager(), processor.getClass(), processor.getIdentifier())) {
                try {
                    hasActiveThreads = true;

                    activateThread();
                    try {
                        ReflectionUtils.invokeMethodsWithAnnotation(OnScheduled.class, processor, processContext);
                    } finally {
                        deactivateThread();
                    }

                    if (
                        (desiredState == ScheduledState.RUNNING && scheduledState.compareAndSet(ScheduledState.STARTING, ScheduledState.RUNNING))
                            || (desiredState == ScheduledState.RUN_ONCE && scheduledState.compareAndSet(ScheduledState.RUN_ONCE, ScheduledState.RUN_ONCE))
                    ) {
                        LOG.debug("Successfully completed the @OnScheduled methods of {}; will now start triggering processor to run", processor);
                        schedulingAgentCallback.trigger(); // callback provided by StandardProcessScheduler to essentially initiate component's onTrigger() cycle
                    } else {
                        LOG.info("Successfully invoked @OnScheduled methods of {} but scheduled state is no longer STARTING so will stop processor now; current state = {}, desired state = {}",
                            processor, scheduledState.get(), desiredState);

                        // can only happen if stopProcessor was called before service was transitioned to RUNNING state
                        activateThread();
                        try {
                            ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext);
                            ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnStopped.class, processor, processContext);
                            hasActiveThreads = false;
                        } finally {
                            deactivateThread();
                        }

                        scheduledState.set(ScheduledState.STOPPED);

                        if (desiredState == ScheduledState.DISABLED) {
                            final boolean disabled = scheduledState.compareAndSet(ScheduledState.STOPPED, ScheduledState.DISABLED);
                            if (disabled) {
                                LOG.info("After stopping {}, determined that Desired State is DISABLED so disabled processor", processor);
                            }
                        }
                    }
                } finally {
                    schedulingAgentCallback.onTaskComplete();
                }
            } catch (Exception e) {
                final Throwable cause = (e instanceof InvocationTargetException) ? e.getCause() : e;
                procLog.error("Failed to properly initialize Processor. If still scheduled to run, NiFi will attempt to "
                    + "initialize and run the Processor again after the 'Administrative Yield Duration' has elapsed. Failure is due to " + cause, cause);

                // If processor's task completed Exceptionally, then we want to retry initiating the start (if Processor is still scheduled to run).
                try (final NarCloseable nc = NarCloseable.withComponentNarLoader(getExtensionManager(), processor.getClass(), processor.getIdentifier())) {
                    activateThread();
                    try {
                        ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext);
                        ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnStopped.class, processor, processContext);
                        hasActiveThreads = false;
                    } finally {
                        deactivateThread();
                    }
                }

                // make sure we only continue retry loop if STOP action wasn't initiated
                if (scheduledState.get() != ScheduledState.STOPPING && scheduledState.get() != ScheduledState.RUN_ONCE) {
                    // re-initiate the entire process
                    final Runnable initiateStartTask = () -> initiateStart(taskScheduler, administrativeYieldMillis, timeoutMilis, processContextFactory, schedulingAgentCallback);
                    taskScheduler.schedule(initiateStartTask, administrativeYieldMillis, TimeUnit.MILLISECONDS);
                } else {
                    scheduledState.set(ScheduledState.STOPPED);
                }
            }

            return null;
        };

        // Trigger the task in a background thread.
        final Future<?> taskFuture = schedulingAgentCallback.scheduleTask(startupTask);

        // Trigger a task periodically to check if @OnScheduled task completed. Once it has,
        // this task will call SchedulingAgentCallback#onTaskComplete.
        // However, if the task times out, we need to be able to cancel the monitoring. So, in order
        // to do this, we use #scheduleWithFixedDelay and then make that Future available to the task
        // itself by placing it into an AtomicReference.
        final AtomicReference<Future<?>> futureRef = new AtomicReference<>();
        final Runnable monitoringTask = new Runnable() {
            @Override
            public void run() {
                Future<?> monitoringFuture = futureRef.get();
                if (monitoringFuture == null) { // Future is not yet available. Just return and wait for the next invocation.
                    return;
                }

               monitorAsyncTask(taskFuture, monitoringFuture, completionTimestampRef.get());
            }
        };

        final Future<?> future = taskScheduler.scheduleWithFixedDelay(monitoringTask, 1, 10, TimeUnit.MILLISECONDS);
        futureRef.set(future);
    }

    /**
     * Will idempotently stop the processor using the following sequence: <i>
     * <ul>
     * <li>Transition (atomically) Processor's scheduled state from RUNNING to
     * STOPPING. If the above state transition succeeds, then invoke any method
     * on the Processor with the {@link OnUnscheduled} annotation. Once those methods
     * have been called and returned (either normally or exceptionally), start checking
     * to see if all of the Processor's active threads have finished. If not, check again
     * every 100 milliseconds until they have.
     * Once all after threads have completed, the processor's @OnStopped operation will be invoked
     * and its scheduled state is set to STOPPED which completes processor stop
     * sequence.</li>
     * </ul>
     * </i>
     *
     * <p>
     * If for some reason processor's scheduled state can not be transitioned to
     * STOPPING (e.g., the processor didn't finish @OnScheduled operation when
     * stop was called), the attempt will be made to transition processor's
     * scheduled state from STARTING to STOPPING which will allow
     * {@link #start(ScheduledExecutorService, long, long, Supplier, SchedulingAgentCallback, boolean)}
     * method to initiate processor's shutdown upon exiting @OnScheduled
     * operation, otherwise the processor's scheduled state will remain
     * unchanged ensuring that multiple calls to this method are idempotent.
     * </p>
     */
    @Override
    public CompletableFuture<Void> stop(final ProcessScheduler processScheduler, final ScheduledExecutorService executor, final ProcessContext processContext,
            final SchedulingAgent schedulingAgent, final LifecycleState scheduleState) {

        final Processor processor = processorRef.get().getProcessor();
        LOG.info("Stopping processor: " + this);
        desiredState = ScheduledState.STOPPED;

        final CompletableFuture<Void> future = new CompletableFuture<>();
        // will ensure that the Processor represented by this node can only be stopped once
        if (this.scheduledState.compareAndSet(ScheduledState.RUNNING, ScheduledState.STOPPING) || this.scheduledState.compareAndSet(ScheduledState.RUN_ONCE, ScheduledState.STOPPING)) {
            scheduleState.incrementActiveThreadCount(null);

            // will continue to monitor active threads, invoking OnStopped once there are no
            // active threads (with the exception of the thread performing shutdown operations)
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        if (scheduleState.isScheduled()) {
                            schedulingAgent.unschedule(StandardProcessorNode.this, scheduleState);

                            activateThread();
                            try (final NarCloseable nc = NarCloseable.withComponentNarLoader(getExtensionManager(), processor.getClass(), processor.getIdentifier())) {
                                ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext);
                            } finally {
                                deactivateThread();
                            }
                        }

                        // all threads are complete if the active thread count is 1. This is because this thread that is
                        // performing the lifecycle actions counts as 1 thread.
                        final boolean allThreadsComplete = scheduleState.getActiveThreadCount() == 1;
                        if (allThreadsComplete) {
                            activateThread();
                            try (final NarCloseable nc = NarCloseable.withComponentNarLoader(getExtensionManager(), processor.getClass(), processor.getIdentifier())) {
                                ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnStopped.class, processor, processContext);
                            } finally {
                                deactivateThread();
                            }

                            scheduleState.decrementActiveThreadCount(null);
                            hasActiveThreads = false;
                            scheduledState.set(ScheduledState.STOPPED);
                            future.complete(null);

                            // This can happen only when we join a cluster. In such a case, we can inherit a flow from the cluster that says that
                            // the Processor is to be running. However, if the Processor is already in the process of stopping, we cannot immediately
                            // start running the Processor. As a result, we check here, since the Processor is stopped, and then immediately start the
                            // Processor if need be.
                            final ScheduledState desired = StandardProcessorNode.this.desiredState;
                            if (desired == ScheduledState.RUNNING) {
                                LOG.info("Finished stopping {} but desired state is now RUNNING so will start processor", this);
                                processScheduler.startProcessor(StandardProcessorNode.this, true);
                            } else if (desired == ScheduledState.DISABLED) {
                                final boolean updated = scheduledState.compareAndSet(ScheduledState.STOPPED, ScheduledState.DISABLED);

                                if (updated) {
                                    LOG.info("Finished stopping {} but desired state is now DISABLED so disabled processor", this);
                                } else {
                                    LOG.info("Finished stopping {} but desired state is now DISABLED. Scheduled State could not be transitioned from STOPPED to DISABLED, "
                                        + "though, so will allow the other thread to finish state transition. Current state is {}", this, scheduledState.get());
                                }
                            }
                        } else {
                            // Not all of the active threads have finished. Try again in 100 milliseconds.
                            executor.schedule(this, 100, TimeUnit.MILLISECONDS);
                        }
                    } catch (final Exception e) {
                        LOG.warn("Failed while shutting down processor " + processor, e);
                    }
                }
            });
        } else {
            // We do compareAndSet() instead of set() to ensure that Processor
            // stoppage is handled consistently including a condition where
            // Processor never got a chance to transition to RUNNING state
            // before stop() was called. If that happens the stop processor
            // routine will be initiated in start() method, otherwise the IF
            // part will handle the stop processor routine.
            final boolean updated = this.scheduledState.compareAndSet(ScheduledState.STARTING, ScheduledState.STOPPING);
            if (updated) {
                LOG.debug("Transitioned state of {} from STARTING to STOPPING", this);
            }

            future.complete(null);
        }

        return future;
    }

    @Override
    public ScheduledState getDesiredState() {
        return desiredState;
    }

    private void monitorAsyncTask(final Future<?> taskFuture, final Future<?> monitoringFuture, final long completionTimestamp) {
        if (taskFuture.isDone()) {
            monitoringFuture.cancel(false); // stop scheduling this task
        } else if (System.currentTimeMillis() > completionTimestamp) {
            // Task timed out. Request an interrupt of the processor task
            taskFuture.cancel(true);

            // Stop monitoring the processor. We have interrupted the thread so that's all we can do. If the processor responds to the interrupt, then
            // it will be re-scheduled. If it does not, then it will either keep the thread indefinitely or eventually finish, at which point
            // the Processor will begin running.
            monitoringFuture.cancel(false);

            final Processor processor = processorRef.get().getProcessor();
            LOG.warn("Timed out while waiting for OnScheduled of "
                + processor + " to finish. An attempt is made to cancel the task via Thread.interrupt(). However it does not "
                + "guarantee that the task will be canceled since the code inside current OnScheduled operation may "
                + "have been written to ignore interrupts which may result in a runaway thread. This could lead to more issues, "
                + "eventually requiring NiFi to be restarted. This is usually a bug in the target Processor '"
                + processor + "' that needs to be documented, reported and eventually fixed.");
        }
    }

    @Override
    public String getProcessGroupIdentifier() {
        final ProcessGroup group = getProcessGroup();
        return group == null ? null : group.getIdentifier();
    }

    @Override
    public Optional<String> getVersionedComponentId() {
        return Optional.ofNullable(versionedComponentId.get());
    }

    @Override
    public void setVersionedComponentId(final String versionedComponentId) {
        boolean updated = false;
        while (!updated) {
            final String currentId = this.versionedComponentId.get();

            if (currentId == null) {
                updated = this.versionedComponentId.compareAndSet(null, versionedComponentId);
            } else if (currentId.equals(versionedComponentId)) {
                return;
            } else if (versionedComponentId == null) {
                updated = this.versionedComponentId.compareAndSet(currentId, null);
            } else {
                throw new IllegalStateException(this + " is already under version control");
            }
        }
    }
}