From 54feee05184a465e7f3d841a16ca3e107121760e Mon Sep 17 00:00:00 2001 From: Mark Payne Date: Mon, 7 Sep 2026 10:11:48 -0400 Subject: [PATCH] NIFI-16305 Fix Connector mock framework behavior --- .../connector/server/ConnectorTestRunner.java | 5 +- .../server/StandardConnectorMockServer.java | 63 ++++---- .../ConnectorTestRunnerSecretProvider.java | 12 +- .../StandardConnectorMockServerTest.java | 82 ++++++++++ ...ConnectorTestRunnerSecretsManagerTest.java | 89 +++++++++++ .../connectors/tests/CreateConnectorIT.java | 30 ++-- .../mock/connectors/tests/FlowSnapshotIT.java | 95 ++++++++++++ .../connectors/FlowSnapshotConnector.java | 142 ++++++++++++++++++ ...apache.nifi.components.connector.Connector | 3 +- 9 files changed, 478 insertions(+), 43 deletions(-) create mode 100644 nifi-connector-mock-bundle/nifi-connector-mock-server/src/test/java/org/apache/nifi/mock/connector/server/StandardConnectorMockServerTest.java create mode 100644 nifi-connector-mock-bundle/nifi-connector-mock-server/src/test/java/org/apache/nifi/mock/connector/server/secrets/ConnectorTestRunnerSecretsManagerTest.java create mode 100644 nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-integration-tests/src/test/java/org/apache/nifi/mock/connectors/tests/FlowSnapshotIT.java create mode 100644 nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-test-connectors/src/main/java/org/apache/nifi/mock/connectors/FlowSnapshotConnector.java diff --git a/nifi-connector-mock-bundle/nifi-connector-mock-api/src/main/java/org/apache/nifi/mock/connector/server/ConnectorTestRunner.java b/nifi-connector-mock-bundle/nifi-connector-mock-api/src/main/java/org/apache/nifi/mock/connector/server/ConnectorTestRunner.java index 42f4c1821ee9..9678330f676b 100644 --- a/nifi-connector-mock-bundle/nifi-connector-mock-api/src/main/java/org/apache/nifi/mock/connector/server/ConnectorTestRunner.java +++ b/nifi-connector-mock-bundle/nifi-connector-mock-api/src/main/java/org/apache/nifi/mock/connector/server/ConnectorTestRunner.java @@ -136,7 +136,10 @@ public interface ConnectorTestRunner extends Closeable { AssetReference addAsset(String assetName, InputStream contents); /** - * Starts the Connector, beginning processing of data through its managed flow. + * Starts the Connector, beginning processing of data through its managed flow. Failures detected while + * initiating startup are propagated to the caller; lifecycle startup continues asynchronously. + * + * @throws IllegalStateException if the Connector cannot begin startup */ void startConnector(); diff --git a/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/StandardConnectorMockServer.java b/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/StandardConnectorMockServer.java index 15e54d6d971f..bbca1347a4de 100644 --- a/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/StandardConnectorMockServer.java +++ b/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/StandardConnectorMockServer.java @@ -61,14 +61,12 @@ import org.apache.nifi.engine.FlowEngine; import org.apache.nifi.events.VolatileBulletinRepository; import org.apache.nifi.flow.VersionedExternalFlow; -import org.apache.nifi.flow.VersionedParameter; import org.apache.nifi.flow.VersionedParameterContext; import org.apache.nifi.flow.VersionedProcessGroup; import org.apache.nifi.groups.ProcessGroup; import org.apache.nifi.mock.connector.server.secrets.ConnectorTestRunnerSecretProvider; import org.apache.nifi.mock.connector.server.secrets.ConnectorTestRunnerSecretsManager; import org.apache.nifi.nar.ExtensionMapping; -import org.apache.nifi.parameter.Parameter; import org.apache.nifi.parameter.ParameterContext; import org.apache.nifi.processor.Processor; import org.apache.nifi.registry.flow.mapping.ComponentIdLookup; @@ -97,11 +95,13 @@ import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; -import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.concurrent.CancellationException; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; import java.util.concurrent.TimeoutException; import java.util.jar.JarFile; import java.util.stream.Stream; @@ -328,7 +328,9 @@ public void addSecret(final String name, final String value) { @Override public SecretReference createSecretReference(final String secretName) { - return new SecretReference(ConnectorTestRunnerSecretProvider.SECRET_PROVIDER_ID, ConnectorTestRunnerSecretProvider.SECRET_PROVIDER_NAME, secretName, secretName); + final String fullyQualifiedName = ConnectorTestRunnerSecretProvider.SECRET_PROVIDER_NAME + "." + + ConnectorTestRunnerSecretProvider.GROUP_NAME + "." + secretName; + return new SecretReference(ConnectorTestRunnerSecretProvider.SECRET_PROVIDER_ID, ConnectorTestRunnerSecretProvider.SECRET_PROVIDER_NAME, secretName, fullyQualifiedName); } @Override @@ -359,7 +361,33 @@ public AssetReference addAsset(final String assetName, final InputStream content public void startConnector() { initialFlowFileTransferCounts = connectorNode.getFlowFileTransferCounts(); - connectorNode.start(flowEngine); + final Future startFuture = connectorNode.start(flowEngine); + // Property resolution and validation failures complete the Future before ConnectorNode.start() returns. + // Lifecycle startup is asynchronous and can remain pending while the node retries, so do not wait for it here. + if (!startFuture.isDone()) { + return; + } + + final IllegalStateException startupFailure; + try { + startFuture.get(); + return; + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + startupFailure = new IllegalStateException("Interrupted while checking Connector startup", e); + } catch (final ExecutionException e) { + startupFailure = new IllegalStateException("Failed to start Connector", e.getCause()); + } catch (final CancellationException e) { + startupFailure = new IllegalStateException("Connector startup was cancelled", e); + } + + try { + connectorNode.stop(flowEngine); + } catch (final RuntimeException stopFailure) { + startupFailure.addSuppressed(stopFailure); + } + + throw startupFailure; } @Override @@ -505,7 +533,7 @@ private VersionedExternalFlow createFlowSnapshot(final FrameworkFlowContext flow final ParameterContext parameterContext = processGroup.getParameterContext(); if (parameterContext != null) { final Map parameterContexts = new HashMap<>(); - final VersionedParameterContext versionedParameterContext = createVersionedParameterContext(parameterContext); + final VersionedParameterContext versionedParameterContext = flowMapper.mapParameterContext(parameterContext); parameterContexts.put(versionedParameterContext.getName(), versionedParameterContext); externalFlow.setParameterContexts(parameterContexts); } @@ -513,29 +541,6 @@ private VersionedExternalFlow createFlowSnapshot(final FrameworkFlowContext flow return externalFlow; } - private VersionedParameterContext createVersionedParameterContext(final ParameterContext parameterContext) { - final VersionedParameterContext versionedParameterContext = new VersionedParameterContext(); - versionedParameterContext.setName(parameterContext.getName()); - versionedParameterContext.setDescription(parameterContext.getDescription()); - versionedParameterContext.setIdentifier(parameterContext.getIdentifier()); - - final Set versionedParameters = new LinkedHashSet<>(); - for (final Parameter parameter : parameterContext.getParameters().values()) { - final VersionedParameter versionedParameter = new VersionedParameter(); - versionedParameter.setName(parameter.getDescriptor().getName()); - versionedParameter.setDescription(parameter.getDescriptor().getDescription()); - versionedParameter.setSensitive(parameter.getDescriptor().isSensitive()); - versionedParameter.setProvided(parameter.getDescriptor().isSensitive()); - if (!parameter.getDescriptor().isSensitive()) { - versionedParameter.setValue(parameter.getValue()); - } - versionedParameters.add(versionedParameter); - } - versionedParameterContext.setParameters(versionedParameters); - - return versionedParameterContext; - } - @Override public void mockProcessor(final String processorType, final Class mockProcessorClass) { mockExtensionMapper.mockProcessor(processorType, mockProcessorClass.getName()); diff --git a/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/secrets/ConnectorTestRunnerSecretProvider.java b/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/secrets/ConnectorTestRunnerSecretProvider.java index 94fd6b9c114f..9a512b344d3e 100644 --- a/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/secrets/ConnectorTestRunnerSecretProvider.java +++ b/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/secrets/ConnectorTestRunnerSecretProvider.java @@ -60,7 +60,7 @@ public List getAllSecrets() { .name(entry.getKey()) .value(entry.getValue()) .authorizable(AUTHORIZABLE) - .fullyQualifiedName(GROUP_NAME + "." + entry.getKey()) + .fullyQualifiedName(SECRET_PROVIDER_NAME + "." + GROUP_NAME + "." + entry.getKey()) .build(); secrets.add(secret); @@ -72,8 +72,14 @@ public List getAllSecrets() { @Override public List getSecrets(final List fullyQualifiedSecretNames) { final List matchingSecrets = new ArrayList<>(); + final String secretNamePrefix = SECRET_PROVIDER_NAME + "." + GROUP_NAME + "."; - for (final String secretName : fullyQualifiedSecretNames) { + for (final String fullyQualifiedSecretName : fullyQualifiedSecretNames) { + if (!fullyQualifiedSecretName.startsWith(secretNamePrefix)) { + continue; + } + + final String secretName = fullyQualifiedSecretName.substring(secretNamePrefix.length()); final String value = secrets.get(secretName); if (value != null) { @@ -82,7 +88,7 @@ public List getSecrets(final List fullyQualifiedSecretNames) { .providerName(SECRET_PROVIDER_NAME) .groupName(GROUP_NAME) .name(secretName) - .fullyQualifiedName(GROUP_NAME + "." + secretName) + .fullyQualifiedName(fullyQualifiedSecretName) .value(value) .authorizable(AUTHORIZABLE) .build(); diff --git a/nifi-connector-mock-bundle/nifi-connector-mock-server/src/test/java/org/apache/nifi/mock/connector/server/StandardConnectorMockServerTest.java b/nifi-connector-mock-bundle/nifi-connector-mock-server/src/test/java/org/apache/nifi/mock/connector/server/StandardConnectorMockServerTest.java new file mode 100644 index 000000000000..b9ac2efb58ab --- /dev/null +++ b/nifi-connector-mock-bundle/nifi-connector-mock-server/src/test/java/org/apache/nifi/mock/connector/server/StandardConnectorMockServerTest.java @@ -0,0 +1,82 @@ +/* + * 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.mock.connector.server; + +import org.apache.nifi.components.connector.ConnectorNode; +import org.apache.nifi.connectable.FlowFileTransferCounts; +import org.apache.nifi.engine.FlowEngine; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Future; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@ExtendWith(MockitoExtension.class) +class StandardConnectorMockServerTest { + + @Mock + private ConnectorNode connectorNode; + + @Mock + private FlowEngine flowEngine; + + @Mock + private Future startFuture; + + @InjectMocks + private StandardConnectorMockServer server; + + @Test + void testStartConnectorPropagatesFailureAndRequestsStop() { + final IllegalStateException startFailure = new IllegalStateException("Connector is not valid"); + when(connectorNode.getFlowFileTransferCounts()).thenReturn(new FlowFileTransferCounts(0, 0, 0, 0)); + when(connectorNode.start(flowEngine)).thenReturn(CompletableFuture.failedFuture(startFailure)); + + final IllegalStateException exception = assertThrows(IllegalStateException.class, server::startConnector); + + assertEquals("Failed to start Connector", exception.getMessage()); + assertSame(startFailure, exception.getCause()); + verify(connectorNode).stop(flowEngine); + } + + @Test + void testStartConnectorDoesNotWaitForPendingStart() throws Exception { + when(connectorNode.getFlowFileTransferCounts()).thenReturn(new FlowFileTransferCounts(0, 0, 0, 0)); + when(connectorNode.start(flowEngine)).thenReturn(startFuture); + when(startFuture.isDone()).thenReturn(false); + + server.startConnector(); + + verify(startFuture).isDone(); + verify(startFuture, never()).get(); + verify(startFuture, never()).get(anyLong(), any()); + verify(connectorNode, never()).stop(flowEngine); + } +} diff --git a/nifi-connector-mock-bundle/nifi-connector-mock-server/src/test/java/org/apache/nifi/mock/connector/server/secrets/ConnectorTestRunnerSecretsManagerTest.java b/nifi-connector-mock-bundle/nifi-connector-mock-server/src/test/java/org/apache/nifi/mock/connector/server/secrets/ConnectorTestRunnerSecretsManagerTest.java new file mode 100644 index 000000000000..9071c913f491 --- /dev/null +++ b/nifi-connector-mock-bundle/nifi-connector-mock-server/src/test/java/org/apache/nifi/mock/connector/server/secrets/ConnectorTestRunnerSecretsManagerTest.java @@ -0,0 +1,89 @@ +/* + * 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.mock.connector.server.secrets; + +import org.apache.nifi.components.connector.Secret; +import org.apache.nifi.components.connector.SecretReference; +import org.apache.nifi.mock.connector.server.StandardConnectorMockServer; +import org.junit.jupiter.api.Test; + +import java.util.Optional; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class ConnectorTestRunnerSecretsManagerTest { + private static final String SECRET_NAME = "password"; + private static final String SECRET_VALUE = "secret-value"; + private static final String FULLY_QUALIFIED_SECRET_NAME = ConnectorTestRunnerSecretProvider.SECRET_PROVIDER_NAME + "." + + ConnectorTestRunnerSecretProvider.GROUP_NAME + "." + SECRET_NAME; + + @Test + void testGetSecretUsingDiscoveredFullyQualifiedName() { + final ConnectorTestRunnerSecretsManager secretsManager = new ConnectorTestRunnerSecretsManager(); + secretsManager.addSecret(SECRET_NAME, SECRET_VALUE); + + final Secret discoveredSecret = secretsManager.getAllSecrets().getFirst(); + assertEquals(FULLY_QUALIFIED_SECRET_NAME, discoveredSecret.getFullyQualifiedName()); + + final SecretReference secretReference = new SecretReference( + discoveredSecret.getProviderId(), + discoveredSecret.getProviderName(), + discoveredSecret.getName(), + discoveredSecret.getFullyQualifiedName() + ); + + final Optional resolvedSecret = secretsManager.getSecret(secretReference); + + assertTrue(resolvedSecret.isPresent(), "Discovered secret should be resolved using its fully qualified name"); + final Secret secret = resolvedSecret.orElseThrow(); + assertEquals(SECRET_NAME, secret.getName()); + assertEquals(FULLY_QUALIFIED_SECRET_NAME, secret.getFullyQualifiedName()); + assertEquals(SECRET_VALUE, secret.getValue()); + } + + @Test + void testCreateSecretReferenceUsesFullyQualifiedName() { + final ConnectorTestRunnerSecretsManager secretsManager = new ConnectorTestRunnerSecretsManager(); + secretsManager.addSecret(SECRET_NAME, SECRET_VALUE); + final StandardConnectorMockServer server = new StandardConnectorMockServer(); + + final SecretReference secretReference = server.createSecretReference(SECRET_NAME); + + assertEquals(FULLY_QUALIFIED_SECRET_NAME, secretReference.getFullyQualifiedName()); + assertEquals(SECRET_VALUE, secretsManager.getSecret(secretReference).orElseThrow().getValue()); + } + + @Test + void testGetSecretWithDotInName() { + final String secretName = "database.password"; + final ConnectorTestRunnerSecretsManager secretsManager = new ConnectorTestRunnerSecretsManager(); + secretsManager.addSecret(secretName, SECRET_VALUE); + + final Secret discoveredSecret = secretsManager.getAllSecrets().getFirst(); + final SecretReference secretReference = new SecretReference( + discoveredSecret.getProviderId(), + discoveredSecret.getProviderName(), + discoveredSecret.getName(), + discoveredSecret.getFullyQualifiedName() + ); + + assertEquals("TestRunnerSecretsManager.Default.database.password", discoveredSecret.getFullyQualifiedName()); + assertEquals(SECRET_VALUE, secretsManager.getSecret(secretReference).orElseThrow().getValue()); + } +} diff --git a/nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-integration-tests/src/test/java/org/apache/nifi/mock/connectors/tests/CreateConnectorIT.java b/nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-integration-tests/src/test/java/org/apache/nifi/mock/connectors/tests/CreateConnectorIT.java index 1b8c49e49f40..f3d529c6a5ef 100644 --- a/nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-integration-tests/src/test/java/org/apache/nifi/mock/connectors/tests/CreateConnectorIT.java +++ b/nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-integration-tests/src/test/java/org/apache/nifi/mock/connectors/tests/CreateConnectorIT.java @@ -32,9 +32,11 @@ import java.util.List; import java.util.Optional; import java.util.Set; +import java.util.concurrent.TimeoutException; -import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; public class CreateConnectorIT { @@ -68,7 +70,7 @@ public void testCreateStartAndStopGenerateAndUpdateConnector() throws IOExceptio } @Test - public void testStopConnectorWithTimeoutStopsRunningConnector() throws IOException { + public void testStopConnectorWithTimeoutStopsRunningConnector() throws IOException, TimeoutException { try (final ConnectorTestRunner testRunner = new StandardConnectorTestRunner.Builder() .connectorClassName("org.apache.nifi.mock.connectors.GenerateAndLog") .narLibraryDirectory(new File("target/libDir")) @@ -76,13 +78,7 @@ public void testStopConnectorWithTimeoutStopsRunningConnector() throws IOExcepti testRunner.startConnector(); - // Exercises the timeout-aware overload: it initiates the asynchronous stop and actively polls the - // Connector's state until it reaches STOPPED, returning as soon as it does rather than after a single - // fixed blocking wait. Because start is asynchronous, a stop issued immediately afterwards may have to - // ride through the node's internal stop retries (every 10 seconds) before the state settles, so the - // budget is generous enough to stay deterministic on a slow CI runner; the poll still returns the - // instant the Connector reports STOPPED. - assertDoesNotThrow(() -> testRunner.stopConnector(Duration.ofSeconds(120))); + testRunner.stopConnector(Duration.ofSeconds(120)); } } @@ -102,6 +98,22 @@ public void testConnectorWithMissingBundleFailsValidate() throws IOException { } } + @Test + public void testConnectorWithMissingBundleFailsStart() throws IOException { + try (final ConnectorTestRunner testRunner = new StandardConnectorTestRunner.Builder() + .connectorClassName("org.apache.nifi.mock.connectors.MissingBundleConnector") + .narLibraryDirectory(new File("target/libDir")) + .build()) { + + final IllegalStateException exception = assertThrows(IllegalStateException.class, testRunner::startConnector); + assertEquals("Failed to start Connector", exception.getMessage()); + + final IllegalStateException cause = assertInstanceOf(IllegalStateException.class, exception.getCause()); + assertTrue(cause.getMessage().contains("com.example.nonexistent:missing-nar:1.0.0")); + assertTrue(cause.getMessage().contains("com.example.nonexistent.MissingProcessor")); + } + } + private Optional findProcessorByType(final Set processors, final String type) { for (final VersionedProcessor processor : processors) { if (type.equals(processor.getType())) { diff --git a/nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-integration-tests/src/test/java/org/apache/nifi/mock/connectors/tests/FlowSnapshotIT.java b/nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-integration-tests/src/test/java/org/apache/nifi/mock/connectors/tests/FlowSnapshotIT.java new file mode 100644 index 000000000000..ce0c4465879e --- /dev/null +++ b/nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-integration-tests/src/test/java/org/apache/nifi/mock/connectors/tests/FlowSnapshotIT.java @@ -0,0 +1,95 @@ +/* + * 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.mock.connectors.tests; + +import org.apache.nifi.components.connector.AssetReference; +import org.apache.nifi.components.connector.FlowUpdateException; +import org.apache.nifi.flow.VersionedAsset; +import org.apache.nifi.flow.VersionedExternalFlow; +import org.apache.nifi.flow.VersionedParameter; +import org.apache.nifi.flow.VersionedParameterContext; +import org.apache.nifi.mock.connector.StandardConnectorTestRunner; +import org.apache.nifi.mock.connector.server.ConnectorTestRunner; +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayInputStream; +import java.io.File; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class FlowSnapshotIT { + private static final String CONNECTOR_CLASS = "org.apache.nifi.mock.connectors.FlowSnapshotConnector"; + private static final String CONFIGURATION_STEP = "Snapshot Configuration"; + private static final String ASSET_IDENTIFIER_PROPERTY = "Asset Identifier"; + private static final String ASSET_NAME = "certificate.pem"; + private static final String SENSITIVE_PARAMETER_NAME = "sensitive_parameter"; + private static final String ASSET_PARAMETER_NAME = "asset_parameter"; + + @Test + public void testWorkingAndActiveFlowSnapshotsPreserveParameterMetadata() throws IOException, FlowUpdateException { + try (final ConnectorTestRunner runner = new StandardConnectorTestRunner.Builder() + .connectorClassName(CONNECTOR_CLASS) + .narLibraryDirectory(new File("target/libDir")) + .build()) { + + final ByteArrayInputStream assetContents = new ByteArrayInputStream("certificate contents".getBytes(StandardCharsets.UTF_8)); + final AssetReference assetReference = runner.addAsset(ASSET_NAME, assetContents); + final String assetIdentifier = assetReference.getAssetIdentifiers().iterator().next(); + + runner.configure(CONFIGURATION_STEP, Map.of(ASSET_IDENTIFIER_PROPERTY, assetIdentifier)); + assertSnapshotParameters(runner.getWorkingFlowSnapshot(), assetIdentifier); + + runner.applyUpdate(); + assertSnapshotParameters(runner.getActiveFlowSnapshot(), assetIdentifier); + } + } + + private void assertSnapshotParameters(final VersionedExternalFlow snapshot, final String assetIdentifier) { + assertEquals(1, snapshot.getParameterContexts().size()); + final VersionedParameterContext parameterContext = snapshot.getParameterContexts().values().iterator().next(); + + final VersionedParameter sensitiveParameter = getParameter(parameterContext, SENSITIVE_PARAMETER_NAME); + final VersionedParameter assetParameter = getParameter(parameterContext, ASSET_PARAMETER_NAME); + assertTrue(sensitiveParameter.isSensitive()); + assertFalse(sensitiveParameter.isProvided()); + assertNull(sensitiveParameter.getValue()); + assertFalse(assetParameter.isSensitive()); + assertFalse(assetParameter.isProvided()); + assertNull(assetParameter.getValue()); + assertNotNull(assetParameter.getReferencedAssets()); + assertEquals(1, assetParameter.getReferencedAssets().size()); + + final VersionedAsset versionedAsset = assetParameter.getReferencedAssets().getFirst(); + assertEquals(assetIdentifier, versionedAsset.getIdentifier()); + assertEquals(ASSET_NAME, versionedAsset.getName()); + } + + private VersionedParameter getParameter(final VersionedParameterContext parameterContext, final String name) { + return parameterContext.getParameters().stream() + .filter(parameter -> name.equals(parameter.getName())) + .findFirst() + .orElseThrow(); + } +} diff --git a/nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-test-connectors/src/main/java/org/apache/nifi/mock/connectors/FlowSnapshotConnector.java b/nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-test-connectors/src/main/java/org/apache/nifi/mock/connectors/FlowSnapshotConnector.java new file mode 100644 index 000000000000..fb67e70aa1d7 --- /dev/null +++ b/nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-test-connectors/src/main/java/org/apache/nifi/mock/connectors/FlowSnapshotConnector.java @@ -0,0 +1,142 @@ +/* + * 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.mock.connectors; + +import org.apache.nifi.components.ConfigVerificationResult; +import org.apache.nifi.components.connector.AbstractConnector; +import org.apache.nifi.components.connector.BundleCompatibility; +import org.apache.nifi.components.connector.ConfigurationStep; +import org.apache.nifi.components.connector.ConnectorPropertyDescriptor; +import org.apache.nifi.components.connector.ConnectorPropertyGroup; +import org.apache.nifi.components.connector.FlowUpdateException; +import org.apache.nifi.components.connector.components.FlowContext; +import org.apache.nifi.components.connector.util.VersionedFlowUtils; +import org.apache.nifi.flow.VersionedAsset; +import org.apache.nifi.flow.VersionedExternalFlow; +import org.apache.nifi.flow.VersionedParameter; +import org.apache.nifi.flow.VersionedParameterContext; +import org.apache.nifi.flow.VersionedProcessor; + +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * Test Connector that installs sensitive and asset-backed parameters for exercising flow snapshots. + */ +public class FlowSnapshotConnector extends AbstractConnector { + private static final String FLOW_RESOURCE = "flows/Generate_and_Update.json"; + private static final String STEP_NAME = "Snapshot Configuration"; + private static final String ASSET_IDENTIFIER_PROPERTY_NAME = "Asset Identifier"; + private static final String PARAMETER_CONTEXT_NAME = "Snapshot Parameter Context"; + private static final String SENSITIVE_PARAMETER_NAME = "sensitive_parameter"; + private static final String ASSET_PARAMETER_NAME = "asset_parameter"; + private static final String ASSET_NAME = "certificate.pem"; + + private static final ConnectorPropertyDescriptor ASSET_IDENTIFIER = new ConnectorPropertyDescriptor.Builder() + .name(ASSET_IDENTIFIER_PROPERTY_NAME) + .description("Identifier of the asset referenced by the snapshot test flow") + .required(true) + .build(); + + private static final ConnectorPropertyGroup PROPERTY_GROUP = new ConnectorPropertyGroup.Builder() + .name("Snapshot Properties") + .addProperty(ASSET_IDENTIFIER) + .build(); + + private static final ConfigurationStep CONFIGURATION_STEP = new ConfigurationStep.Builder() + .name(STEP_NAME) + .propertyGroups(List.of(PROPERTY_GROUP)) + .build(); + + @Override + public VersionedExternalFlow getInitialFlow() { + return VersionedFlowUtils.loadFlowFromResource(FLOW_RESOURCE); + } + + @Override + public VersionedExternalFlow getActiveFlow(final FlowContext activeFlowContext) { + final String assetIdentifier = getAssetIdentifier(activeFlowContext); + return assetIdentifier == null ? getInitialFlow() : createConfiguredFlow(assetIdentifier); + } + + @Override + public List getConfigurationSteps() { + return List.of(CONFIGURATION_STEP); + } + + @Override + protected void onStepConfigured(final String stepName, final FlowContext workingContext) throws FlowUpdateException { + if (STEP_NAME.equals(stepName)) { + final String assetIdentifier = getAssetIdentifier(workingContext); + if (assetIdentifier != null) { + getInitializationContext().updateFlow(workingContext, createConfiguredFlow(assetIdentifier), BundleCompatibility.RESOLVE_BUNDLE); + } + } + } + + @Override + public void applyUpdate(final FlowContext workingContext, final FlowContext activeContext) throws FlowUpdateException { + final String assetIdentifier = getAssetIdentifier(workingContext); + if (assetIdentifier != null) { + getInitializationContext().updateFlow(activeContext, createConfiguredFlow(assetIdentifier), BundleCompatibility.RESOLVE_BUNDLE); + } + } + + @Override + public List verifyConfigurationStep(final String stepName, final Map overrides, final FlowContext workingContext) { + return List.of(); + } + + private String getAssetIdentifier(final FlowContext flowContext) { + return flowContext.getConfigurationContext().getProperty(CONFIGURATION_STEP, ASSET_IDENTIFIER).getValue(); + } + + private VersionedExternalFlow createConfiguredFlow(final String assetIdentifier) { + final VersionedExternalFlow flow = getInitialFlow(); + + final VersionedParameter sensitiveParameter = new VersionedParameter(); + sensitiveParameter.setName(SENSITIVE_PARAMETER_NAME); + sensitiveParameter.setValue("sensitive-value"); + sensitiveParameter.setSensitive(true); + sensitiveParameter.setProvided(false); + + final VersionedAsset asset = new VersionedAsset(); + asset.setIdentifier(assetIdentifier); + asset.setName(ASSET_NAME); + + final VersionedParameter assetParameter = new VersionedParameter(); + assetParameter.setName(ASSET_PARAMETER_NAME); + assetParameter.setSensitive(false); + assetParameter.setProvided(false); + assetParameter.setReferencedAssets(List.of(asset)); + + final VersionedParameterContext parameterContext = new VersionedParameterContext(); + parameterContext.setName(PARAMETER_CONTEXT_NAME); + parameterContext.setParameters(Set.of(sensitiveParameter, assetParameter)); + flow.setParameterContexts(Map.of(PARAMETER_CONTEXT_NAME, parameterContext)); + flow.getFlowContents().setParameterContextName(PARAMETER_CONTEXT_NAME); + + final VersionedProcessor generateFlowFile = VersionedFlowUtils.findProcessor(flow.getFlowContents(), + processor -> processor.getType().equals("org.apache.nifi.processors.standard.GenerateFlowFile")) + .orElseThrow(); + generateFlowFile.getProperties().put("Custom Text", "#{" + ASSET_PARAMETER_NAME + "}"); + + return flow; + } +} diff --git a/nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-test-connectors/src/main/resources/META-INF/services/org.apache.nifi.components.connector.Connector b/nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-test-connectors/src/main/resources/META-INF/services/org.apache.nifi.components.connector.Connector index 379d9026452a..391ca0e87593 100644 --- a/nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-test-connectors/src/main/resources/META-INF/services/org.apache.nifi.components.connector.Connector +++ b/nifi-connector-mock-bundle/nifi-connector-mock-test-bundle/nifi-connector-mock-test-connectors/src/main/resources/META-INF/services/org.apache.nifi.components.connector.Connector @@ -16,4 +16,5 @@ org.apache.nifi.mock.connectors.AllowableValuesConnector org.apache.nifi.mock.connectors.GenerateAndLog org.apache.nifi.mock.connectors.MissingBundleConnector -org.apache.nifi.mock.connectors.CronScheduleConnector \ No newline at end of file +org.apache.nifi.mock.connectors.CronScheduleConnector +org.apache.nifi.mock.connectors.FlowSnapshotConnector