NIFI-16470: Add a flow evolution hook and persist the managed flow.

src/main/java/org/apache/nifi/components/connector/Connector.java

@@ -93,11 +93,11 @@ public interface Connector {
      * </p>
      *
      * <p>
-     * This method must be idempotent and free of side effects.
+     * Build the flow from the given context's configuration, not from the live group. The method must be idempotent and
+     * must not change the flow it returns.
      * </p>
      *
-     * @param activeFlowContext the Connector's current Active Flow Context, used as the source of configuration
-     *                          from which the returned flow is built
+     * @param activeFlowContext the context whose configuration the flow is built from
      * @return the Connector's authoritative view of the flow for the given context
      */
     VersionedExternalFlow getActiveFlow(FlowContext activeFlowContext);
@@ -239,6 +239,20 @@ public interface Connector {
      */
     List<DescribedValue> fetchAllowableValues(String stepName, String propertyName, FlowContext flowContext, String filter);
 
+    /**
+     * Called when a change is detected in the Connector version, while draining and before the new flow replaces the
+     * Active flow. Return a Future that completes when preparation is finished; the framework may cancel it. The
+     * default implementation does nothing. Not called when the Connector leaves Troubleshooting; the managed flow must
+     * be empty then.
+     *
+     * @param context the pre-upgrade flow and the upgraded bundle
+     * @return a Future that completes when preparation is finished
+     * @throws FlowUpdateException if the pre-upgrade flow cannot be prepared
+     */
+    default CompletableFuture<Void> prepareForFlowEvolution(final FlowEvolutionContext context) throws FlowUpdateException {
+        return CompletableFuture.completedFuture(null);
+    }
+
     /**
      * Drains any in-flight FlowFiles from the flow associated with the given Flow Context by processing the existing data
      * but not accepting any new data.

src/main/java/org/apache/nifi/components/connector/FlowEvolutionContext.java

@@ -0,0 +1,37 @@
+/*
+ * 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.components.connector;
+
+import org.apache.nifi.components.connector.components.FlowContext;
+import org.apache.nifi.flow.Bundle;
+
+/**
+ * Context for {@link Connector#prepareForFlowEvolution(FlowEvolutionContext)} during an upgrade drain.
+ */
+public interface FlowEvolutionContext {
+
+    /**
+     * @return the pre-upgrade Active flow, with configuration already migrated
+     */
+    FlowContext getPreUpgradeActiveFlowContext();
+
+    /**
+     * @return the bundle that installs the upgraded flow
+     */
+    Bundle getPostUpgradeBundle();
+}

src/main/java/org/apache/nifi/flow/VersionedConnector.java

@@ -27,7 +27,9 @@ public class VersionedConnector {
     private List<VersionedConfigurationStep> workingFlowConfiguration;
     private String type;
     private Bundle bundle;
+    private Bundle managedFlowBundle;
     private VersionedProcessGroup managedProcessGroup;
+    private VersionedParameterContext managedParameterContext;
 
     public String getInstanceIdentifier() {
         return instanceIdentifier;
@@ -77,6 +79,11 @@ public void setType(final String type) {
         this.type = type;
     }
 
+    /**
+     * Returns the Connector bundle loaded on restart.
+     *
+     * @return the Connector code to load, or {@code null} if none was persisted
+     */
     public Bundle getBundle() {
         return bundle;
     }
@@ -86,14 +93,24 @@ public void setBundle(final Bundle bundle) {
     }
 
     /**
-     * Returns the contents of the Connector's Managed Process Group. This field is populated only when the Connector is
-     * in Troubleshooting mode at the time the flow is persisted, so that any user modifications made while in Troubleshooting
-     * survive a restart. For any other Connector state, this field is {@code null} and the Managed Process Group's contents
-     * are reconstructed by re-applying the configuration steps upon restore. If the Connector's scheduled state is not
-     * {@link VersionedConnectorState#TROUBLESHOOTING TROUBLESHOOTING}, this field is ignored.
+     * Returns the bundle that built {@link #getManagedProcessGroup()}. Differs from {@link #getBundle()} only while an
+     * upgrade is pending.
+     *
+     * @return the bundle that built the flow, or {@code null} if it was not persisted
+     */
+    public Bundle getManagedFlowBundle() {
+        return managedFlowBundle;
+    }
+
+    public void setManagedFlowBundle(final Bundle managedFlowBundle) {
+        this.managedFlowBundle = managedFlowBundle;
+    }
+
+    /**
+     * Returns the last applied flow. While the Connector is in Troubleshooting, this holds the user's edits so they
+     * survive a restart.
      *
-     * @return the persisted contents of the Managed Process Group, or {@code null} if the Connector was not in
-     *         Troubleshooting mode when the flow was persisted
+     * @return the last applied flow, or {@code null} if none was persisted
      */
     public VersionedProcessGroup getManagedProcessGroup() {
         return managedProcessGroup;
@@ -102,4 +119,15 @@ public VersionedProcessGroup getManagedProcessGroup() {
     public void setManagedProcessGroup(final VersionedProcessGroup managedProcessGroup) {
         this.managedProcessGroup = managedProcessGroup;
     }
+
+    /**
+     * @return the parameter context for the managed group, or {@code null} if none was persisted
+     */
+    public VersionedParameterContext getManagedParameterContext() {
+        return managedParameterContext;
+    }
+
+    public void setManagedParameterContext(final VersionedParameterContext managedParameterContext) {
+        this.managedParameterContext = managedParameterContext;
+    }
 }