class StandardFlowService implements FlowService, ProtocolHandler {

    @Override
    public void load(final DataFlow dataFlow) throws IOException, FlowSerializationException, FlowSynchronizationException, UninheritableFlowException, MissingBundleException {
        if (configuredForClustering) {
            // Create the initial flow from disk if it exists, or from serializing the empty root group in flow controller
            final DataFlow initialFlow = (dataFlow == null) ? createDataFlow() : dataFlow;
            if (logger.isTraceEnabled()) {
                logger.trace("InitialFlow = " + new String(initialFlow.getFlow(), StandardCharsets.UTF_8));
            }

            // Sync the initial flow into the flow controller so that if the flow came from disk we loaded the
            // whole flow into the flow controller and applied any bundle upgrades
            writeLock.lock();
            try {
                loadFromBytes(initialFlow, true);
            } finally {
                writeLock.unlock();
            }

            // Get the proposed flow by serializing the flow controller which now has the synced version from above
            final DataFlow proposedFlow = createDataFlowFromController();
            if (logger.isTraceEnabled()) {
                logger.trace("ProposedFlow = " + new String(proposedFlow.getFlow(), StandardCharsets.UTF_8));
            }

            /*
             * Attempt to connect to the cluster. If the manager is able to
             * provide a data flow, then the manager will send a connection
             * response. If the manager was unable to be located, then
             * the response will be null and we should load the local dataflow
             * and heartbeat until a manager is located.
             */
            final boolean localFlowEmpty = StandardFlowSynchronizer.isEmpty(proposedFlow);
            final ConnectionResponse response = connect(true, localFlowEmpty, proposedFlow);

            // obtain write lock while we are updating the controller. We need to ensure that we don't
            // obtain the lock before calling connect(), though, or we will end up getting a deadlock
            // because the node that is receiving the connection request won't be able to get the current
            // flow, as that requires a read lock.
            writeLock.lock();
            try {
                if (response == null || response.shouldTryLater()) {
                    logger.info("Flow controller will load local dataflow and suspend connection handshake until a cluster connection response is received.");

                    // set node ID on controller before we start heartbeating because heartbeat needs node ID
                    controller.setNodeId(nodeId);
                    clusterCoordinator.setLocalNodeIdentifier(nodeId);

                    // set node as clustered, since it is trying to connect to a cluster
                    controller.setClustered(true, null);
                    clusterCoordinator.setConnected(false);

                    controller.setConnectionStatus(new NodeConnectionStatus(nodeId, DisconnectionCode.NOT_YET_CONNECTED));

                    /*
                     * Start heartbeating. Heartbeats will fail because we can't reach
                     * the manager, but when we locate the manager, the node will
                     * reconnect and establish a connection to the cluster. The
                     * heartbeat is the trigger that will cause the manager to
                     * issue a reconnect request.
                     */
                    controller.startHeartbeating();

                    // Initialize the controller after the flow is loaded so we don't take any actions on repos until everything is good
                    initializeController();

                    // notify controller that flow is initialized
                    try {
                        controller.onFlowInitialized(autoResumeState);
                    } catch (final Exception ex) {
                        logger.warn("Unable to start all processors due to invalid flow configuration.");
                        if (logger.isDebugEnabled()) {
                            logger.warn(StringUtils.EMPTY, ex);
                        }
                    }
                } else {
                    try {
                        loadFromConnectionResponse(response);
                    } catch (final Exception e) {
                        logger.error("Failed to load flow from cluster due to: " + e, e);
                        handleConnectionFailure(e);
                        throw new IOException(e);
                    }
                }

                // save the flow in the controller so we write out the latest flow with any updated bundles to disk
                dao.save(controller, true);

            } finally {
                writeLock.unlock();
            }
        } else {
            writeLock.lock();
            try {
                // operating in standalone mode, so load proposed flow and initialize the controller
                loadFromBytes(dataFlow, true);
                initializeController();
                dao.save(controller, true);
            } finally {
                writeLock.unlock();
            }
        }
    }

    private  StandardFlowService(
            final FlowController controller,
            final NiFiProperties nifiProperties,
            final NodeProtocolSenderListener senderListener,
            final StringEncryptor encryptor,
            final boolean configuredForClustering,
            final ClusterCoordinator clusterCoordinator,
            final RevisionManager revisionManager,
            final Authorizer authorizer);

}