(nifi) branch NIFI-15258 updated: NIFI-15312: Implementation of Parameter Provider based Secrets Manager; updated mock framework to support secrets; validation cleanup (#10638)

[mcgilman]

This is an automated email from the ASF dual-hosted git repository.

mcgilman pushed a commit to branch NIFI-15258
in repository https://gitbox.apache.org/repos/asf/nifi.git


The following commit(s) were added to refs/heads/NIFI-15258 by this push:
     new 292d27d8881 NIFI-15312: Implementation of Parameter Provider based 
Secrets Manager; updated mock framework to support secrets; validation cleanup 
(#10638)
292d27d8881 is described below

commit 292d27d88810c207cf81d6af804594dfa5fd7e94
Author: Mark Payne <marka...@hotmail.com>
AuthorDate: Fri Dec 12 18:01:16 2025 -0500

    NIFI-15312: Implementation of Parameter Provider based Secrets Manager; 
updated mock framework to support secrets; validation cleanup (#10638)
---
 nifi-assembly/pom.xml                              |  12 +-
 .../java/org/apache/nifi/util/NiFiProperties.java  |   3 +
 .../mock/connector/server/ConnectorTestRunner.java |  12 +
 .../server/MockConnectorInitializationContext.java |   2 +-
 .../server/StandardConnectorMockServer.java        |  46 ++-
 .../secrets/ConnectorTestRunnerSecretProvider.java |  86 ++++++
 .../secrets/ConnectorTestRunnerSecretsManager.java |  99 +++++++
 ...ifi.components.connector.secrets.SecretsManager |  16 ++
 .../connector/StandardConnectorTestRunner.java     |  24 ++
 .../src/main/resources/nifi.properties             |   3 +
 .../nifi/connectors/kafkas3/KafkaToS3IT.java       |  78 +++--
 .../secrets/ParameterProviderSecretProvider.java   |  86 ++++++
 .../secrets/ParameterProviderSecretsManager.java   | 151 ++++++++++
 .../connector/secrets/StandardSecret.java          | 127 ++++++++
 ...andardSecretsManagerInitializationContext.java} |  21 +-
 .../parameter/StandardParameterProviderNode.java   |  22 +-
 ...ifi.components.connector.secrets.SecretsManager |  16 ++
 .../TestParameterProviderSecretsManager.java       | 318 +++++++++++++++++++++
 .../connector/ConnectorConfiguration.java          |   9 +
 .../components/connector/ConnectorRepository.java  |   3 +
 .../ConnectorRepositoryInitializationContext.java  |   3 +
 .../FrameworkConnectorInitializationContext.java   |   1 +
 ...eworkConnectorInitializationContextBuilder.java |   1 +
 .../SecretProvider.java}                           |  16 +-
 .../SecretsManager.java}                           |  24 +-
 .../SecretsManagerInitializationContext.java}      |  10 +-
 .../nifi/controller/ParameterProviderNode.java     |  13 +
 .../StandardConnectorConfigurationContext.java     |   2 +
 .../StandardConnectorInitializationContext.java    |   1 +
 .../connector/StandardConnectorNode.java           | 119 +++++++-
 ...StandardConnectorRepoInitializationContext.java |   9 +
 .../connector/StandardConnectorRepository.java     |   8 +
 .../apache/nifi/controller/ExtensionBuilder.java   |   7 +-
 .../org/apache/nifi/controller/FlowController.java |  43 ++-
 .../nifi/controller/flow/StandardFlowManager.java  |   4 +-
 .../TestStandardConnectorConfigurationContext.java |   1 +
 .../connector/TestStandardConnectorNode.java       |   1 +
 37 files changed, 1296 insertions(+), 101 deletions(-)

diff --git a/nifi-assembly/pom.xml b/nifi-assembly/pom.xml
index 82688c4ef63..81585f30cdc 100644
--- a/nifi-assembly/pom.xml
+++ b/nifi-assembly/pom.xml
@@ -471,12 +471,6 @@ language governing permissions and limitations under the 
License. -->
             <version>2.7.0-SNAPSHOT</version>
             <type>nar</type>
         </dependency>
-        <dependency>
-            <groupId>org.apache.nifi</groupId>
-            <artifactId>nifi-kafka-to-s3-nar</artifactId>
-            <version>2.7.0-SNAPSHOT</version>
-            <type>nar</type>
-        </dependency>
         <dependency>
             <groupId>org.apache.nifi</groupId>
             <artifactId>nifi-groovyx-nar</artifactId>
@@ -939,6 +933,12 @@ language governing permissions and limitations under the 
License. -->
             <version>2.7.0-SNAPSHOT</version>
             <type>nar</type>
         </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-kafka-to-s3-nar</artifactId>
+            <version>2.7.0-SNAPSHOT</version>
+            <type>nar</type>
+        </dependency>
         <!-- AspectJ library needed by the Java Agent used for native library 
loading (see bootstrap.conf) -->
         <dependency>
             <groupId>org.aspectj</groupId>
diff --git 
a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
 
