NIFI-16283 Support changing Connector versions

nifi-docs/src/main/asciidoc/toolkit-guide.adoc

@@ -142,6 +142,7 @@ The following are available commands:
  nifi get-controller-configuration
  nifi update-controller-configuration
  nifi change-version-processor
+ nifi change-version-connector
  nifi upload-nar
  nifi download-nar
  nifi delete-nar
@@ -313,7 +314,7 @@ For example, typing tab at an empty prompt should display possible commands for
 Typing "nifi " and then a tab will show the sub-commands for NiFi:
 
  #> nifi
- change-version-processor          delete-flow-analysis-rule         export-reporting-task             get-policy                        list-user-groups                  pg-export                         pg-delete
+ change-version-connector          change-version-processor          delete-flow-analysis-rule         export-reporting-task             get-policy                        list-user-groups                  pg-export                         pg-delete
  cluster-summary                   delete-node                       export-reporting-tasks            get-reg-client-id                 list-users                        pg-get-all-versions               pg-stop-version-control
  connect-node                      delete-param                      fetch-params                      get-reporting-task                logout-access-token               pg-get-param-context              set-inherited-param-contexts
  create-flow-analysis-rule         delete-param-context              get-access-token                  get-reporting-tasks               merge-param-context               pg-get-services                   remove-inherited-param-contexts

nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ConnectorEntityMerger.java

@@ -101,6 +101,14 @@ private static void mergeDtos(final ConnectorDTO clientDto, final Map<NodeIdenti
             return;
         }
 
+        for (final ConnectorDTO nodeConnector : dtoMap.values()) {
+            if (nodeConnector != null) {
+                if (clientDto.getMultipleVersionsAvailable() == null || !Boolean.TRUE.equals(nodeConnector.getMultipleVersionsAvailable())) {
+                    clientDto.setMultipleVersionsAvailable(Boolean.FALSE);
+                }
+            }
+        }
+
         mergeState(clientDto, dtoMap);
 
         // Merge configuration steps to handle dynamic property descriptors

nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/manager/ConnectorEntityMergerTest.java

@@ -76,6 +76,31 @@ void testMergeConnectorStateWithNull() {
         assertEquals("RUNNING", clientEntity.getComponent().getState());
     }
 
