This is an automated email from the ASF dual-hosted git repository.
markap14 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-api.git
The following commit(s) were added to refs/heads/main by this push:
new d7d0a74 NIFI-16470: Add a flow evolution hook and persist the managed
flow. (#111)
d7d0a74 is described below
commit d7d0a743a9d763182eada2764a4427007adca553
Author: Bob Paulin <[email protected]>
AuthorDate: Fri Oct 16 15:53:01 2026 -0500
NIFI-16470: Add a flow evolution hook and persist the managed flow. (#111)
prepareForFlowEvolution runs when the Connector version changes, while
draining. The managed group, its bundle, and its parameter context are
persisted.
---
.../nifi/components/connector/Connector.java | 20 +++++++++--
.../components/connector/FlowEvolutionContext.java | 37 +++++++++++++++++++
.../org/apache/nifi/flow/VersionedConnector.java | 42 ++++++++++++++++++----
3 files changed, 89 insertions(+), 10 deletions(-)
diff --git a/src/main/java/org/apache/nifi/components/connector/Connector.java
b/src/main/java/org/apache/nifi/components/connector/Connector.java
index c1f696a..853db0d 100644
--- a/src/main/java/org/apache/nifi/components/connector/Connector.java
+++ b/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.
diff --git
a/src/main/java/org/apache/nifi/components/connector/FlowEvolutionContext.java
b/src/main/java/org/apache/nifi/components/connector/FlowEvolutionContext.java
new file mode 100644
index 0000000..05006f3
--- /dev/null
+++
b/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();
+}
diff --git a/src/main/java/org/apache/nifi/flow/VersionedConnector.java
b/src/main/java/org/apache/nifi/flow/VersionedConnector.java
index 064bd58..2629fee 100644
--- a/src/main/java/org/apache/nifi/flow/VersionedConnector.java
+++ b/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 class VersionedConnector {
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 class VersionedConnector {
}
/**
- * 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 class VersionedConnector {
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;
+ }
}