b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
index 21e18ce91eb..ecfa644fd37 100644
--- 
a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
+++ 
b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
@@ -139,6 +139,9 @@ public class NiFiProperties extends ApplicationProperties {
     // Connector Repository properties
     public static final String CONNECTOR_REPOSITORY_IMPLEMENTATION = 
"nifi.components.connectors.repository.implementation";
 
+    // Secrets Manager properties
+    public static final String SECRETS_MANAGER_IMPLEMENTATION = 
"nifi.secrets.manager.implementation";
+
     // security properties
     public static final String SECURITY_KEYSTORE = "nifi.security.keystore";
     public static final String SECURITY_KEYSTORE_TYPE = 
"nifi.security.keystoreType";
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 8dc37302294..5fe172fa780 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
@@ -18,7 +18,9 @@
 package org.apache.nifi.mock.connector.server;
 
 import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.components.connector.ConnectorValueReference;
 import org.apache.nifi.components.connector.FlowUpdateException;
+import org.apache.nifi.components.connector.SecretReference;
 import org.apache.nifi.components.connector.StepConfiguration;
 
 import java.io.Closeable;
@@ -27,6 +29,8 @@ import java.util.List;
 import java.util.Map;
 
 public interface ConnectorTestRunner extends Closeable {
+    String SECRET_PROVIDER_ID = "TestRunnerSecretsManager";
+    String SECRET_PROVIDER_NAME = "TestRunnerSecretsManager";
 
     void applyUpdate() throws FlowUpdateException;
 
@@ -34,10 +38,18 @@ public interface ConnectorTestRunner extends Closeable {
 
     void configure(String stepName, Map<String, String> propertyValues) throws 
FlowUpdateException;
 
+    void configure(String stepName, Map<String, String> propertyValues, 
Map<String, ConnectorValueReference> propertyReferences) throws 
FlowUpdateException;
+
+    SecretReference createSecretReference(String secretName);
+
     ConnectorConfigVerificationResult verifyConfiguration(String stepName, 
Map<String, String> propertyValueOverrides);
 
+    ConnectorConfigVerificationResult verifyConfiguration(String stepName, 
Map<String, String> propertyValueOverrides, Map<String, 
ConnectorValueReference> referenceOverrides);
+
     ConnectorConfigVerificationResult verifyConfiguration(String stepName, 
StepConfiguration configurationOverrides);
 
+    void addSecret(String name, String value);
+
     void startConnector();
 
     void stopConnector();
diff --git 
a/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/MockConnectorInitializationContext.java
 
b/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/MockConnectorInitializationContext.java
index 939035bc5ec..93cb6fd44ab 100644
--- 
a/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/MockConnectorInitializationContext.java
+++ 
b/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/MockConnectorInitializationContext.java
@@ -22,7 +22,7 @@ import 
org.apache.nifi.components.connector.FrameworkConnectorInitializationCont
 import org.apache.nifi.components.connector.FlowUpdateException;
 import 
org.apache.nifi.components.connector.FrameworkConnectorInitializationContext;
 import org.apache.nifi.components.connector.FrameworkFlowContext;
-import org.apache.nifi.components.connector.SecretsManager;
+import org.apache.nifi.components.connector.secrets.SecretsManager;
 import org.apache.nifi.components.connector.components.FlowContext;
 import org.apache.nifi.flow.VersionedExternalFlow;
 import org.apache.nifi.flow.VersionedProcessGroup;
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 590e5a80490..0e5a2d86ab6 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
@@ -30,9 +30,11 @@ import 
org.apache.nifi.components.connector.ConnectorRepository;
 import org.apache.nifi.components.connector.ConnectorState;
 import org.apache.nifi.components.connector.ConnectorValueReference;
 import org.apache.nifi.components.connector.FlowUpdateException;
+import org.apache.nifi.components.connector.SecretReference;
 import 
org.apache.nifi.components.connector.StandaloneConnectorRequestReplicator;
 import org.apache.nifi.components.connector.StepConfiguration;
 import org.apache.nifi.components.connector.StringLiteralValue;
+import org.apache.nifi.components.connector.secrets.SecretsManager;
 import org.apache.nifi.components.state.StateManagerProvider;
 import org.apache.nifi.components.validation.DisabledServiceValidationResult;
 import org.apache.nifi.components.validation.ValidationState;
@@ -48,6 +50,7 @@ import org.apache.nifi.diagnostics.DiagnosticsFactory;
 import org.apache.nifi.encrypt.PropertyEncryptor;
 import org.apache.nifi.engine.FlowEngine;
 import org.apache.nifi.events.VolatileBulletinRepository;
+import 
org.apache.nifi.mock.connector.server.secrets.ConnectorTestRunnerSecretsManager;
 import org.apache.nifi.nar.ExtensionMapping;
 import org.apache.nifi.processor.Processor;
 import org.apache.nifi.reporting.BulletinRepository;
@@ -74,6 +77,7 @@ public class StandardConnectorMockServer implements 
ConnectorMockServer {
     private FlowController flowController;
     private MockExtensionDiscoveringManager extensionManager;
     private ConnectorNode connectorNode;
+    private ConnectorRepository connectorRepository;
     private FlowEngine flowEngine;
     private MockExtensionMapper mockExtensionMapper;
     private FlowFileTransferCounts initialFlowFileTransferCounts = new 
FlowFileTransferCounts(0L, 0L, 0L, 0L);
@@ -113,7 +117,7 @@ public class StandardConnectorMockServer implements 
ConnectorMockServer {
             throw new RuntimeException("Failed to initialize FlowFile 
Repository", e);
         }
 
-        final ConnectorRepository connectorRepository = 
flowController.getConnectorRepository();
+        connectorRepository = flowController.getConnectorRepository();
         if (!(connectorRepository instanceof MockConnectorRepository)) {
             throw new IllegalStateException("Connector Repository is not an 
instance of MockConnectorRepository");
         }
@@ -210,19 +214,32 @@ public class StandardConnectorMockServer implements 
ConnectorMockServer {
 
     @Override
     public void configure(final String stepName, final Map<String, String> 
propertyValues) throws FlowUpdateException {
-        final StepConfiguration configuration = 
toStringLiteralConfiguration(propertyValues);
-        configure(stepName, configuration);
+        configure(stepName, propertyValues, Collections.emptyMap());
+    }
+
+    @Override
+    public void configure(final String stepName, final Map<String, String> 
propertyValues, final Map<String, ConnectorValueReference> propertyReferences) 
throws FlowUpdateException {
+        final StepConfiguration stepConfiguration = 
createStepConfiguration(propertyValues, propertyReferences);
+        configure(stepName, stepConfiguration);
     }
 
     @Override
     public ConnectorConfigVerificationResult verifyConfiguration(final String 
stepName, final Map<String, String> propertyValueOverrides) {
-        final StepConfiguration configuration = 
toStringLiteralConfiguration(propertyValueOverrides);
+        return verifyConfiguration(stepName, propertyValueOverrides, 
Collections.emptyMap());
+    }
+
+    @Override
+    public ConnectorConfigVerificationResult verifyConfiguration(final String 
stepName, final Map<String, String> propertyValueOverrides,
+            final Map<String, ConnectorValueReference> referenceOverrides) {
+
+        final StepConfiguration configuration = 
createStepConfiguration(propertyValueOverrides, referenceOverrides);
         return verifyConfiguration(stepName, configuration);
     }
 
-    private StepConfiguration toStringLiteralConfiguration(final Map<String, 
String> propertyValues) {
-        final Map<String, ConnectorValueReference> references = new 
HashMap<>(propertyValues.size());
-        propertyValues.forEach((key, value) -> references.put(key, new 
StringLiteralValue(value)));
+    private StepConfiguration createStepConfiguration(final Map<String, 
String> propertyValues, final Map<String, ConnectorValueReference> 
propertyReferences) {
+        final Map<String, ConnectorValueReference> references = new 
HashMap<>();
+        propertyValues.forEach((key, value) -> references.put(key, value == 
null ? null : new StringLiteralValue(value)));
+        references.putAll(propertyReferences);
         return new StepConfiguration(references);
     }
 
@@ -232,6 +249,21 @@ public class StandardConnectorMockServer implements 
ConnectorMockServer {
         return new MockServerConfigVerificationResult(results);
     }
 
+    @Override
+    public void addSecret(final String name, final String value) {
+        final SecretsManager secretsManager = 
connectorRepository.getSecretsManager();
+        if (!(secretsManager instanceof final 
ConnectorTestRunnerSecretsManager testRunnerSecretsManager)) {
+            throw new IllegalStateException("Secrets Manager is not an 
instance of ConnectorTestRunnerSecretsManager");
+        }
+
+        testRunnerSecretsManager.addSecret(name, value);
+    }
+
+    @Override
+    public SecretReference createSecretReference(final String secretName) {
+        return new SecretReference(ConnectorTestRunner.SECRET_PROVIDER_ID, 
ConnectorTestRunner.SECRET_PROVIDER_NAME, secretName);
+    }
+
     @Override
     public void startConnector() {
         initialFlowFileTransferCounts = 
connectorNode.getFlowFileTransferCounts();
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
new file mode 100644
index 00000000000..16e05ff4f0b
--- /dev/null
+++ 
b/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/secrets/ConnectorTestRunnerSecretProvider.java
@@ -0,0 +1,86 @@
+/*
+ * 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.secrets.SecretProvider;
+import org.apache.nifi.components.connector.secrets.StandardSecret;
+import org.apache.nifi.mock.connector.server.ConnectorTestRunner;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class ConnectorTestRunnerSecretProvider implements SecretProvider {
+    public static final String GROUP_NAME = "Default";
+
+    private final Map<String, String> secrets = new HashMap<>();
+
+    public void addSecret(final String key, final String value) {
+        this.secrets.put(key, value);
+    }
+
+    @Override
+    public String getProviderId() {
+        return ConnectorTestRunner.SECRET_PROVIDER_ID;
+    }
+
+    @Override
+    public String getProviderName() {
+        return ConnectorTestRunner.SECRET_PROVIDER_ID;
+    }
+
+    @Override
+    public List<Secret> getAllSecrets() {
+        final List<Secret> secrets = new ArrayList<>();
+        for (final Map.Entry<String, String> entry : this.secrets.entrySet()) {
+            final Secret secret = new StandardSecret.Builder()
+                .providerName(ConnectorTestRunner.SECRET_PROVIDER_NAME)
+                .groupName(GROUP_NAME)
+                .name(entry.getKey())
+                .value(entry.getValue())
+                .build();
+
+            secrets.add(secret);
+        }
+
+        return secrets;
+    }
+
+    @Override
+    public List<Secret> getSecrets(final List<String> 
fullyQualifiedSecretNames) {
+        final List<Secret> matchingSecrets = new ArrayList<>();
+
+        for (final String secretName : fullyQualifiedSecretNames) {
+            final String value = secrets.get(secretName);
+
+            if (value != null) {
+                final Secret secret = new StandardSecret.Builder()
+                    .providerName(ConnectorTestRunner.SECRET_PROVIDER_NAME)
+                    .groupName(GROUP_NAME)
+                    .name(secretName)
+                    .value(value)
+                    .build();
+
+                matchingSecrets.add(secret);
+            }
+        }
+        return matchingSecrets;
+    }
+}
diff --git 
a/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/secrets/ConnectorTestRunnerSecretsManager.java
 
b/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/secrets/ConnectorTestRunnerSecretsManager.java
new file mode 100644
index 00000000000..8af0f6a8b9e
--- /dev/null
+++ 
b/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/java/org/apache/nifi/mock/connector/server/secrets/ConnectorTestRunnerSecretsManager.java
@@ -0,0 +1,99 @@
+/*
+ * 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.components.connector.secrets.SecretProvider;
+import org.apache.nifi.components.connector.secrets.SecretsManager;
+import 
org.apache.nifi.components.connector.secrets.SecretsManagerInitializationContext;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+public class ConnectorTestRunnerSecretsManager implements SecretsManager {
+    private final ConnectorTestRunnerSecretProvider secretProvider = new 
ConnectorTestRunnerSecretProvider();
+
+    @Override
+    public void initialize(final SecretsManagerInitializationContext 
initializationContext) {
+    }
+
+    public void addSecret(final String name, final String value) {
+        secretProvider.addSecret(name, value);
+    }
+
+    @Override
+    public List<Secret> getAllSecrets() {
+        return secretProvider.getAllSecrets();
+    }
+
+    @Override
+    public Set<SecretProvider> getSecretProviders() {
+        return Set.of(secretProvider);
+    }
+
+    @Override
+    public Optional<Secret> getSecret(final SecretReference secretReference) {
+        // Check that appropriate provider given
+        final SecretProvider provider = getProvider(secretReference);
+        if (provider == null) {
+            return Optional.empty();
+        }
+
+        final List<Secret> secrets = 
provider.getSecrets(List.of(secretReference.getSecretName()));
+        return secrets.isEmpty() ? Optional.empty() : 
Optional.of(secrets.getFirst());
+    }
+
+    private SecretProvider getProvider(final SecretReference secretReference) {
+        final String providerId = secretReference.getProviderId();
+        if (providerId != null) {
+            for (final SecretProvider provider : getSecretProviders()) {
+                if (provider.getProviderId().equals(providerId)) {
+                    return provider;
+                }
+            }
+
+            return null;
+        }
+
+        final String providerName = secretReference.getProviderName();
+        if (providerName != null) {
+            for (final SecretProvider provider : getSecretProviders()) {
+                if (provider.getProviderName().equals(providerName)) {
+                    return provider;
+                }
+            }
+        }
+
+        return null;
+    }
+
+    @Override
+    public Map<SecretReference, Secret> getSecrets(final Set<SecretReference> 
secretReferences) {
+        final Map<SecretReference, Secret> secrets = new HashMap<>();
+        for (final SecretReference reference : secretReferences) {
+            final Secret secret = getSecret(reference).orElse(null);
+            secrets.put(reference, secret);
+        }
+
+        return secrets;
+    }
+}
diff --git 
a/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/resources/META-INF/services/org.apache.nifi.components.connector.secrets.SecretsManager
 
b/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/resources/META-INF/services/org.apache.nifi.components.connector.secrets.SecretsManager
new file mode 100644
index 00000000000..a1c0de54d1f
--- /dev/null
+++ 
b/nifi-connector-mock-bundle/nifi-connector-mock-server/src/main/resources/META-INF/services/org.apache.nifi.components.connector.secrets.SecretsManager
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.nifi.mock.connector.server.secrets.ConnectorTestRunnerSecretsManager
diff --git 
a/nifi-connector-mock-bundle/nifi-connector-mock/src/main/java/org/apache/nifi/mock/connector/StandardConnectorTestRunner.java
 
b/nifi-connector-mock-bundle/nifi-connector-mock/src/main/java/org/apache/nifi/mock/connector/StandardConnectorTestRunner.java
index 5cf5d5ef709..71636a5af1a 100644
--- 
a/nifi-connector-mock-bundle/nifi-connector-mock/src/main/java/org/apache/nifi/mock/connector/StandardConnectorTestRunner.java
+++ 
b/nifi-connector-mock-bundle/nifi-connector-mock/src/main/java/org/apache/nifi/mock/connector/StandardConnectorTestRunner.java
@@ -20,7 +20,9 @@ package org.apache.nifi.mock.connector;
 import org.apache.nifi.NiFiServer;
 import org.apache.nifi.bundle.Bundle;
 import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.components.connector.ConnectorValueReference;
 import org.apache.nifi.components.connector.FlowUpdateException;
+import org.apache.nifi.components.connector.SecretReference;
 import org.apache.nifi.components.connector.StepConfiguration;
 import org.apache.nifi.mock.connector.server.ConnectorConfigVerificationResult;
 import org.apache.nifi.mock.connector.server.ConnectorMockServer;
@@ -131,16 +133,38 @@ public class StandardConnectorTestRunner implements 
ConnectorTestRunner, Closeab
         mockServer.configure(stepName, propertyValues);
     }
 
+    @Override
+    public void configure(final String stepName, final Map<String, String> 
propertyValues, final Map<String, ConnectorValueReference> propertyReferences) 
throws FlowUpdateException {
+        mockServer.configure(stepName, propertyValues, propertyReferences);
+    }
+
+    @Override
+    public SecretReference createSecretReference(final String secretName) {
+        return mockServer.createSecretReference(secretName);
+    }
+
     @Override
     public ConnectorConfigVerificationResult verifyConfiguration(final String 
stepName, final Map<String, String> propertyValueOverrides) {
         return mockServer.verifyConfiguration(stepName, 
propertyValueOverrides);
     }
 
+    @Override
+    public ConnectorConfigVerificationResult verifyConfiguration(final String 
stepName, final Map<String, String> propertyValueOverrides,
+            final Map<String, ConnectorValueReference> referenceOverrides) {
+
+        return mockServer.verifyConfiguration(stepName, 
propertyValueOverrides, referenceOverrides);
+    }
+
     @Override
     public ConnectorConfigVerificationResult verifyConfiguration(final String 
stepName, final StepConfiguration configurationOverrides) {
         return mockServer.verifyConfiguration(stepName, 
configurationOverrides);
     }
 
+    @Override
+    public void addSecret(final String name, final String value) {
+        mockServer.addSecret(name, value);
+    }
+
     @Override
     public void startConnector() {
         mockServer.startConnector();
diff --git 
a/nifi-connector-mock-bundle/nifi-connector-mock/src/main/resources/nifi.properties
 
b/nifi-connector-mock-bundle/nifi-connector-mock/src/main/resources/nifi.properties
index 046a2980b21..071f1cca3c6 100644
--- 
a/nifi-connector-mock-bundle/nifi-connector-mock/src/main/resources/nifi.properties
+++ 
b/nifi-connector-mock-bundle/nifi-connector-mock/src/main/resources/nifi.properties
@@ -94,6 +94,9 @@ 
nifi.nar.persistence.provider.properties.directory=target/nifi-storage/nar_repos
 # Asset Management
 nifi.asset.manager.properties.directory=target/nifi-storage/assets
 
+# Secrets Manager
+nifi.secrets.manager.implementation=org.apache.nifi.mock.connector.server.secrets.ConnectorTestRunnerSecretsManager
+
 # Site to Site properties
 nifi.remote.input.host=localhost
 nifi.remote.input.secure=true
diff --git 
a/nifi-connectors/nifi-kafka-to-s3-bundle/nifi-kafka-to-s3-integration-tests/src/test/java/org/apache/nifi/connectors/kafkas3/KafkaToS3IT.java
 
b/nifi-connectors/nifi-kafka-to-s3-bundle/nifi-kafka-to-s3-integration-tests/src/test/java/org/apache/nifi/connectors/kafkas3/KafkaToS3IT.java
index 6ccdc34fd9d..f715ea2e4e7 100644
--- 
a/nifi-connectors/nifi-kafka-to-s3-bundle/nifi-kafka-to-s3-integration-tests/src/test/java/org/apache/nifi/connectors/kafkas3/KafkaToS3IT.java
+++ 
b/nifi-connectors/nifi-kafka-to-s3-bundle/nifi-kafka-to-s3-integration-tests/src/test/java/org/apache/nifi/connectors/kafkas3/KafkaToS3IT.java
@@ -32,6 +32,7 @@ import org.apache.kafka.common.config.SaslConfigs;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.apache.nifi.components.ConfigVerificationResult;
 import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.components.connector.ConnectorValueReference;
 import org.apache.nifi.components.connector.FlowUpdateException;
 import org.apache.nifi.mock.connector.StandardConnectorTestRunner;
 import org.apache.nifi.mock.connector.server.ConnectorConfigVerificationResult;
@@ -299,18 +300,19 @@ public class KafkaToS3IT {
             "Kafka Brokers", "localhost:9093",
             "Security Protocol", "SASL_PLAINTEXT",
             "SASL Mechanism", "PLAIN",
-            "Username", SCRAM_USERNAME,
-            "Password", SCRAM_PASSWORD
+            "Username", SCRAM_USERNAME
+        );
+        runner.addSecret("kafka-password", SCRAM_PASSWORD);
+        final Map<String, ConnectorValueReference> kafkaServerSecretRefs = 
Map.of(
+            "Password", runner.createSecretReference("kafka-password")
         );
-
-        runner.applyUpdate();
 
         // Perform verification to ensure that valid server configuration 
passes
-        final ConnectorConfigVerificationResult connectionVerificationResults 
= runner.verifyConfiguration("Kafka Connection", kafkaServerConfig);
+        final ConnectorConfigVerificationResult connectionVerificationResults 
= runner.verifyConfiguration("Kafka Connection", kafkaServerConfig, 
kafkaServerSecretRefs);
         connectionVerificationResults.assertNoFailures();
 
         // Apply the configuration that we've now validated
-        runner.configure("Kafka Connection", kafkaServerConfig);
+        runner.configure("Kafka Connection", kafkaServerConfig, 
kafkaServerSecretRefs);
 
         // Perform verification to ensure that valid topic configuration passes
         final Map<String, String> topic1Config = Map.of(
@@ -360,8 +362,11 @@ public class KafkaToS3IT {
             "Kafka Brokers", "localhost:9093",
             "Security Protocol", "SASL_PLAINTEXT",
             "SASL Mechanism", "PLAIN",
-            "Username", SCRAM_USERNAME,
-            "Password", SCRAM_PASSWORD
+            "Username", SCRAM_USERNAME
+        );
+        runner.addSecret("kafka-password", SCRAM_PASSWORD);
+        final Map<String, ConnectorValueReference> kafkaSecretRefs = Map.of(
+            "Password", runner.createSecretReference("kafka-password")
         );
 
         final Map<String, String> kafkaTopicConfig = Map.of(
@@ -378,14 +383,17 @@ public class KafkaToS3IT {
             Map.entry("S3 Endpoint Override URL", 
localStackContainer.getEndpoint().toString()),
             Map.entry("S3 Authentication Strategy", "Access Key ID and Secret 
Key"),
             Map.entry("S3 Access Key ID", localStackContainer.getAccessKey()),
-            Map.entry("S3 Secret Access Key", 
localStackContainer.getSecretKey()),
             Map.entry("Target Object Size", "1 MB"),
             Map.entry("Merge Latency", "1 sec")
         );
+        runner.addSecret("s3-secret-key", localStackContainer.getSecretKey());
+        final Map<String, ConnectorValueReference> s3SecretRefs = Map.of(
+            "S3 Secret Access Key", 
runner.createSecretReference("s3-secret-key")
+        );
 
-        runner.configure("Kafka Connection", kafkaServerConfig);
+        runner.configure("Kafka Connection", kafkaServerConfig, 
kafkaSecretRefs);
         runner.configure("Kafka Topics", kafkaTopicConfig);
-        runner.configure("S3 Configuration", s3Config);
+        runner.configure("S3 Configuration", s3Config, s3SecretRefs);
         runner.applyUpdate();
 
         final List<ValidationResult> validationResults = runner.validate();
@@ -449,14 +457,17 @@ public class KafkaToS3IT {
             "Security Protocol", "SASL_PLAINTEXT",
             "SASL Mechanism", "PLAIN",
             "Username", SCRAM_USERNAME,
-            "Password", SCRAM_PASSWORD,
             "Schema Registry URL", getSchemaRegistryUrl()
         );
+        runner.addSecret("kafka-password", SCRAM_PASSWORD);
+        final Map<String, ConnectorValueReference> kafkaSecretRefs = Map.of(
+            "Password", runner.createSecretReference("kafka-password")
+        );
 
-        final ConnectorConfigVerificationResult connectionVerificationResults 
= runner.verifyConfiguration("Kafka Connection", kafkaConnectionConfig);
+        final ConnectorConfigVerificationResult connectionVerificationResults 
= runner.verifyConfiguration("Kafka Connection", kafkaConnectionConfig, 
kafkaSecretRefs);
         connectionVerificationResults.assertNoFailures();
 
-        runner.configure("Kafka Connection", kafkaConnectionConfig);
+        runner.configure("Kafka Connection", kafkaConnectionConfig, 
kafkaSecretRefs);
 
         final Map<String, String> avroTopicConfig = Map.of(
             "Topic Names", "avro-topic",
@@ -517,9 +528,12 @@ public class KafkaToS3IT {
             "Security Protocol", "SASL_PLAINTEXT",
             "SASL Mechanism", "PLAIN",
             "Username", SCRAM_USERNAME,
-            "Password", SCRAM_PASSWORD,
             "Schema Registry URL", getSchemaRegistryUrl()
         );
+        runner.addSecret("kafka-password", SCRAM_PASSWORD);
+        final Map<String, ConnectorValueReference> kafkaSecretRefs = Map.of(
+            "Password", runner.createSecretReference("kafka-password")
+        );
 
         final Map<String, String> kafkaTopicConfig = Map.of(
             "Topic Names", "user-events",
@@ -535,14 +549,17 @@ public class KafkaToS3IT {
             Map.entry("S3 Endpoint Override URL", 
localStackContainer.getEndpoint().toString()),
             Map.entry("S3 Authentication Strategy", "Access Key ID and Secret 
Key"),
             Map.entry("S3 Access Key ID", localStackContainer.getAccessKey()),
-            Map.entry("S3 Secret Access Key", 
localStackContainer.getSecretKey()),
             Map.entry("Target Object Size", "1 MB"),
             Map.entry("Merge Latency", "1 sec")
         );
+        runner.addSecret("s3-secret-key", localStackContainer.getSecretKey());
+        final Map<String, ConnectorValueReference> s3SecretRefs = Map.of(
+            "S3 Secret Access Key", 
runner.createSecretReference("s3-secret-key")
+        );
 
-        runner.configure("Kafka Connection", kafkaConnectionConfig);
+        runner.configure("Kafka Connection", kafkaConnectionConfig, 
kafkaSecretRefs);
         runner.configure("Kafka Topics", kafkaTopicConfig);
-        runner.configure("S3 Configuration", s3Config);
+        runner.configure("S3 Configuration", s3Config, s3SecretRefs);
         runner.applyUpdate();
 
         final List<ValidationResult> validationResults = runner.validate();
@@ -613,8 +630,11 @@ public class KafkaToS3IT {
             "Kafka Brokers", "localhost:9093",
             "Security Protocol", "SASL_PLAINTEXT",
             "SASL Mechanism", "PLAIN",
-            "Username", SCRAM_USERNAME,
-            "Password", SCRAM_PASSWORD
+            "Username", SCRAM_USERNAME
+        );
+        runner.addSecret("kafka-password", SCRAM_PASSWORD);
+        final Map<String, ConnectorValueReference> kafkaSecretRefs = Map.of(
+            "Password", runner.createSecretReference("kafka-password")
         );
 
         final Map<String, String> jsonTopicConfig = Map.of(
@@ -631,14 +651,17 @@ public class KafkaToS3IT {
             Map.entry("S3 Endpoint Override URL", 
"http://invalid-s3-endpoint:9999";),
             Map.entry("S3 Authentication Strategy", "Access Key ID and Secret 
Key"),
             Map.entry("S3 Access Key ID", localStackContainer.getAccessKey()),
-            Map.entry("S3 Secret Access Key", 
localStackContainer.getSecretKey()),
             Map.entry("Target Object Size", "1 MB"),
             Map.entry("Merge Latency", "1 sec")
         );
+        runner.addSecret("s3-secret-key", localStackContainer.getSecretKey());
+        final Map<String, ConnectorValueReference> s3SecretRefs = Map.of(
+            "S3 Secret Access Key", 
runner.createSecretReference("s3-secret-key")
+        );
 
-        runner.configure("Kafka Connection", kafkaServerConfig);
+        runner.configure("Kafka Connection", kafkaServerConfig, 
kafkaSecretRefs);
         runner.configure("Kafka Topics", jsonTopicConfig);
-        runner.configure("S3 Configuration", s3InvalidConfig);
+        runner.configure("S3 Configuration", s3InvalidConfig, s3SecretRefs);
         runner.applyUpdate();
 
         // Run the Connector with the invalid S3 endpoint to queue the JSON 
data. Wait for data to be queued up.
@@ -658,12 +681,11 @@ public class KafkaToS3IT {
             Map.entry("S3 Endpoint Override URL", 
localStackContainer.getEndpoint().toString()),
             Map.entry("S3 Authentication Strategy", "Access Key ID and Secret 
Key"),
             Map.entry("S3 Access Key ID", localStackContainer.getAccessKey()),
-            Map.entry("S3 Secret Access Key", 
localStackContainer.getSecretKey()),
             Map.entry("Target Object Size", "1 MB"),
             Map.entry("Merge Latency", "1 sec")
         );
 
-        runner.configure("S3 Configuration", s3ValidJsonConfig);
+        runner.configure("S3 Configuration", s3ValidJsonConfig, s3SecretRefs);
         runner.applyUpdate();
 
         // Make sure there is no data in S3 yet.
@@ -679,7 +701,6 @@ public class KafkaToS3IT {
             Map.entry("S3 Endpoint Override URL", 
localStackContainer.getEndpoint().toString()),
             Map.entry("S3 Authentication Strategy", "Access Key ID and Secret 
Key"),
             Map.entry("S3 Access Key ID", localStackContainer.getAccessKey()),
-            Map.entry("S3 Secret Access Key", 
localStackContainer.getSecretKey()),
             Map.entry("Target Object Size", "1 MB"),
             Map.entry("Merge Latency", "1 sec")
         );
@@ -697,13 +718,12 @@ public class KafkaToS3IT {
             "Security Protocol", "SASL_PLAINTEXT",
             "SASL Mechanism", "PLAIN",
             "Username", SCRAM_USERNAME,
-            "Password", SCRAM_PASSWORD,
             "Schema Registry URL", getSchemaRegistryUrl()
         );
 
-        runner.configure("Kafka Connection", 
kafkaConnectionWithSchemaRegistry);
+        runner.configure("Kafka Connection", 
kafkaConnectionWithSchemaRegistry, kafkaSecretRefs);
         runner.configure("Kafka Topics", avroTopicConfig);
-        runner.configure("S3 Configuration", s3ValidAvroConfig);
+        runner.configure("S3 Configuration", s3ValidAvroConfig, s3SecretRefs);
         runner.applyUpdate();
 
         // After draining, there should be one JSON file in S3.
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/components/connector/secrets/ParameterProviderSecretProvider.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/components/connector/secrets/ParameterProviderSecretProvider.java
new file mode 100644
index 00000000000..2f058f3c2a0
--- /dev/null
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/components/connector/secrets/ParameterProviderSecretProvider.java
@@ -0,0 +1,86 @@
+/*
+ * 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.secrets;
+
+import org.apache.nifi.components.connector.Secret;
+import org.apache.nifi.controller.ParameterProviderNode;
+import org.apache.nifi.parameter.Parameter;
+import org.apache.nifi.parameter.ParameterDescriptor;
+import org.apache.nifi.parameter.ParameterGroup;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class ParameterProviderSecretProvider implements SecretProvider {
+    private final ParameterProviderNode parameterProvider;
+
+    public ParameterProviderSecretProvider(final ParameterProviderNode 
parameterProvider) {
+        this.parameterProvider = parameterProvider;
+    }
+
+    @Override
+    public String getProviderId() {
+        return parameterProvider.getIdentifier();
+    }
+
+    @Override
+    public String getProviderName() {
+        return parameterProvider.getName();
+    }
+
+    @Override
+    public List<Secret> getAllSecrets() {
+        final List<Secret> secrets = new ArrayList<>();
+
+        final List<ParameterGroup> parameterGroups = 
parameterProvider.fetchParameterValues();
+        for (final ParameterGroup group : parameterGroups) {
+            for (final Parameter parameter : group.getParameters()) {
+                final Secret secret = createSecret(group.getGroupName(), 
parameter);
+                secrets.add(secret);
+            }
+        }
+
+        return secrets;
+    }
+
+    private Secret createSecret(final String groupName, final Parameter 
parameter) {
+        final ParameterDescriptor descriptor = parameter.getDescriptor();
+
+        return new StandardSecret.Builder()
+            .providerName(getProviderName())
+            .groupName(groupName)
+            .name(descriptor.getName())
+            .description(descriptor.getDescription())
+            .value(parameter.getValue())
+            .build();
+    }
+
+    @Override
+    public List<Secret> getSecrets(final List<String> 
fullyQualifiedSecretNames) {
+        final List<ParameterGroup> parameterGroups = 
parameterProvider.fetchParameterValues(fullyQualifiedSecretNames);
+        final List<Secret> secrets = new ArrayList<>();
+        for (final ParameterGroup group : parameterGroups) {
+            for (final Parameter parameter : group.getParameters()) {
+                final Secret secret = createSecret(group.getGroupName(), 
parameter);
+                secrets.add(secret);
+            }
+        }
+
+        return secrets;
+    }
+}
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/components/connector/secrets/ParameterProviderSecretsManager.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/components/connector/secrets/ParameterProviderSecretsManager.java
new file mode 100644
index 00000000000..50919c8b886
--- /dev/null
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/components/connector/secrets/ParameterProviderSecretsManager.java
@@ -0,0 +1,151 @@
+/*
+ * 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.secrets;
+
+import org.apache.nifi.components.connector.Secret;
+import org.apache.nifi.components.connector.SecretReference;
+import org.apache.nifi.controller.ParameterProviderNode;
+import org.apache.nifi.controller.flow.FlowManager;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public class ParameterProviderSecretsManager implements SecretsManager {
+    private FlowManager flowManager;
+
+    @Override
+    public void initialize(final SecretsManagerInitializationContext 
initializationContext) {
+        this.flowManager = initializationContext.getFlowManager();
+    }
+
+    @Override
+    public List<Secret> getAllSecrets() {
+        final List<Secret> secrets = new ArrayList<>();
+        for (final SecretProvider provider : getSecretProviders()) {
+            secrets.addAll(provider.getAllSecrets());
+        }
+
+        // Sort secrets by Provider Name, then Group Name, then Secret Name
+        secrets.sort(Comparator.comparing(Secret::getProviderName)
+            .thenComparing(Secret::getGroupName)
+            .thenComparing(Secret::getName));
+
+        return secrets;
+    }
+
+    @Override
+    public Set<SecretProvider> getSecretProviders() {
+        final Set<SecretProvider> providers = new HashSet<>();
+        for (final ParameterProviderNode parameterProviderNode : 
flowManager.getAllParameterProviders()) {
+            providers.add(new 
ParameterProviderSecretProvider(parameterProviderNode));
+        }
+
+        return providers;
+    }
+
+    @Override
+    public Optional<Secret> getSecret(final SecretReference secretReference) {
+        final SecretProvider provider = findProvider(secretReference);
+        if (provider == null) {
+            return Optional.empty();
+        }
+
+        final List<Secret> secrets = 
provider.getSecrets(List.of(secretReference.getSecretName()));
+        if (secrets.isEmpty()) {
+            return Optional.empty();
+        }
+
+        return Optional.of(secrets.getFirst());
+    }
+
+    @Override
+    public Map<SecretReference, Secret> getSecrets(final Set<SecretReference> 
secretReferences) {
+        // Partition secret references by Provider
+        final Map<SecretProvider, Set<SecretReference>> referencesByProvider = 
new HashMap<>();
+        for (final SecretReference secretReference : secretReferences) {
+            final SecretProvider provider = findProvider(secretReference);
+            referencesByProvider.computeIfAbsent(provider, k -> new 
HashSet<>()).add(secretReference);
+        }
+
+        final Map<SecretReference, Secret> secrets = new HashMap<>();
+        for (final Map.Entry<SecretProvider, Set<SecretReference>> entry : 
referencesByProvider.entrySet()) {
+            final SecretProvider provider = entry.getKey();
+            final Set<SecretReference> references = entry.getValue();
+
+            // If no provider found, be sure to map to a null Secret rather 
than skipping
+            if (provider == null) {
+                for (final SecretReference secretReference : references) {
+                    secrets.put(secretReference, null);
+                }
+
+                continue;
+            }
+
+            final List<String> secretNames = new ArrayList<>();
+            references.forEach(ref -> secretNames.add(ref.getSecretName()));
+            final List<Secret> retrievedSecrets = 
provider.getSecrets(secretNames);
+            final Map<String, Secret> secretsByName = retrievedSecrets.stream()
+                .collect(Collectors.toMap(Secret::getName, 
Function.identity()));
+
+            for (final SecretReference secretReference : references) {
+                final Secret secret = 
secretsByName.get(secretReference.getSecretName());
+                secrets.put(secretReference, secret);
+            }
+        }
+
+        return secrets;
+    }
+
+    private SecretProvider findProvider(final SecretReference secretReference) 
{
+        final Set<SecretProvider> providers = getSecretProviders();
+
+        // Search first by Provider ID, if it's provided.
+        final String providerId = secretReference.getProviderId();
+        if (providerId != null) {
+            for (final SecretProvider provider : providers) {
+                if (providerId.equals(provider.getProviderId())) {
+                    return provider;
+                }
+            }
+
+            // If ID is provided but doesn't match, do not consider name.
+            return null;
+        }
+
+        // No Provider found by ID, search by Provider Name
+        final String providerName = secretReference.getProviderName();
+        if (providerName != null) {
+            for (final SecretProvider provider : providers) {
+                if (providerName.equals(provider.getProviderName())) {
+                    return provider;
+                }
+            }
+        }
+
+        // No Provider found
+        return null;
+    }
+}
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/components/connector/secrets/StandardSecret.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/components/connector/secrets/StandardSecret.java
new file mode 100644
index 00000000000..443b19cd06f
--- /dev/null
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/components/connector/secrets/StandardSecret.java
@@ -0,0 +1,127 @@
+/*
+ * 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.secrets;
+
+import org.apache.nifi.components.connector.Secret;
+
+import java.util.Objects;
+
+public class StandardSecret implements Secret {
+    private final String providerName;
+    private final String groupName;
+    private final String name;
+    private final String description;
+    private final String value;
+
+    private StandardSecret(final Builder builder) {
+        this.providerName = builder.providerName;
+        this.groupName = builder.groupName;
+        this.name = builder.name;
+        this.description = builder.description;
+        this.value = builder.value;
+    }
+
+    @Override
+    public String getProviderName() {
+        return providerName;
+    }
+
+    @Override
+    public String getGroupName() {
+        return groupName;
+    }
+
+    @Override
+    public String getName() {
+        return name;
+    }
+
+    @Override
+    public String getDescription() {
+        return description;
+    }
+
+    @Override
+    public String getValue() {
+        return value;
+    }
+
+    @Override
+    public String toString() {
+        return "StandardSecret[providerName=%s, groupName=%s, name=%s, 
description=%s]".formatted(
+             providerName, groupName, name, description);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(providerName, groupName, name, description);
+    }
+
+    @Override
+    public boolean equals(final Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (obj == null || getClass() != obj.getClass()) {
+            return false;
+        }
+        final StandardSecret other = (StandardSecret) obj;
+        return Objects.equals(this.providerName, other.providerName)
+            && Objects.equals(this.groupName, other.groupName)
+            &&  Objects.equals(this.name, other.name)
+            &&  Objects.equals(this.description, other.description);
+    }
+
+    public static class Builder {
+        private String providerName;
+        private String groupName;
+        private String name;
+        private String description;
+        private String value;
+
+        public Builder providerName(String providerName) {
+            this.providerName = providerName;
+            return this;
+        }
+
+        public Builder groupName(String groupName) {
+            this.groupName = groupName;
+            return this;
+        }
+
+        public Builder name(String name) {
+            this.name = name;
+            return this;
+        }
+
+        public Builder description(String description) {
+            this.description = description;
+            return this;
+        }
+
+        public Builder value(String value) {
+            this.value = value;
+            return this;
+        }
+
+        public StandardSecret build() {
+            return new StandardSecret(this);
+        }
+    }
+
+}
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepositoryInitializationContext.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/components/connector/secrets/StandardSecretsManagerInitializationContext.java
similarity index 66%
copy from 
nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepositoryInitializationContext.java
copy to 
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/components/connector/secrets/StandardSecretsManagerInitializationContext.java
index 5b0b923ac7b..ae5013768a8 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepositoryInitializationContext.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/components/connector/secrets/StandardSecretsManagerInitializationContext.java
@@ -15,20 +15,19 @@
  * limitations under the License.
  */
 
-package org.apache.nifi.components.connector;
+package org.apache.nifi.components.connector.secrets;
 
-import org.apache.nifi.controller.NodeTypeProvider;
 import org.apache.nifi.controller.flow.FlowManager;
-import org.apache.nifi.nar.ExtensionManager;
 
-public interface ConnectorRepositoryInitializationContext {
+public class StandardSecretsManagerInitializationContext implements 
SecretsManagerInitializationContext {
+    private final FlowManager flowManager;
 
-    FlowManager getFlowManager();
-
-    ExtensionManager getExtensionManager();
-
-    NodeTypeProvider getNodeTypeProvider();
-
-    ConnectorRequestReplicator getRequestReplicator();
+    public StandardSecretsManagerInitializationContext(final FlowManager 
flowManager) {
+        this.flowManager = flowManager;
+    }
 
+    @Override
+    public FlowManager getFlowManager() {
+        return flowManager;
+    }
 }
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/parameter/StandardParameterProviderNode.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/parameter/StandardParameterProviderNode.java
index cf07875660f..2d17556c29e 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/parameter/StandardParameterProviderNode.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/parameter/StandardParameterProviderNode.java
@@ -276,16 +276,32 @@ public class StandardParameterProviderNode extends 
AbstractComponentNode impleme
     }
 
     @Override
-    public void fetchParameters() {
+    public List<ParameterGroup> fetchParameterValues() {
         final ParameterProvider parameterProvider = 
parameterProviderRef.get().getParameterProvider();
         final ConfigurationContext configurationContext = 
getConfigurationContext();
-        List<ParameterGroup> fetchedParameterGroups;
+
         try (final NarCloseable ignored = 
NarCloseable.withComponentNarLoader(getExtensionManager(), 
parameterProvider.getClass(), parameterProvider.getIdentifier())) {
-            fetchedParameterGroups = 
parameterProvider.fetchParameters(configurationContext);
+            return parameterProvider.fetchParameters(configurationContext);
         } catch (final IOException | RuntimeException e) {
             throw new IllegalStateException(String.format("Error fetching 
parameters for %s: %s", this, e.getMessage()), e);
         }
+    }
+
+    @Override
+    public List<ParameterGroup> fetchParameterValues(final List<String> 
fullyQualifiedParameterNames) {
+        final ParameterProvider parameterProvider = 
parameterProviderRef.get().getParameterProvider();
+        final ConfigurationContext configurationContext = 
getConfigurationContext();
 
+        try (final NarCloseable ignored = 
NarCloseable.withComponentNarLoader(getExtensionManager(), 
parameterProvider.getClass(), parameterProvider.getIdentifier())) {
+            return parameterProvider.fetchParameters(configurationContext, 
fullyQualifiedParameterNames);
+        } catch (final IOException | RuntimeException e) {
+            throw new IllegalStateException(String.format("Error fetching 
parameters for %s: %s", this, e.getMessage()), e);
+        }
+    }
+
+    @Override
+    public void fetchParameters() {
+        final List<ParameterGroup> fetchedParameterGroups = 
fetchParameterValues();
         if (fetchedParameterGroups == null || 
fetchedParameterGroups.isEmpty()) {
             return;
         }
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/resources/META-INF/services/org.apache.nifi.components.connector.secrets.SecretsManager
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/resources/META-INF/services/org.apache.nifi.components.connector.secrets.SecretsManager
new file mode 100644
index 00000000000..082fffbd8f1
--- /dev/null
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/resources/META-INF/services/org.apache.nifi.components.connector.secrets.SecretsManager
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.nifi.components.connector.secrets.ParameterProviderSecretsManager
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/components/connector/secrets/TestParameterProviderSecretsManager.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/components/connector/secrets/TestParameterProviderSecretsManager.java
new file mode 100644
index 00000000000..51a50973ef0
--- /dev/null
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/components/connector/secrets/TestParameterProviderSecretsManager.java
@@ -0,0 +1,318 @@
+/*
+ * 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.secrets;
+
+import org.apache.nifi.components.connector.Secret;
+import org.apache.nifi.components.connector.SecretReference;
+import org.apache.nifi.controller.ParameterProviderNode;
+import org.apache.nifi.controller.flow.FlowManager;
+import org.apache.nifi.parameter.Parameter;
+import org.apache.nifi.parameter.ParameterDescriptor;
+import org.apache.nifi.parameter.ParameterGroup;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+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;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class TestParameterProviderSecretsManager {
+
+    private static final String PROVIDER_1_ID = "provider-1-id";
+    private static final String PROVIDER_1_NAME = "Provider One";
+    private static final String PROVIDER_2_ID = "provider-2-id";
+    private static final String PROVIDER_2_NAME = "Provider Two";
+
+    private static final String GROUP_1_NAME = "Group One";
+    private static final String GROUP_2_NAME = "Group Two";
+
+    private static final String SECRET_1_NAME = "secret-one";
+    private static final String SECRET_1_DESCRIPTION = "First secret";
+    private static final String SECRET_1_VALUE = "secret-value-one";
+
+    private static final String SECRET_2_NAME = "secret-two";
+    private static final String SECRET_2_DESCRIPTION = "Second secret";
+    private static final String SECRET_2_VALUE = "secret-value-two";
+
+    private static final String SECRET_3_NAME = "secret-three";
+    private static final String SECRET_3_DESCRIPTION = "Third secret";
+    private static final String SECRET_3_VALUE = "secret-value-three";
+
+    private ParameterProviderSecretsManager secretsManager;
+
+    @BeforeEach
+    public void setup() {
+        final FlowManager flowManager = mock(FlowManager.class);
+        final ParameterProviderNode providerNode1 = 
createMockedParameterProviderNode(PROVIDER_1_ID, PROVIDER_1_NAME, GROUP_1_NAME,
+            createParameter(SECRET_1_NAME, SECRET_1_DESCRIPTION, 
SECRET_1_VALUE),
+            createParameter(SECRET_2_NAME, SECRET_2_DESCRIPTION, 
SECRET_2_VALUE));
+        final ParameterProviderNode providerNode2 = 
createMockedParameterProviderNode(PROVIDER_2_ID, PROVIDER_2_NAME, GROUP_2_NAME,
+            createParameter(SECRET_3_NAME, SECRET_3_DESCRIPTION, 
SECRET_3_VALUE));
+
+        final Set<ParameterProviderNode> providers = new HashSet<>();
+        providers.add(providerNode1);
+        providers.add(providerNode2);
+        when(flowManager.getAllParameterProviders()).thenReturn(providers);
+
+        secretsManager = new ParameterProviderSecretsManager();
+        final SecretsManagerInitializationContext initContext = new 
StandardSecretsManagerInitializationContext(flowManager);
+        secretsManager.initialize(initContext);
+    }
+
+    private ParameterProviderNode createMockedParameterProviderNode(final 
String id, final String name, final String groupName, final Parameter... 
parameters) {
+        final ParameterProviderNode node = mock(ParameterProviderNode.class);
+        when(node.getIdentifier()).thenReturn(id);
+        when(node.getName()).thenReturn(name);
+
+        final List<Parameter> parameterList = List.of(parameters);
+        final ParameterGroup group = new ParameterGroup(groupName, 
parameterList);
+        final List<ParameterGroup> groups = List.of(group);
+
+        when(node.fetchParameterValues()).thenReturn(groups);
+
+        when(node.fetchParameterValues(anyList())).thenAnswer(invocation -> {
+            final List<String> requestedNames = invocation.getArgument(0);
+            final List<Parameter> matchingParameters = parameterList.stream()
+                    .filter(p -> 
requestedNames.contains(p.getDescriptor().getName()))
+                    .toList();
+            if (matchingParameters.isEmpty()) {
+                return List.of();
+            }
+            return List.of(new ParameterGroup(groupName, matchingParameters));
+        });
+
+        return node;
+    }
+
+    private Parameter createParameter(final String name, final String 
description, final String value) {
+        final ParameterDescriptor descriptor = new 
ParameterDescriptor.Builder()
+                .name(name)
+                .description(description)
+                .build();
+        return new Parameter.Builder()
+                .descriptor(descriptor)
+                .value(value)
+                .build();
+    }
+
+    private SecretReference createSecretReference(final String providerId, 
final String providerName, final String secretName) {
+        final SecretReference reference = mock(SecretReference.class);
+        when(reference.getProviderId()).thenReturn(providerId);
+        when(reference.getProviderName()).thenReturn(providerName);
+        when(reference.getSecretName()).thenReturn(secretName);
+        return reference;
+    }
+
+    @Test
+    public void 
testGetSecretProvidersReturnsOneProviderPerMockedParameterProvider() {
+        final Set<SecretProvider> providers = 
secretsManager.getSecretProviders();
+
+        assertEquals(2, providers.size());
+
+        boolean foundProvider1 = false;
+        boolean foundProvider2 = false;
+        for (final SecretProvider provider : providers) {
+            if (PROVIDER_1_ID.equals(provider.getProviderId())) {
+                assertEquals(PROVIDER_1_NAME, provider.getProviderName());
+                foundProvider1 = true;
+            } else if (PROVIDER_2_ID.equals(provider.getProviderId())) {
+                assertEquals(PROVIDER_2_NAME, provider.getProviderName());
+                foundProvider2 = true;
+            }
+        }
+
+        assertTrue(foundProvider1);
+        assertTrue(foundProvider2);
+    }
+
+    @Test
+    public void testGetAllSecretsRetrievesSecretsFromAllProviders() {
+        final List<Secret> allSecrets = secretsManager.getAllSecrets();
+
+        assertEquals(3, allSecrets.size());
+
+        boolean foundSecret1 = false;
+        boolean foundSecret2 = false;
+        boolean foundSecret3 = false;
+
+        for (final Secret secret : allSecrets) {
+            if (SECRET_1_NAME.equals(secret.getName())) {
+                assertEquals(PROVIDER_1_NAME, secret.getProviderName());
+                assertEquals(GROUP_1_NAME, secret.getGroupName());
+                assertEquals(SECRET_1_VALUE, secret.getValue());
+                foundSecret1 = true;
+            } else if (SECRET_2_NAME.equals(secret.getName())) {
+                assertEquals(PROVIDER_1_NAME, secret.getProviderName());
+                assertEquals(GROUP_1_NAME, secret.getGroupName());
+                assertEquals(SECRET_2_VALUE, secret.getValue());
+                foundSecret2 = true;
+            } else if (SECRET_3_NAME.equals(secret.getName())) {
+                assertEquals(PROVIDER_2_NAME, secret.getProviderName());
+                assertEquals(GROUP_2_NAME, secret.getGroupName());
+                assertEquals(SECRET_3_VALUE, secret.getValue());
+                foundSecret3 = true;
+            }
+        }
+
+        assertTrue(foundSecret1);
+        assertTrue(foundSecret2);
+        assertTrue(foundSecret3);
+    }
+
+    @Test
+    public void testGetSecretReturnsPopulatedOptionalWhenSecretIsFoundById() {
+        final SecretReference reference = createSecretReference(PROVIDER_1_ID, 
null, SECRET_1_NAME);
+
+        final Optional<Secret> result = secretsManager.getSecret(reference);
+
+        assertTrue(result.isPresent());
+        final Secret secret = result.get();
+        assertEquals(SECRET_1_NAME, secret.getName());
+        assertEquals(SECRET_1_VALUE, secret.getValue());
+        assertEquals(PROVIDER_1_NAME, secret.getProviderName());
+    }
+
+    @Test
+    public void testGetSecretReturnsPopulatedOptionalWhenSecretIsFoundByName() 
{
+        final SecretReference reference = createSecretReference(null, 
PROVIDER_2_NAME, SECRET_3_NAME);
+
+        final Optional<Secret> result = secretsManager.getSecret(reference);
+
+        assertTrue(result.isPresent());
+        final Secret secret = result.get();
+        assertEquals(SECRET_3_NAME, secret.getName());
+        assertEquals(SECRET_3_VALUE, secret.getValue());
+        assertEquals(PROVIDER_2_NAME, secret.getProviderName());
+    }
+
+    @Test
+    public void 
testGetSecretReturnsEmptyOptionalWhenInvalidSecretNameProvided() {
+        final SecretReference reference = createSecretReference(PROVIDER_1_ID, 
PROVIDER_1_NAME, "non-existent-secret");
+        final Optional<Secret> result = secretsManager.getSecret(reference);
+        assertFalse(result.isPresent());
+    }
+
+    @Test
+    public void testGetSecretReturnsEmptyOptionalWhenProviderNotFound() {
+        final SecretReference reference = 
createSecretReference("invalid-provider-id", "Invalid Provider", SECRET_1_NAME);
+        final Optional<Secret> result = secretsManager.getSecret(reference);
+        assertFalse(result.isPresent());
+    }
+
+    @Test
+    public void testGetSecretsReturnsAllSecretsProperlyMappedWhenAllAreFound() 
{
+        final SecretReference reference1 = 
createSecretReference(PROVIDER_1_ID, PROVIDER_1_NAME, SECRET_1_NAME);
+        final SecretReference reference2 = 
createSecretReference(PROVIDER_1_ID, PROVIDER_1_NAME, SECRET_2_NAME);
+        final SecretReference reference3 = 
createSecretReference(PROVIDER_2_ID, PROVIDER_2_NAME, SECRET_3_NAME);
+        final Set<SecretReference> references = Set.of(reference1, reference2, 
reference3);
+
+        final Map<SecretReference, Secret> results = 
secretsManager.getSecrets(references);
+        assertEquals(3, results.size());
+
+        final Secret secret1 = results.get(reference1);
+        assertNotNull(secret1);
+        assertEquals(SECRET_1_NAME, secret1.getName());
+        assertEquals(SECRET_1_VALUE, secret1.getValue());
+
+        final Secret secret2 = results.get(reference2);
+        assertNotNull(secret2);
+        assertEquals(SECRET_2_NAME, secret2.getName());
+        assertEquals(SECRET_2_VALUE, secret2.getValue());
+
+        final Secret secret3 = results.get(reference3);
+        assertNotNull(secret3);
+        assertEquals(SECRET_3_NAME, secret3.getName());
+        assertEquals(SECRET_3_VALUE, secret3.getValue());
+    }
+
+    @Test
+    public void testGetSecretsReturnsNullValueForSecretWithInvalidName() {
+        final SecretReference validReference = 
createSecretReference(PROVIDER_1_ID, PROVIDER_1_NAME, SECRET_1_NAME);
+        final SecretReference invalidReference = 
createSecretReference(PROVIDER_1_ID, PROVIDER_1_NAME, "non-existent-secret");
+        final Set<SecretReference> references = Set.of(validReference, 
invalidReference);
+
+        final Map<SecretReference, Secret> results = 
secretsManager.getSecrets(references);
+        assertEquals(2, results.size());
+
+        final Secret validSecret = results.get(validReference);
+        assertNotNull(validSecret);
+        assertEquals(SECRET_1_NAME, validSecret.getName());
+
+        assertTrue(results.containsKey(invalidReference));
+        assertNull(results.get(invalidReference));
+    }
+
+    @Test
+    public void testGetSecretsReturnsNullValueForSecretWithInvalidProvider() {
+        final SecretReference validReference = 
createSecretReference(PROVIDER_1_ID, PROVIDER_1_NAME, SECRET_1_NAME);
+        final SecretReference invalidProviderReference = 
createSecretReference("invalid-id", "Invalid Provider", SECRET_1_NAME);
+        final Set<SecretReference> references = Set.of(validReference, 
invalidProviderReference);
+
+        final Map<SecretReference, Secret> results = 
secretsManager.getSecrets(references);
+        assertEquals(2, results.size());
+
+        final Secret validSecret = results.get(validReference);
+        assertNotNull(validSecret);
+
+        assertTrue(results.containsKey(invalidProviderReference));
+        assertNull(results.get(invalidProviderReference));
+    }
+
+    @Test
+    public void testGetSecretsFromMultipleProvidersWithMixedResults() {
+        final SecretReference provider1Secret1 = 
createSecretReference(PROVIDER_1_ID, null, SECRET_1_NAME);
+        final SecretReference provider1Invalid = 
createSecretReference(PROVIDER_1_ID, null, "invalid-secret");
+        final SecretReference provider2Secret3 = 
createSecretReference(PROVIDER_2_ID, null, SECRET_3_NAME);
+        final SecretReference invalidProvider = 
createSecretReference("invalid-id", null, SECRET_1_NAME);
+
+        final Set<SecretReference> references = Set.of(provider1Secret1, 
provider1Invalid, provider2Secret3, invalidProvider);
+        final Map<SecretReference, Secret> results = 
secretsManager.getSecrets(references);
+        assertEquals(4, results.size());
+
+        assertNotNull(results.get(provider1Secret1));
+        assertEquals(SECRET_1_NAME, results.get(provider1Secret1).getName());
+
+        assertTrue(results.containsKey(provider1Invalid));
+        assertNull(results.get(provider1Invalid));
+
+        assertNotNull(results.get(provider2Secret3));
+        assertEquals(SECRET_3_NAME, results.get(provider2Secret3).getName());
+
+        assertTrue(results.containsKey(invalidProvider));
+        assertNull(results.get(invalidProvider));
+    }
+
+    @Test
+    public void testGetSecretProviderSearchesByIdFirst() {
+        final SecretReference referenceWithBothIdAndName = 
createSecretReference(PROVIDER_1_ID, PROVIDER_2_NAME, SECRET_1_NAME);
+        final Optional<Secret> result = 
secretsManager.getSecret(referenceWithBothIdAndName);
+        assertTrue(result.isPresent());
+        assertEquals(PROVIDER_1_NAME, result.get().getProviderName());
+    }
+}
+
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorConfiguration.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorConfiguration.java
index 88c54d29988..4261e2af3d2 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorConfiguration.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorConfiguration.java
@@ -32,6 +32,15 @@ public class ConnectorConfiguration {
         return stepConfigurations;
     }
 
+    public NamedStepConfiguration getNamedStepConfiguration(final String name) 
{
+        for (final NamedStepConfiguration stepConfiguration : 
stepConfigurations) {
+            if (stepConfiguration.stepName().equals(name)) {
+                return stepConfiguration;
+            }
+        }
+        return null;
+    }
+
     @Override
     public boolean equals(final Object other) {
         if (other == null || getClass() != other.getClass()) {
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepository.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepository.java
index 3ea48b996a8..c951a29f596 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepository.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepository.java
@@ -17,6 +17,7 @@
 
 package org.apache.nifi.components.connector;
 
+import org.apache.nifi.components.connector.secrets.SecretsManager;
 import org.apache.nifi.flow.Bundle;
 import org.apache.nifi.flow.VersionedConfigurationStep;
 
@@ -81,6 +82,8 @@ public interface ConnectorRepository {
 
     void inheritConfiguration(ConnectorNode connector, 
List<VersionedConfigurationStep> flowConfiguration, Bundle flowContextBundle) 
throws FlowUpdateException;
 
+    SecretsManager getSecretsManager();
+
     /**
      * Creates a new ConnectorStateTransition instance for managing the 
lifecycle state of a connector.
      *
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepositoryInitializationContext.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepositoryInitializationContext.java
index 5b0b923ac7b..0f0c50c3943 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepositoryInitializationContext.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepositoryInitializationContext.java
@@ -17,6 +17,7 @@
 
 package org.apache.nifi.components.connector;
 
+import org.apache.nifi.components.connector.secrets.SecretsManager;
 import org.apache.nifi.controller.NodeTypeProvider;
 import org.apache.nifi.controller.flow.FlowManager;
 import org.apache.nifi.nar.ExtensionManager;
@@ -27,6 +28,8 @@ public interface ConnectorRepositoryInitializationContext {
 
     ExtensionManager getExtensionManager();
 
+    SecretsManager getSecretsManager();
+
     NodeTypeProvider getNodeTypeProvider();
 
     ConnectorRequestReplicator getRequestReplicator();
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/FrameworkConnectorInitializationContext.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/FrameworkConnectorInitializationContext.java
index ed977c4b1ea..8749e48192b 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/FrameworkConnectorInitializationContext.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/FrameworkConnectorInitializationContext.java
@@ -18,6 +18,7 @@
 package org.apache.nifi.components.connector;
 
 import org.apache.nifi.asset.AssetManager;
+import org.apache.nifi.components.connector.secrets.SecretsManager;
 
 public interface FrameworkConnectorInitializationContext extends 
ConnectorInitializationContext {
 
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/FrameworkConnectorInitializationContextBuilder.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/FrameworkConnectorInitializationContextBuilder.java
index b19262e27cc..030e333a847 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/FrameworkConnectorInitializationContextBuilder.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/FrameworkConnectorInitializationContextBuilder.java
@@ -18,6 +18,7 @@
 package org.apache.nifi.components.connector;
 
 import org.apache.nifi.asset.AssetManager;
+import org.apache.nifi.components.connector.secrets.SecretsManager;
 import org.apache.nifi.logging.ComponentLog;
 
 public interface FrameworkConnectorInitializationContextBuilder {
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/FrameworkConnectorInitializationContext.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/secrets/SecretProvider.java
similarity index 71%
copy from 
nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/FrameworkConnectorInitializationContext.java
copy to 
nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/secrets/SecretProvider.java
index ed977c4b1ea..3c0a2f4e5f1 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/FrameworkConnectorInitializationContext.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/secrets/SecretProvider.java
@@ -15,14 +15,20 @@
  * limitations under the License.
  */
 
-package org.apache.nifi.components.connector;
+package org.apache.nifi.components.connector.secrets;
 
-import org.apache.nifi.asset.AssetManager;
+import org.apache.nifi.components.connector.Secret;
 
-public interface FrameworkConnectorInitializationContext extends 
ConnectorInitializationContext {
+import java.util.List;
 
-    SecretsManager getSecretsManager();
+public interface SecretProvider {
 
-    AssetManager getAssetManager();
+    String getProviderId();
+
+    String getProviderName();
+
+    List<Secret> getAllSecrets();
+
+    List<Secret> getSecrets(List<String> fullyQualifiedSecretNames);
 
 }
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepositoryInitializationContext.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/secrets/SecretsManager.java
similarity index 57%
copy from 
nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepositoryInitializationContext.java
copy to 
nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/secrets/SecretsManager.java
index 5b0b923ac7b..62365ed940e 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepositoryInitializationContext.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/secrets/SecretsManager.java
@@ -15,20 +15,26 @@
  * limitations under the License.
  */
 
-package org.apache.nifi.components.connector;
+package org.apache.nifi.components.connector.secrets;
 
-import org.apache.nifi.controller.NodeTypeProvider;
-import org.apache.nifi.controller.flow.FlowManager;
-import org.apache.nifi.nar.ExtensionManager;
+import org.apache.nifi.components.connector.Secret;
+import org.apache.nifi.components.connector.SecretReference;
 
-public interface ConnectorRepositoryInitializationContext {
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
 
-    FlowManager getFlowManager();
+public interface SecretsManager {
 
-    ExtensionManager getExtensionManager();
+    void initialize(SecretsManagerInitializationContext initializationContext);
 
-    NodeTypeProvider getNodeTypeProvider();
+    List<Secret> getAllSecrets();
 
-    ConnectorRequestReplicator getRequestReplicator();
+    Set<SecretProvider> getSecretProviders();
+
+    Optional<Secret> getSecret(SecretReference secretReference);
+
+    Map<SecretReference, Secret> getSecrets(Set<SecretReference> 
secretReferences);
 
 }
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/FrameworkConnectorInitializationContext.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/secrets/SecretsManagerInitializationContext.java
similarity index 75%
copy from 
nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/FrameworkConnectorInitializationContext.java
copy to 
nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/secrets/SecretsManagerInitializationContext.java
index ed977c4b1ea..4f4cf8e9a86 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/FrameworkConnectorInitializationContext.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/secrets/SecretsManagerInitializationContext.java
@@ -15,14 +15,12 @@
  * limitations under the License.
  */
 
-package org.apache.nifi.components.connector;
+package org.apache.nifi.components.connector.secrets;
 
-import org.apache.nifi.asset.AssetManager;
+import org.apache.nifi.controller.flow.FlowManager;
 
-public interface FrameworkConnectorInitializationContext extends 
ConnectorInitializationContext {
+public interface SecretsManagerInitializationContext {
 
-    SecretsManager getSecretsManager();
-
-    AssetManager getAssetManager();
+    FlowManager getFlowManager();
 
 }
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ParameterProviderNode.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ParameterProviderNode.java
index 873c6c29dba..f663d3624a1 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ParameterProviderNode.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ParameterProviderNode.java
@@ -49,6 +49,19 @@ public interface ParameterProviderNode extends ComponentNode 
{
      */
     void fetchParameters();
 
+    /**
+     * Fetches all parameter values from the Parameter Provider and returns 
them. This method does not cache the results for later retrieval.
+     * @return all Parameter Groups with Parameter Names and Values
+     */
+    List<ParameterGroup> fetchParameterValues();
+
+    /**
+     * Fetches parameters that match the provided fully qualified parameter 
names. This method does not cache the results for later retrieval.
+     * @param fullyQualifiedParameterNames fully qualified names of parameters 
to fetch
+     * @return Parameter Groups with Parameter Names and Values that match the 
provided names
+     */
+    List<ParameterGroup> fetchParameterValues(List<String> 
fullyQualifiedParameterNames);
+
     /**
      * Find a named Parameter Group cached from previous request to fetch 
Parameters from the configured Parameter Provider
      *
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorConfigurationContext.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorConfigurationContext.java
index 6940112ad9f..975bc2aafa4 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorConfigurationContext.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorConfigurationContext.java
@@ -18,6 +18,8 @@
 package org.apache.nifi.components.connector;
 
 import org.apache.nifi.asset.AssetManager;
+import org.apache.nifi.components.connector.secrets.SecretProvider;
+import org.apache.nifi.components.connector.secrets.SecretsManager;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorInitializationContext.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorInitializationContext.java
index 89730a5bacc..c31cad237f1 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorInitializationContext.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorInitializationContext.java
@@ -19,6 +19,7 @@ package org.apache.nifi.components.connector;
 
 import org.apache.nifi.asset.AssetManager;
 import org.apache.nifi.components.connector.components.FlowContext;
+import org.apache.nifi.components.connector.secrets.SecretsManager;
 import org.apache.nifi.flow.VersionedExternalFlow;
 import org.apache.nifi.flow.VersionedProcessGroup;
 import org.apache.nifi.logging.ComponentLog;
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java
index c1772de2928..7e8839dedf4 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java
@@ -577,17 +577,18 @@ public class StandardConnectorNode implements 
ConnectorNode {
             }
 
             final ConfigurationStep configurationStep = optionalStep.get();
-            final List<ValidationResult> validationResults = 
getConnector().validateConfigurationStep(configurationStep, configContext, 
validationContext);
+
+            final List<ValidationResult> validationResults = new ArrayList<>();
+            validatePropertyReferences(configurationStep, 
configurationOverrides, validationResults);
+
+            if (validationResults.isEmpty()) {
+                final List<ValidationResult> implValidationResults = 
getConnector().validateConfigurationStep(configurationStep, configContext, 
validationContext);
+                validationResults.addAll(implValidationResults);
+            }
 
             final List<ConfigVerificationResult> invalidConfigResults = 
validationResults.stream()
                 .filter(result -> !result.isValid())
-                .map(validationResult -> new ConfigVerificationResult.Builder()
-                    .verificationStepName("Property Validation - " + 
validationResult.getSubject())
-                    .outcome(Outcome.FAILED)
-                    .subject(validationResult.getSubject())
-                    .explanation(validationResult.getExplanation())
-                    .build()
-                )
+                .map(this::createConfigVerificationResult)
                 .toList();
 
             if (invalidConfigResults.isEmpty()) {
@@ -595,15 +596,25 @@ public class StandardConnectorNode implements 
ConnectorNode {
                     .verificationStepName("Property Validation")
                     .outcome(Outcome.SUCCESSFUL)
                     .build());
+
+                
results.addAll(getConnector().verifyConfigurationStep(stepName, 
resolvedPropertyOverrides, workingFlowContext));
             } else {
                 results.addAll(invalidConfigResults);
             }
 
-            results.addAll(getConnector().verifyConfigurationStep(stepName, 
resolvedPropertyOverrides, workingFlowContext));
             return results;
         }
     }
 
+    private ConfigVerificationResult createConfigVerificationResult(final 
ValidationResult validationResult) {
+        return new ConfigVerificationResult.Builder()
+            .verificationStepName("Property Validation - " + 
validationResult.getSubject())
+            .outcome(validationResult.isValid() ? Outcome.SUCCESSFUL : 
Outcome.FAILED)
+            .subject(validationResult.getSubject())
+            .explanation(validationResult.getExplanation())
+            .build();
+    }
+
     private Map<String, String> resolvePropertyReferences(final 
StepConfiguration configurationOverrides) {
         final Map<String, String> resolvedProperties = new HashMap<>();
 
@@ -756,18 +767,23 @@ public class StandardConnectorNode implements 
ConnectorNode {
 
             final ConnectorValidationContext validationContext = 
createValidationContext(activeFlowContext);
 
-            List<ValidationResult> allResults;
-            try {
-                allResults = getConnector().validate(activeFlowContext, 
validationContext);
-            } catch (final Exception e) {
-                allResults = List.of(new ValidationResult.Builder()
+            final List<ValidationResult> allResults = new ArrayList<>();
+            validatePropertyReferences(allResults);
+
+            if (allResults.isEmpty()) {
+                try {
+                    final List<ValidationResult> implValidationResults = 
getConnector().validate(activeFlowContext, validationContext);
+                    allResults.addAll(implValidationResults);
+                } catch (final Exception e) {
+                    allResults.add(new ValidationResult.Builder()
                         .subject("Validation Failure")
                         .valid(false)
                         .explanation("Encountered a failure while attempting 
to perform validation: " + e.getMessage())
                         .build());
+                }
             }
 
-            if (allResults == null) {
+            if (allResults.isEmpty()) {
                 return new ValidationState(ValidationStatus.VALID, 
Collections.emptyList());
             }
 
@@ -786,6 +802,79 @@ public class StandardConnectorNode implements 
ConnectorNode {
         }
     }
 
+
+    private void validatePropertyReferences(final List<ValidationResult> 
allResults) {
+        final List<ConfigurationStep> configurationSteps = 
getConnector().getConfigurationSteps(activeFlowContext);
+        final ConnectorConfiguration connectorConfiguration = 
activeFlowContext.getConfigurationContext().toConnectorConfiguration();
+
+        for (final ConfigurationStep step : configurationSteps) {
+            final NamedStepConfiguration namedStepConfig = 
connectorConfiguration.getNamedStepConfiguration(step.getName());
+            if (namedStepConfig == null) {
+                continue;
+            }
+
+            validatePropertyReferences(step, namedStepConfig.configuration(), 
allResults);
+        }
+    }
+
+    private void validatePropertyReferences(final ConfigurationStep step, 
final StepConfiguration stepConfig, final List<ValidationResult> allResults) {
+        for (final ConnectorPropertyGroup propertyGroup : 
step.getPropertyGroups()) {
+            for (final ConnectorPropertyDescriptor descriptor : 
propertyGroup.getProperties()) {
+                final PropertyType propertyType = descriptor.getType();
+                final ConnectorValueReference reference = 
stepConfig.getPropertyValue(descriptor.getName());
+
+                final String subject = step.getName() + " / " + 
descriptor.getName();
+
+                if (!isReferenceAllowed(reference, propertyType)) {
+                    final String providedReferenceType = switch 
(reference.getValueType()) {
+                        case ASSET_REFERENCE -> "<Asset reference>";
+                        case SECRET_REFERENCE -> "<Secret reference>";
+                        case STRING_LITERAL -> "<Explicit value>";
+                    };
+
+                    final String expectedReferenceType = propertyType == 
PropertyType.SECRET ? "a Secret reference" : "an Explicit value";
+
+                    allResults.add(new ValidationResult.Builder()
+                        .subject(subject)
+                        .input(providedReferenceType)
+                        .explanation("This property must be configured with " 
+ expectedReferenceType)
+                        .build());
+                }
+            }
+        }
+    }
+
+    private boolean isReferenceAllowed(final ConnectorValueReference 
reference, final PropertyType propertyType) {
+        // If the reference is null or its value is unset, then it is allowed
+        if (reference == null) {
+            return true;
+        }
+
+        switch (reference) {
+            case StringLiteralValue stringLiteralValue -> {
+                if (stringLiteralValue.getValue() == null) {
+                    return true;
+                }
+            }
+            case AssetReference assetReference -> {
+                if (assetReference.getAssetIdentifier() == null) {
+                    return true;
+                }
+            }
+            case SecretReference secretReference -> {
+                if (secretReference.getSecretName() == null) {
+                    return true;
+                }
+            }
+        }
+
+        if (propertyType == PropertyType.SECRET) {
+            return reference.getValueType() == 
ConnectorValueType.SECRET_REFERENCE;
+        }
+
+        return reference.getValueType() != ConnectorValueType.SECRET_REFERENCE;
+    }
+
     private ConnectorValidationContext createValidationContext(final 
FrameworkFlowContext context) {
         final DescribedValueProvider allowableValueProvider = (stepName, 
propertyName) ->
             fetchAllowableValues(stepName, propertyName, context);
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepoInitializationContext.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepoInitializationContext.java
index d899d136816..4a1b8ed8b31 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepoInitializationContext.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepoInitializationContext.java
@@ -17,6 +17,7 @@
 
 package org.apache.nifi.components.connector;
 
+import org.apache.nifi.components.connector.secrets.SecretsManager;
 import org.apache.nifi.controller.NodeTypeProvider;
 import org.apache.nifi.controller.flow.FlowManager;
 import org.apache.nifi.nar.ExtensionManager;
@@ -24,15 +25,18 @@ import org.apache.nifi.nar.ExtensionManager;
 public class StandardConnectorRepoInitializationContext implements 
ConnectorRepositoryInitializationContext {
     private final FlowManager flowManager;
     private final ExtensionManager extensionManager;
+    private final SecretsManager secretsManager;
     private final NodeTypeProvider nodeTypeProvider;
     private final ConnectorRequestReplicator requestReplicator;
 
     public StandardConnectorRepoInitializationContext(final FlowManager 
flowManager,
                                                      final ExtensionManager 
extensionManager,
+                                                     final SecretsManager 
secretsManager,
                                                      final NodeTypeProvider 
nodeTypeProvider,
                                                      final 
ConnectorRequestReplicator requestReplicator) {
         this.flowManager = flowManager;
         this.extensionManager = extensionManager;
+        this.secretsManager = secretsManager;
         this.nodeTypeProvider = nodeTypeProvider;
         this.requestReplicator = requestReplicator;
     }
@@ -47,6 +51,11 @@ public class StandardConnectorRepoInitializationContext 
implements ConnectorRepo
         return extensionManager;
     }
 
+    @Override
+    public SecretsManager getSecretsManager() {
+        return secretsManager;
+    }
+
     @Override
     public NodeTypeProvider getNodeTypeProvider() {
         return nodeTypeProvider;
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepository.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepository.java
index 615595d42f8..4cb87da4cfb 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepository.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepository.java
@@ -18,6 +18,7 @@
 package org.apache.nifi.components.connector;
 
 import org.apache.nifi.annotation.lifecycle.OnRemoved;
+import org.apache.nifi.components.connector.secrets.SecretsManager;
 import org.apache.nifi.engine.FlowEngine;
 import org.apache.nifi.flow.Bundle;
 import org.apache.nifi.flow.VersionedConfigurationStep;
@@ -43,11 +44,13 @@ public class StandardConnectorRepository implements 
ConnectorRepository {
 
     private volatile ExtensionManager extensionManager;
     private volatile ConnectorRequestReplicator requestReplicator;
+    private volatile SecretsManager secretsManager;
 
     @Override
     public void initialize(final ConnectorRepositoryInitializationContext 
context) {
         this.extensionManager = context.getExtensionManager();
         this.requestReplicator = context.getRequestReplicator();
+        this.secretsManager = context.getSecretsManager();
     }
 
     @Override
@@ -173,6 +176,11 @@ public class StandardConnectorRepository implements 
ConnectorRepository {
         }
     }
 
+    @Override
+    public SecretsManager getSecretsManager() {
+        return secretsManager;
+    }
+
     @Override
     public ConnectorStateTransition createStateTransition(final String type, 
final String id) {
         final String componentDescription = "StandardConnectorNode[id=" + id + 
", type=" + type + "]";
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/ExtensionBuilder.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/ExtensionBuilder.java
index 0607ccc0ab6..b628c91a232 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/ExtensionBuilder.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/ExtensionBuilder.java
@@ -42,7 +42,6 @@ import 
org.apache.nifi.components.connector.FrameworkConnectorInitializationCont
 import org.apache.nifi.components.connector.FrameworkFlowContext;
 import org.apache.nifi.components.connector.GhostConnector;
 import 
org.apache.nifi.components.connector.MutableConnectorConfigurationContext;
-import org.apache.nifi.components.connector.SecretsManager;
 import org.apache.nifi.components.connector.StandardConnectorNode;
 import org.apache.nifi.components.connector.StepConfiguration;
 import org.apache.nifi.components.connector.StringLiteralValue;
@@ -543,7 +542,7 @@ public class ExtensionBuilder {
                    for (final ConnectorPropertyDescriptor descriptor : 
propertyGroup.getProperties()) {
                        final String name = descriptor.getName();
                        final String defaultValue = 
descriptor.getDefaultValue();
-                       defaultValues.put(name, new 
StringLiteralValue(defaultValue));
+                       defaultValues.put(name, defaultValue == null ? null : 
new StringLiteralValue(defaultValue));
                    }
                }
 
@@ -553,15 +552,13 @@ public class ExtensionBuilder {
    }
 
    private FrameworkConnectorInitializationContext 
createConnectorInitializationContext(final ProcessGroup managedProcessGroup, 
final ComponentLog componentLog) {
-
        final String name = type.contains(".") ? 
StringUtils.substringAfterLast(type, ".") : type;
-       final SecretsManager secretsManager = null;
 
        return connectorInitializationContextBuilder
            .identifier(identifier)
            .name(name)
            .componentLog(componentLog)
-           .secretsManager(secretsManager)
+           
.secretsManager(flowController.getConnectorRepository().getSecretsManager())
            .assetManager(flowController.getAssetManager())
            .build();
    }
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
index a7fc6ddbe7f..6ce20753f81 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
@@ -53,6 +53,10 @@ import 
org.apache.nifi.components.connector.ConnectorRepositoryInitializationCon
 import org.apache.nifi.components.connector.ConnectorRequestReplicator;
 import 
org.apache.nifi.components.connector.StandardConnectorRepoInitializationContext;
 import org.apache.nifi.components.connector.StandardConnectorRepository;
+import 
org.apache.nifi.components.connector.secrets.ParameterProviderSecretsManager;
+import org.apache.nifi.components.connector.secrets.SecretsManager;
+import 
org.apache.nifi.components.connector.secrets.SecretsManagerInitializationContext;
+import 
org.apache.nifi.components.connector.secrets.StandardSecretsManagerInitializationContext;
 import org.apache.nifi.components.monitor.LongRunningTaskMonitor;
 import org.apache.nifi.components.state.StateManagerProvider;
 import org.apache.nifi.components.state.StateProvider;
@@ -269,6 +273,7 @@ public class FlowController implements 
ReportingTaskProvider, FlowAnalysisRulePr
     public static final String DEFAULT_SWAP_MANAGER_IMPLEMENTATION = 
"org.apache.nifi.controller.FileSystemSwapManager";
     public static final String DEFAULT_ASSET_MANAGER_IMPLEMENTATION = 
StandardAssetManager.class.getName();
     public static final String DEFAULT_CONNECTOR_REPOSITORY_IMPLEMENTATION = 
StandardConnectorRepository.class.getName();
+    public static final String DEFAULT_SECRETS_MANAGER_IMPLEMENTATION = 
ParameterProviderSecretsManager.class.getName();
 
     public static final String GRACEFUL_SHUTDOWN_PERIOD = 
"nifi.flowcontroller.graceful.shutdown.seconds";
     public static final long DEFAULT_GRACEFUL_SHUTDOWN_SECONDS = 10;
@@ -603,7 +608,8 @@ public class FlowController implements 
ReportingTaskProvider, FlowAnalysisRulePr
         controllerServiceResolver = new 
StandardControllerServiceResolver(authorizer, flowManager, new 
VersionedComponentFlowMapper(extensionManager),
                 controllerServiceProvider, new 
StandardControllerServiceApiLookup(extensionManager));
 
-        connectorRepository = createConnectorRepository(nifiProperties, 
extensionManager, flowManager, this, connectorRequestReplicator);
+        final SecretsManager secretsManager = 
createSecretsManager(nifiProperties, extensionManager, flowManager);
+        connectorRepository = createConnectorRepository(nifiProperties, 
extensionManager, flowManager, secretsManager, this, 
connectorRequestReplicator);
 
         final PythonBridge rawPythonBridge = 
createPythonBridge(nifiProperties, controllerServiceProvider);
         final ClassLoader pythonBridgeClassLoader = 
rawPythonBridge.getClass().getClassLoader();
@@ -876,7 +882,7 @@ public class FlowController implements 
ReportingTaskProvider, FlowAnalysisRulePr
     }
 
     private static ConnectorRepository createConnectorRepository(final 
NiFiProperties properties, final ExtensionDiscoveringManager extensionManager, 
final FlowManager flowManager,
-                final NodeTypeProvider nodeTypeProvider, final 
ConnectorRequestReplicator requestReplicator) {
+                final SecretsManager secretsManager, final NodeTypeProvider 
nodeTypeProvider, final ConnectorRequestReplicator requestReplicator) {
 
         final String implementationClassName = 
properties.getProperty(NiFiProperties.CONNECTOR_REPOSITORY_IMPLEMENTATION, 
DEFAULT_CONNECTOR_REPOSITORY_IMPLEMENTATION);
 
@@ -884,12 +890,14 @@ public class FlowController implements 
ReportingTaskProvider, FlowAnalysisRulePr
             // Discover implementations of Connector Repository. This is not 
done at startup because the ConnectorRepository class is not
             // provided in the list of standard extension points. This is due 
to the fact that ConnectorRepository lives in the nifi-framework-core-api, and
             // does not make sense to refactor it into some other module due 
to its dependencies, simply to allow it to be discovered at startup.
-            
extensionManager.discoverExtensions(extensionManager.getAllBundles(), 
Set.of(ConnectorRepository.class), true);
+            final Set<Class<?>> additionalExtensionTypes = 
Set.of(ConnectorRepository.class, SecretsManager.class);
+            
extensionManager.discoverExtensions(extensionManager.getAllBundles(), 
additionalExtensionTypes, true);
             final ConnectorRepository created = 
NarThreadContextClassLoader.createInstance(extensionManager, 
implementationClassName, ConnectorRepository.class, properties);
 
             final ConnectorRepositoryInitializationContext 
initializationContext = new StandardConnectorRepoInitializationContext(
                 flowManager,
                 extensionManager,
+                secretsManager,
                 nodeTypeProvider,
                 requestReplicator
             );
@@ -901,6 +909,35 @@ public class FlowController implements 
ReportingTaskProvider, FlowAnalysisRulePr
                 }
             }
 
+            LOG.info("Created Connector Repository of type {}", 
created.getClass().getSimpleName());
+
+            return created;
+        } catch (final Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private static SecretsManager createSecretsManager(final NiFiProperties 
properties, final ExtensionDiscoveringManager extensionManager, final 
FlowManager flowManager) {
+        final String implementationClassName = 
properties.getProperty(NiFiProperties.SECRETS_MANAGER_IMPLEMENTATION, 
DEFAULT_SECRETS_MANAGER_IMPLEMENTATION);
+
+        try {
+            // Discover implementations of Secrets Manager. This is not done 
at startup because the SecretsManager class is not
+            // provided in the list of standard extension points. This is due 
to the fact that SecretsManager lives in the nifi-framework-core-api, and
+            // does not make sense to refactor it into some other module due 
to its dependencies, simply to allow it to be discovered at startup.
+            
extensionManager.discoverExtensions(extensionManager.getAllBundles(), 
Set.of(SecretsManager.class), true);
+            final SecretsManager created = 
NarThreadContextClassLoader.createInstance(extensionManager, 
implementationClassName, SecretsManager.class, properties);
+
+            final SecretsManagerInitializationContext initializationContext = 
new StandardSecretsManagerInitializationContext(flowManager);
+
+            synchronized (created) {
+                // Ensure that any NAR dependencies are available when we 
initialize the ConnectorRepository
+                try (final NarCloseable ignored = 
NarCloseable.withComponentNarLoader(extensionManager, created.getClass(), 
"secrets-manager")) {
+                    created.initialize(initializationContext);
+                }
+            }
+
+            LOG.info("Created Secrets Manager of type {}", 
created.getClass().getSimpleName());
+
             return created;
         } catch (final Exception e) {
             throw new RuntimeException(e);
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/flow/StandardFlowManager.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/flow/StandardFlowManager.java
index d4e5fd6e879..804d0f732e9 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/flow/StandardFlowManager.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/flow/StandardFlowManager.java
@@ -764,7 +764,9 @@ public class StandardFlowManager extends 
AbstractFlowManager implements FlowMana
 
         final ConnectorRepository connectorRepository = 
flowController.getConnectorRepository();
         final ConnectorStateTransition stateTransition = 
connectorRepository.createStateTransition(type, id);
-        final StandardConnectorConfigurationContext activeConfigurationContext 
= new StandardConnectorConfigurationContext(flowController.getAssetManager(), 
null);
+        final StandardConnectorConfigurationContext activeConfigurationContext 
= new StandardConnectorConfigurationContext(
+            flowController.getAssetManager(), 
flowController.getConnectorRepository().getSecretsManager());
+
         final ProcessGroupFactory processGroupFactory = groupId -> 
createProcessGroup(groupId, false);
         final FlowContextFactory flowContextFactory = new 
FlowControllerFlowContextFactory(flowController, managedRootGroup, 
activeConfigurationContext, processGroupFactory);
 
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorConfigurationContext.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorConfigurationContext.java
index f3da3a8e50d..9c8b6b6e087 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorConfigurationContext.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorConfigurationContext.java
@@ -18,6 +18,7 @@
 package org.apache.nifi.components.connector;
 
 import org.apache.nifi.asset.AssetManager;
+import org.apache.nifi.components.connector.secrets.SecretsManager;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
diff --git 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorNode.java
 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorNode.java
index 40f2323b9bb..5147fb0d041 100644
--- 
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorNode.java
+++ 
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorNode.java
@@ -23,6 +23,7 @@ import org.apache.nifi.components.ConfigVerificationResult;
 import org.apache.nifi.components.ValidationResult;
 import org.apache.nifi.components.connector.components.FlowContext;
 import org.apache.nifi.components.connector.components.FlowContextType;
+import org.apache.nifi.components.connector.secrets.SecretsManager;
 import org.apache.nifi.controller.flow.FlowManager;
 import org.apache.nifi.engine.FlowEngine;
 import org.apache.nifi.flow.Bundle;