+    @Test
+    void testMergeMultipleVersionsAvailable() {
+        final ConnectorEntity clientEntity = createConnectorEntity("connector1", "STOPPED");
+        final ConnectorEntity nodeEntity = createConnectorEntity("connector1", "STOPPED");
+        clientEntity.getComponent().setMultipleVersionsAvailable(true);
+
+        final Map<NodeIdentifier, ConnectorEntity> entityMap = new HashMap<>();
+        entityMap.put(getNodeIdentifier("client", 8000), clientEntity);
+        entityMap.put(getNodeIdentifier("node", 8001), nodeEntity);
+
+        nodeEntity.getComponent().setMultipleVersionsAvailable(false);
+        ConnectorEntityMerger.merge(clientEntity, entityMap);
+        assertFalse(clientEntity.getComponent().getMultipleVersionsAvailable());
+
+        clientEntity.getComponent().setMultipleVersionsAvailable(true);
+        nodeEntity.getComponent().setMultipleVersionsAvailable(null);
+        ConnectorEntityMerger.merge(clientEntity, entityMap);
+        assertFalse(clientEntity.getComponent().getMultipleVersionsAvailable());
+
+        clientEntity.getComponent().setMultipleVersionsAvailable(true);
+        nodeEntity.getComponent().setMultipleVersionsAvailable(true);
+        ConnectorEntityMerger.merge(clientEntity, entityMap);
+        assertTrue(clientEntity.getComponent().getMultipleVersionsAvailable());
+    }
+
     @Test
     void testMergeConfigurationStepsWithDifferentAllowableValues() {
         final ConnectorEntity clientEntity = createConnectorEntityWithConfig(

nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorNode.java

@@ -58,6 +58,12 @@ public interface ConnectorNode extends ComponentAuthorizable, VersionedComponent
 
     void verifyCanStart();
 
+    void verifyCanReload();
+
+    void verifyCanUpdateBundle(BundleCoordinate bundleCoordinate);
+
+    void replaceConnector(Connector connector, BundleCoordinate bundleCoordinate, ComponentLog componentLog) throws FlowUpdateException;
+
     Connector getConnector();
 
     /**
@@ -99,6 +105,8 @@ default void setCustomLoggingAttributes(final Map<String, String> attributes) {
      */
     boolean isExtensionMissing();
 
+    void resetValidationState();
+
     List<DescribedValue> fetchAllowableValues(String stepName, String propertyName);
 
     List<DescribedValue> fetchAllowableValues(String stepName, String propertyName, String filter);

nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/FrameworkFlowContext.java

@@ -19,9 +19,11 @@
 
 import org.apache.nifi.asset.AssetManager;
 import org.apache.nifi.components.connector.components.FlowContext;
+import org.apache.nifi.flow.Bundle;
 import org.apache.nifi.flow.VersionedExternalFlow;
 import org.apache.nifi.flow.VersionedProcessGroup;
 import org.apache.nifi.groups.ProcessGroup;
+import org.apache.nifi.logging.ComponentLog;
 
 public interface FrameworkFlowContext extends FlowContext {
     ProcessGroup getManagedProcessGroup();
@@ -53,4 +55,12 @@ public interface FrameworkFlowContext extends FlowContext {
      *                                    persisted while the Connector was in Troubleshooting mode
      */
     void restoreTroubleshootingFlow(VersionedProcessGroup troubleshootingProcessGroup);
+
+    default void reload(final Bundle bundle, final ComponentLog connectorLog) {
+        reload(bundle, connectorLog, getConfigurationContext());
+    }
+
+    default void reload(final Bundle bundle, final ComponentLog connectorLog, final MutableConnectorConfigurationContext configurationContext) {
+        throw new UnsupportedOperationException();
+    }
 }

nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/MutableConnectorConfigurationContext.java

@@ -49,6 +49,13 @@ public interface MutableConnectorConfigurationContext extends ConnectorConfigura
      */
     ConfigurationUpdateResult replaceProperties(String stepName, StepConfiguration configuration);
 
+    /**
+     * Removes the named configuration step and its resolved values. If the step is not present, this is a no-op.
+     *
+     * @param stepName the name of the configuration step to remove
+     */
+    void removeStep(String stepName);
+
     /**
      * Resolves all existing property values.
      */

nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ReloadComponent.java

@@ -17,6 +17,8 @@
 package org.apache.nifi.controller;
 
 import org.apache.nifi.bundle.BundleCoordinate;
+import org.apache.nifi.components.connector.ConnectorNode;
+import org.apache.nifi.controller.exception.ConnectorInstantiationException;
 import org.apache.nifi.controller.exception.ControllerServiceInstantiationException;
 import org.apache.nifi.controller.exception.ProcessorInstantiationException;
 import org.apache.nifi.controller.flowanalysis.FlowAnalysisRuleInstantiationException;
@@ -104,4 +106,14 @@ void reload(ParameterProviderNode existingNode, String newType, BundleCoordinate
      */
     void reload(FlowRegistryClientNode existingNode, String newType, BundleCoordinate bundleCoordinate, Set<URL> additionalUrls)
             throws FlowRepositoryClientInstantiationException;
+
+    /**
+     * Changes the underlying Connector held by the node to an instance of the new type.
+     *
+     * @param existingNode the ConnectorNode being updated
+     * @param newType the fully qualified class name of the new type
+     * @param bundleCoordinate the bundle coordinate of the new type
+     * @throws ConnectorInstantiationException if unable to create an instance of the new type
+     */
+    void reload(ConnectorNode existingNode, String newType, BundleCoordinate bundleCoordinate) throws ConnectorInstantiationException;
 }

nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/exception/ConnectorInstantiationException.java

@@ -0,0 +1,30 @@
+/*
+ * 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.exception;
+
+public class ConnectorInstantiationException extends Exception {
+
+    private static final long serialVersionUID = 1L;
+
+    public ConnectorInstantiationException(final String message) {
+        super(message);
+    }
+
+    public ConnectorInstantiationException(final String message, final Throwable cause) {
+        super(message, cause);
+    }
+}

nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorConfigurationContext.java

@@ -258,6 +258,17 @@ public ConfigurationUpdateResult replaceProperties(final String stepName, final
         }
     }
 
+    @Override
+    public void removeStep(final String stepName) {
+        writeLock.lock();
+        try {
+            propertyConfigurations.remove(stepName);
+            resolvedPropertyConfigurations.remove(stepName);
+        } finally {
+            writeLock.unlock();
+        }
+    }
+
     @Override
     public void resolvePropertyValues() {
         writeLock.lock();