You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by mo...@apache.org on 2024/10/02 20:03:02 UTC
(nifi) branch support/nifi-1.x updated: NIFI-12016: Allow use of compatible NAR bundles when loading flow from cluster connection; when determining what bundles are compatible, consider not just any bundle if it's the only one but also any bundle whose version matches the framework version so that when NiFi is upgraded, it is handled more gracefully.
This is an automated email from the ASF dual-hosted git repository.
mosermw pushed a commit to branch support/nifi-1.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/support/nifi-1.x by this push:
new e720ccc309 NIFI-12016: Allow use of compatible NAR bundles when loading flow from cluster connection; when determining what bundles are compatible, consider not just any bundle if it's the only one but also any bundle whose version matches the framework version so that when NiFi is upgraded, it is handled more gracefully.
e720ccc309 is described below
commit e720ccc309b6e1a599f59ed5ec9caaaeaa876abe
Author: Mark Payne <ma...@hotmail.com>
AuthorDate: Wed Aug 30 17:39:31 2023 -0400
NIFI-12016: Allow use of compatible NAR bundles when loading flow from cluster connection; when determining what bundles are compatible, consider not just any bundle if it's the only one but also any bundle whose version matches the framework version so that when NiFi is upgraded, it is handled more gracefully.
NIFI-12016: Removed system tests that are no longer relevant
Co-authored-by: Matt Burgess <ma...@apache.org>
Changes for support/nifi-1.x
This closes #9205.
Signed-off-by: Mike Moser <mo...@apache.org>
---
.../java/org/apache/nifi/util/BundleUtils.java | 38 ++++++--
.../java/org/apache/nifi/util/TestBundleUtils.java | 100 +++++++++++++++++++
.../nifi/controller/StandardFlowService.java | 2 +-
.../system/clustering/FlowSynchronizationIT.java | 108 ---------------------
4 files changed, 129 insertions(+), 119 deletions(-)
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/util/BundleUtils.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/util/BundleUtils.java
index dfd0da346f..17e128ebf4 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/util/BundleUtils.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/util/BundleUtils.java
@@ -20,6 +20,7 @@ import org.apache.nifi.bundle.Bundle;
import org.apache.nifi.bundle.BundleCoordinate;
import org.apache.nifi.flow.VersionedProcessGroup;
import org.apache.nifi.nar.ExtensionManager;
+import org.apache.nifi.nar.NarClassLoadersHolder;
import org.apache.nifi.web.api.dto.BundleDTO;
import java.util.List;
@@ -30,11 +31,29 @@ import java.util.stream.Collectors;
* Utility class for Bundles.
*/
public final class BundleUtils {
- private static Optional<BundleCoordinate> findOptionalBundleForType(final ExtensionManager extensionManager, final String type, final BundleCoordinate desiredCoordinate) {
+ static Optional<BundleCoordinate> findOptionalBundleForType(final ExtensionManager extensionManager, final String type, final Bundle frameworkBundle) {
final List<Bundle> bundles = extensionManager.getBundles(type);
if (bundles.size() == 1) {
return Optional.of(bundles.get(0).getBundleDetails().getCoordinate());
}
+
+ // All NARs that are packaged with NiFi will have the same bundle coordinate as the NiFi framework bundle.
+ // During an upgrade, it's fairly common to have two versions of a NAR: the version shipped with NiFi and another version, perhaps to maintain
+ // backward compatibility to because the new version behaves some different way and the user wants the old behavior in some instances, etc.
+ // In this case, the user may have two versions. For example, version 1.26.0 and 1.27.0 while NiFi is at version 1.27.0.
+ // Now, during upgrade to 1.27.1, there will no longer be a 1.27.0 available. We want to be smart enough to realize that those extension using version
+ // 1.26.0 stay there but those using 1.27.0 upgrade to 1.27.1.
+ // To do this, we always first match on the exact version but this method is called when there's no exact match. So those marked 1.26.0 won't arrive here.
+ // But for those extensions that were using 1.27.0, we want to now look for version 1.27.1 - i.e., the one with the same version as the framework. If we
+ // find that version, then we want to use it. This helps to smooth out the upgrade process even when users have multiple versions of a given NAR.
+ final String frameworkVersion = frameworkBundle.getBundleDetails().getCoordinate().getVersion();
+ for (final Bundle bundle : bundles) {
+ final String componentVersion = bundle.getBundleDetails().getCoordinate().getVersion();
+ if (frameworkVersion.equals(componentVersion)) {
+ return Optional.of(bundle.getBundleDetails().getCoordinate());
+ }
+ }
+
return Optional.empty();
}
@@ -76,18 +95,17 @@ public final class BundleUtils {
private static Optional<BundleCoordinate> findOptionalCompatibleBundle(final ExtensionManager extensionManager, final String type,
- final BundleDTO bundleDTO, final boolean allowCompatibleBundle) {
+ final BundleDTO bundleDTO) {
final BundleCoordinate coordinate = new BundleCoordinate(bundleDTO.getGroup(), bundleDTO.getArtifact(), bundleDTO.getVersion());
final Bundle bundle = extensionManager.getBundle(coordinate);
if (bundle == null) {
- if (allowCompatibleBundle) {
- return findOptionalBundleForType(extensionManager, type, coordinate);
- } else {
- return Optional.empty();
- }
+ return findOptionalBundleForType(extensionManager, type, NarClassLoadersHolder.getInstance().getFrameworkBundle());
} else {
- final List<BundleCoordinate> bundlesForType = extensionManager.getBundles(type).stream().map(b -> b.getBundleDetails().getCoordinate()).collect(Collectors.toList());
+ final List<BundleCoordinate> bundlesForType = extensionManager.getBundles(type).stream()
+ .map(b -> b.getBundleDetails().getCoordinate())
+ .collect(Collectors.toList());
+
if (bundlesForType.contains(coordinate)) {
return Optional.of(coordinate);
} else {
@@ -175,9 +193,9 @@ public final class BundleUtils {
public static Optional<BundleCoordinate> getOptionalCompatibleBundle(final ExtensionManager extensionManager, final String type, final BundleDTO bundleDTO) {
if (bundleDTO == null) {
- return findOptionalBundleForType(extensionManager, type, null);
+ return findOptionalBundleForType(extensionManager, type, NarClassLoadersHolder.getInstance().getFrameworkBundle());
} else {
- return findOptionalCompatibleBundle(extensionManager, type, bundleDTO, true);
+ return findOptionalCompatibleBundle(extensionManager, type, bundleDTO);
}
}
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/util/TestBundleUtils.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/util/TestBundleUtils.java
new file mode 100644
index 0000000000..0368ac513b
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/util/TestBundleUtils.java
@@ -0,0 +1,100 @@
+/*
+ * 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.util;
+
+import org.apache.nifi.bundle.Bundle;
+import org.apache.nifi.bundle.BundleCoordinate;
+import org.apache.nifi.bundle.BundleDetails;
+import org.apache.nifi.nar.ExtensionManager;
+import org.apache.nifi.nar.NarClassLoaders;
+import org.apache.nifi.nar.NarClassLoadersHolder;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.when;
+
+public class TestBundleUtils {
+
+ private static final String PROCESSOR_TYPE = "MyProcessor";
+ private static final String FRAMEWORK_VERSION = "5.0.0";
+
+ private static final Bundle frameworkBundle = createBundle("framework-bundle", FRAMEWORK_VERSION);
+ private static ExtensionManager extensionManager;
+
+
+ @BeforeAll
+ public static void setup() throws IOException, ClassNotFoundException {
+ extensionManager = Mockito.mock(ExtensionManager.class);
+
+ final NarClassLoaders narClassLoaders = NarClassLoadersHolder.getInstance();
+ narClassLoaders.init(null, new File("target/extensions"));
+ }
+
+ @Test
+ public void findOptionalBundleMatchingFramework() throws IOException, ClassNotFoundException {
+ final Bundle frameworkVersionBundle = createBundle("my-bundle", FRAMEWORK_VERSION);
+ final Bundle otherBundle = createBundle("my-bundle", "1.2.3");
+ final List<Bundle> bundles = Arrays.asList(frameworkVersionBundle, otherBundle);
+ when(extensionManager.getBundles(PROCESSOR_TYPE)).thenReturn(bundles);
+
+ final Optional<BundleCoordinate> compatibleCoordinate = BundleUtils.findOptionalBundleForType(extensionManager, PROCESSOR_TYPE, frameworkBundle);
+ assertTrue(compatibleCoordinate.isPresent());
+ assertEquals(frameworkVersionBundle.getBundleDetails().getCoordinate(), compatibleCoordinate.get());
+ }
+
+ @Test
+ public void findOptionalBundleNotMatchingFramework() throws IOException, ClassNotFoundException {
+ final Bundle version3 = createBundle("my-bundle", "3.0.0");
+ final Bundle otherBundle = createBundle("my-bundle", "1.2.3");
+ final List<Bundle> bundles = Arrays.asList(version3, otherBundle);
+ when(extensionManager.getBundles(PROCESSOR_TYPE)).thenReturn(bundles);
+
+ final Optional<BundleCoordinate> compatibleCoordinate = BundleUtils.findOptionalBundleForType(extensionManager, PROCESSOR_TYPE, frameworkBundle);
+ assertFalse(compatibleCoordinate.isPresent());
+ }
+
+ @Test
+ public void testFindOptionalBundleOnlyOneBundle() throws IOException, ClassNotFoundException {
+ final Bundle otherBundle = createBundle("my-bundle", "1.2.3");
+ final List<Bundle> bundles = Collections.singletonList(otherBundle);
+ when(extensionManager.getBundles(PROCESSOR_TYPE)).thenReturn(bundles);
+
+ final Optional<BundleCoordinate> compatibleCoordinate = BundleUtils.findOptionalBundleForType(extensionManager, PROCESSOR_TYPE, frameworkBundle);
+ assertTrue(compatibleCoordinate.isPresent());
+ assertEquals(otherBundle.getBundleDetails().getCoordinate(), compatibleCoordinate.get());
+ }
+
+ private static Bundle createBundle(final String artifactId, final String version) {
+ final BundleDetails bundleDetails = new BundleDetails.Builder()
+ .coordinate(new BundleCoordinate("org.apache.nifi", artifactId, version))
+ .workingDir(new File("target"))
+ .build();
+
+ return new Bundle(bundleDetails, TestBundleUtils.class.getClassLoader());
+ }
+}
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java
index 6a99df2b2c..7617046be9 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java
@@ -1025,7 +1025,7 @@ public class StandardFlowService implements FlowService, ProtocolHandler {
controller.setNodeId(nodeId);
// load new controller state
- loadFromBytes(dataFlow, true, BundleUpdateStrategy.USE_SPECIFIED_OR_FAIL);
+ loadFromBytes(dataFlow, true, BundleUpdateStrategy.USE_SPECIFIED_OR_COMPATIBLE_OR_GHOST);
// set node ID on controller before we start heartbeating because heartbeat needs node ID
clusterCoordinator.setLocalNodeIdentifier(nodeId);
diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/clustering/FlowSynchronizationIT.java b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/clustering/FlowSynchronizationIT.java
index b2134f1eae..21d6de6000 100644
--- a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/clustering/FlowSynchronizationIT.java
+++ b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/clustering/FlowSynchronizationIT.java
@@ -356,84 +356,6 @@ public class FlowSynchronizationIT extends NiFiSystemIT {
});
}
- @Test
- public void testCannotJoinClusterIfMissingNar() throws NiFiClientException, IOException, InterruptedException {
- getClientUtil().createProcessor("GenerateFlowFile");
-
- // Shut down node 2
- disconnectNode(2);
- final NiFiInstance node2 = getNiFiInstance().getNodeInstance(2);
- node2.stop();
-
- // Remove node from the cluster. This way we know when it's attempted to connect
- final Integer node2ApiPort = getNodeApiPort(2);
- removeNode(2);
- removeExtensionsNar(node2);
-
- node2.start(false);
-
- // Wait until node is no longer removed from cluster, which will happen when it starts up and requests to connect
- waitFor(() -> !isNodeRemoved(node2ApiPort));
-
- // Wait for node to show as disconnected because it doesn't have the necessary nar
- waitForNodeState(2, NodeConnectionState.DISCONNECTED);
-
- // We need to restore the extensions nar and restart the node so that subsequent tests can succeed
- restoreExtensionsNar(node2);
- node2.stop();
- node2.start();
-
- waitForAllNodesConnected();
- }
-
- private void removeNode(final int index) throws NiFiClientException, IOException, InterruptedException {
- final NodeDTO nodeDto = getNodeEntity(index).getNode();
- final String nodeId = nodeDto.getNodeId();
- final Integer apiPort = nodeDto.getApiPort();
- getNifiClient().getControllerClient().deleteNode(nodeId);
- waitFor(() -> isNodeRemoved(apiPort));
- }
-
- private Integer getNodeApiPort(final int index) throws NiFiClientException, IOException {
- final NodeDTO nodeDto = getNodeEntity(index).getNode();
- final Integer apiPort = nodeDto.getApiPort();
- return apiPort;
- }
-
- @Test
- public void testCanJoinClusterIfAllNodesMissingNar() throws NiFiClientException, IOException, InterruptedException {
- final ProcessorEntity generate = getClientUtil().createProcessor("GenerateFlowFile");
-
- // Shut down node 2
- disconnectNode(2);
- final NiFiInstance node2 = getNiFiInstance().getNodeInstance(2);
- node2.stop();
-
- final NiFiInstance node1 = getNiFiInstance().getNodeInstance(1);
- node1.stop();
-
- removeExtensionsNar(node1);
- removeExtensionsNar(node2);
-
- node1.start(false);
- node2.start(true);
-
- waitForAllNodesConnected();
-
- assertTrue(getNifiClient().getProcessorClient().getProcessor(generate.getId()).getComponent().getExtensionMissing());
-
- // In order to ensure that subsequent tests are able to operate properly, we need to restore the nar and restart
- node1.stop();
- node2.stop();
-
- restoreExtensionsNar(node1);
- restoreExtensionsNar(node2);
-
- node1.start(false);
- node2.start(true);
- waitForAllNodesConnected();
- }
-
@Test
public void testCannotRemoveComponentsWhileNodeDisconnected() throws NiFiClientException, IOException, InterruptedException {
@@ -500,36 +422,6 @@ public class FlowSynchronizationIT extends NiFiSystemIT {
}
- private void removeExtensionsNar(final NiFiInstance nifiInstance) {
- final File extensionsNar = getExtensionsNar(nifiInstance);
- final File backupFile = new File(extensionsNar.getParentFile(), extensionsNar.getName() + ".backup");
- assertTrue(extensionsNar.renameTo(backupFile));
- }
-
- private void restoreExtensionsNar(final NiFiInstance nifiInstance) {
- final File backupFile = getExtensionsNar(nifiInstance);
- final File extensionsNar = new File(backupFile.getParentFile(), backupFile.getName().replace(".backup", ""));
- assertTrue(backupFile.renameTo(extensionsNar));
- }
-
- private File getExtensionsNar(final NiFiInstance nifiInstance) {
- final File libDir = new File(nifiInstance.getInstanceDirectory(), "lib");
- final File[] testExtensionsNar = libDir.listFiles(file -> file.getName().startsWith("nifi-system-test-extensions-nar-"));
- assertEquals(1, testExtensionsNar.length);
-
- return testExtensionsNar[0];
- }
-
-
- private boolean isNodeRemoved(final int apiPort) {
- try {
- return getNifiClient().getControllerClient().getNodes().getCluster().getNodes().stream()
- .noneMatch(dto -> dto.getApiPort() == apiPort);
- } catch (Exception e) {
- return false;
- }
- }
-
@Test
public void testRestartWithFlowXmlGzNoJson() throws NiFiClientException, IOException {
restartWithOnlySingleFlowPersistenceFile("flow.json.gz");