From e993cfa768055e7e9c6c39f05a92144f1539b9d1 Mon Sep 17 00:00:00 2001 From: Bob Paulin Date: Thu, 8 Oct 2026 17:47:52 -0500 Subject: [PATCH] NIFI-16470: Add a flow evolution hook and persist the managed flow. 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 +++++++-- .../connector/FlowEvolutionContext.java | 37 ++++++++++++++++ .../apache/nifi/flow/VersionedConnector.java | 42 +++++++++++++++---- 3 files changed, 89 insertions(+), 10 deletions(-) create mode 100644 src/main/java/org/apache/nifi/components/connector/FlowEvolutionContext.java 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 { *

* *

- * 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. *

* - * @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 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 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 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; + } }