NIFI-16305 Fix Connector mock framework behavior

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();
 

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<Void> 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,37 +533,14 @@ private VersionedExternalFlow createFlowSnapshot(final FrameworkFlowContext flow
         final ParameterContext parameterContext = processGroup.getParameterContext();
         if (parameterContext != null) {
             final Map<String, VersionedParameterContext> parameterContexts = new HashMap<>();
-            final VersionedParameterContext versionedParameterContext = createVersionedParameterContext(parameterContext);
+            final VersionedParameterContext versionedParameterContext = flowMapper.mapParameterContext(parameterContext);
             parameterContexts.put(versionedParameterContext.getName(), versionedParameterContext);
             externalFlow.setParameterContexts(parameterContexts);
         }
 
         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<VersionedParameter> 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<? extends Processor> mockProcessorClass) {
         mockExtensionMapper.mockProcessor(processorType, mockProcessorClass.getName());

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<Secret> 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<Secret> getAllSecrets() {
     @Override
     public List<Secret> getSecrets(final List<String> fullyQualifiedSecretNames) {
         final List<Secret> 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<Secret> getSecrets(final List<String> fullyQualifiedSecretNames) {
                     .providerName(SECRET_PROVIDER_NAME)
                     .groupName(GROUP_NAME)
                     .name(secretName)
-                    .fullyQualifiedName(GROUP_NAME + "." + secretName)
+                    .fullyQualifiedName(fullyQualifiedSecretName)
                     .value(value)
                     .authorizable(AUTHORIZABLE)
                     .build();

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<Void> 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);
+    }
+}

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<Secret> 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());
+    }
+}

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,21 +70,15 @@ 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"))
                 .build()) {
 
             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<VersionedProcessor> findProcessorByType(final Set<VersionedProcessor> processors, final String type) {
         for (final VersionedProcessor processor : processors) {
             if (type.equals(processor.getType())) {