This is an automated email from the ASF dual-hosted git repository.
sanjeevrk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 1aff3c4 Secretprovider interfaces and some default implementations
(#2855)
1aff3c4 is described below
commit 1aff3c4cdc60ec81b0230289944a97cb1db4d683
Author: Sanjeev Kulkarni <sanjee...@gmail.com>
AuthorDate: Sun Oct 28 07:59:03 2018 -0700
Secretprovider interfaces and some default implementations (#2855)
* Added SecretProvider and SecretProviderConfigurator interfaces and some
implementations
* Added appropriate documentation
* Added validation logic to config provider configurator
* Added more comments
* Took feedback into account
---
.../pulsar/common/functions/FunctionConfig.java | 3 +-
.../org/apache/pulsar/common/io/SinkConfig.java | 3 +-
.../org/apache/pulsar/common/io/SourceConfig.java | 3 +-
pulsar-functions/pom.xml | 1 +
pulsar-functions/{ => secrets}/pom.xml | 54 ++++------
.../secretsprovider/ClearTextSecretsProvider.java | 38 +++++++
.../EnvironmentBasedSecretsProvider.java | 35 +++++++
.../functions/secretsprovider/SecretsProvider.java | 40 ++++++++
.../DefaultSecretsProviderConfigurator.java | 67 ++++++++++++
.../KubernetesSecretsProviderConfigurator.java | 113 +++++++++++++++++++++
.../SecretsProviderConfigurator.java | 72 +++++++++++++
.../ClearTextSecretsProviderTest.java | 37 +++++++
.../EnvironmentBasedSecretsProviderTest.java | 75 ++++++++++++++
.../KubernetesSecretsProviderConfiguratorTest.java | 62 +++++++++++
14 files changed, 566 insertions(+), 37 deletions(-)
diff --git
a/pulsar-common/src/main/java/org/apache/pulsar/common/functions/FunctionConfig.java
b/pulsar-common/src/main/java/org/apache/pulsar/common/functions/FunctionConfig.java
index 2163df1..28a15af 100644
---
a/pulsar-common/src/main/java/org/apache/pulsar/common/functions/FunctionConfig.java
+++
b/pulsar-common/src/main/java/org/apache/pulsar/common/functions/FunctionConfig.java
@@ -82,7 +82,8 @@ public class FunctionConfig {
// This is a map of secretName(aka how the secret is going to be
// accessed in the function via context) to an object that
// encapsulates how the secret is fetched by the underlying
- // secrets provider
+ // secrets provider. The type of an value here can be found by the
+ // SecretProviderConfigurator.getSecretObjectType() method.
private Map<String, Object> secrets;
private Runtime runtime;
private boolean autoAck;
diff --git
a/pulsar-common/src/main/java/org/apache/pulsar/common/io/SinkConfig.java
b/pulsar-common/src/main/java/org/apache/pulsar/common/io/SinkConfig.java
index 355c696..3eacfbc 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/io/SinkConfig.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/io/SinkConfig.java
@@ -58,7 +58,8 @@ public class SinkConfig {
// This is a map of secretName(aka how the secret is going to be
// accessed in the function via context) to an object that
// encapsulates how the secret is fetched by the underlying
- // secrets provider
+ // secrets provider. The type of an value here can be found by the
+ // SecretProviderConfigurator.getSecretObjectType() method.
private Map<String, Object> secrets;
private int parallelism = 1;
private FunctionConfig.ProcessingGuarantees processingGuarantees;
diff --git
a/pulsar-common/src/main/java/org/apache/pulsar/common/io/SourceConfig.java
b/pulsar-common/src/main/java/org/apache/pulsar/common/io/SourceConfig.java
index 9dbe97c..c1b2efe 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/io/SourceConfig.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/io/SourceConfig.java
@@ -49,7 +49,8 @@ public class SourceConfig {
// This is a map of secretName(aka how the secret is going to be
// accessed in the function via context) to an object that
// encapsulates how the secret is fetched by the underlying
- // secrets provider
+ // secrets provider. The type of an value here can be found by the
+ // SecretProviderConfigurator.getSecretObjectType() method.
private Map<String, Object> secrets;
private int parallelism = 1;
private FunctionConfig.ProcessingGuarantees processingGuarantees;
diff --git a/pulsar-functions/pom.xml b/pulsar-functions/pom.xml
index f54effd..a27c21b 100644
--- a/pulsar-functions/pom.xml
+++ b/pulsar-functions/pom.xml
@@ -43,6 +43,7 @@
<module>runtime-shaded</module>
<module>runtime-all</module>
<module>worker</module>
+ <module>secrets</module>
</modules>
<dependencyManagement>
diff --git a/pulsar-functions/pom.xml b/pulsar-functions/secrets/pom.xml
similarity index 53%
copy from pulsar-functions/pom.xml
copy to pulsar-functions/secrets/pom.xml
index f54effd..fa411d7 100644
--- a/pulsar-functions/pom.xml
+++ b/pulsar-functions/secrets/pom.xml
@@ -19,49 +19,35 @@
-->
<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
<modelVersion>4.0.0</modelVersion>
- <packaging>pom</packaging>
<parent>
<groupId>org.apache.pulsar</groupId>
- <artifactId>pulsar</artifactId>
+ <artifactId>pulsar-functions</artifactId>
<version>2.3.0-SNAPSHOT</version>
</parent>
- <artifactId>pulsar-functions</artifactId>
- <name>Pulsar Functions :: Parent</name>
-
- <modules>
- <module>proto</module>
- <module>proto-shaded</module>
- <module>api-java</module>
- <module>java-examples</module>
- <module>utils</module>
- <module>metrics</module>
- <module>instance</module>
- <module>runtime</module>
- <module>runtime-shaded</module>
- <module>runtime-all</module>
- <module>worker</module>
- </modules>
-
- <dependencyManagement>
- <dependencies>
-
- <dependency>
- <groupId>org.projectlombok</groupId>
- <artifactId>lombok</artifactId>
- <version>1.16.12</version>
- </dependency>
-
- </dependencies>
- </dependencyManagement>
+ <artifactId>pulsar-functions-secrets</artifactId>
+ <name>Pulsar Functions :: Secrets</name>
<dependencies>
+ <dependency>
+ <groupId>io.kubernetes</groupId>
+ <artifactId>client-java</artifactId>
+ <version>2.0.0</version>
+ <scope>compile</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-classic</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
<dependency>
- <groupId>org.projectlombok</groupId>
- <artifactId>lombok</artifactId>
- <scope>provided</scope>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-functions-proto</artifactId>
+ <version>${project.version}</version>
</dependency>
</dependencies>
diff --git
a/pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsprovider/ClearTextSecretsProvider.java
b/pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsprovider/ClearTextSecretsProvider.java
new file mode 100644
index 0000000..adb2852
--- /dev/null
+++
b/pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsprovider/ClearTextSecretsProvider.java
@@ -0,0 +1,38 @@
+/**
+ * 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.pulsar.functions.secretsprovider;
+
+/**
+ * This file defines a very basic clear text secrets provider which treats
+ * the secrets as being passed in cleartext.
+ */
+public class ClearTextSecretsProvider implements SecretsProvider {
+ /**
+ * Fetches a secret
+ * @return The actual secret
+ */
+ @Override
+ public String provideSecret(String secretName, Object pathToSecret) {
+ if (pathToSecret != null) {
+ return pathToSecret.toString();
+ } else {
+ return null;
+ }
+ }
+}
\ No newline at end of file
diff --git
a/pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsprovider/EnvironmentBasedSecretsProvider.java
b/pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsprovider/EnvironmentBasedSecretsProvider.java
new file mode 100644
index 0000000..b058709
--- /dev/null
+++
b/pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsprovider/EnvironmentBasedSecretsProvider.java
@@ -0,0 +1,35 @@
+/**
+ * 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.pulsar.functions.secretsprovider;
+
+/**
+ * This defines a very simple Secrets Provider that looks up environment
variable
+ * thats named the same as secretName and fetches it.
+ */
+public class EnvironmentBasedSecretsProvider implements SecretsProvider {
+
+ /**
+ * Fetches a secret
+ * @return The actual secret
+ */
+ @Override
+ public String provideSecret(String secretName, Object pathToSecret) {
+ return System.getenv(secretName);
+ }
+}
\ No newline at end of file
diff --git
a/pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsprovider/SecretsProvider.java
b/pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsprovider/SecretsProvider.java
new file mode 100644
index 0000000..7d5330d
--- /dev/null
+++
b/pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsprovider/SecretsProvider.java
@@ -0,0 +1,40 @@
+/**
+ * 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.pulsar.functions.secretsprovider;
+
+import java.util.Map;
+
+/**
+ * This file defines the SecretsProvider interface. This interface is used by
the function
+ * instances/containers to actually fetch the secrets. What SecretsProvider to
use is
+ * decided by the SecretsProviderConfigurator
+ */
+public interface SecretsProvider {
+ /**
+ * Initialize the SecretsProvider
+ * @return
+ */
+ default void init(Map<String, String> config) {}
+
+ /**
+ * Fetches a secret
+ * @return The actual secret
+ */
+ String provideSecret(String secretName, Object pathToSecret);
+}
\ No newline at end of file
diff --git
a/pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsproviderconfigurator/DefaultSecretsProviderConfigurator.java
b/pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsproviderconfigurator/DefaultSecretsProviderConfigurator.java
new file mode 100644
index 0000000..10f9d54
--- /dev/null
+++
b/pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsproviderconfigurator/DefaultSecretsProviderConfigurator.java
@@ -0,0 +1,67 @@
+/**
+ * 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.pulsar.functions.secretsproviderconfigurator;
+
+import com.google.gson.reflect.TypeToken;
+import io.kubernetes.client.models.V1Container;
+import org.apache.pulsar.functions.proto.Function;
+import org.apache.pulsar.functions.secretsprovider.ClearTextSecretsProvider;
+
+import java.lang.reflect.Type;
+import java.util.Map;
+
+/**
+ * This is a barebones version of a secrets provider which wires in
ClearTextSecretsProvider
+ * to the function instances/containers.
+ * While this is the default configurator, it is highly recommended that for
real-security
+ * you use some alternate provider.
+ */
+public class DefaultSecretsProviderConfigurator implements
SecretsProviderConfigurator {
+ @Override
+ public String getSecretsProviderClassName(Function.FunctionDetails
functionDetails) {
+ switch (functionDetails.getRuntime()) {
+ case JAVA:
+ return ClearTextSecretsProvider.class.getName();
+ case PYTHON:
+ return "secretsprovider.ClearTextSecretsProvider";
+ default:
+ throw new RuntimeException("Unknwon runtime " +
functionDetails.getRuntime());
+ }
+ }
+
+ @Override
+ public Map<String, String>
getSecretsProviderConfig(Function.FunctionDetails functionDetails) {
+ return null;
+ }
+
+ @Override
+ public void configureKubernetesRuntimeSecretsProvider(V1Container
container, Function.FunctionDetails functionDetails) {
+ // noop
+ }
+
+ @Override
+ public void configureProcessRuntimeSecretsProvider(ProcessBuilder
processBuilder, Function.FunctionDetails functionDetails) {
+ // noop
+ }
+
+ @Override
+ public Type getSecretObjectType() {
+ return new TypeToken<String>() {}.getType();
+ }
+}
\ No newline at end of file
diff --git
a/pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsproviderconfigurator/KubernetesSecretsProviderConfigurator.java
b/pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsproviderconfigurator/KubernetesSecretsProviderConfigurator.java
new file mode 100644
index 0000000..0321734
--- /dev/null
+++
b/pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsproviderconfigurator/KubernetesSecretsProviderConfigurator.java
@@ -0,0 +1,113 @@
+/**
+ * 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.pulsar.functions.secretsproviderconfigurator;
+
+import com.google.gson.Gson;
+import com.google.gson.reflect.TypeToken;
+import io.kubernetes.client.models.V1Container;
+import io.kubernetes.client.models.V1EnvVar;
+import io.kubernetes.client.models.V1EnvVarSource;
+import io.kubernetes.client.models.V1SecretKeySelector;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.functions.proto.Function;
+import
org.apache.pulsar.functions.secretsprovider.EnvironmentBasedSecretsProvider;
+
+import java.lang.reflect.Type;
+import java.util.Map;
+
+/**
+ * This file defines the SecretsProviderConfigurator that will be used by
default for running in Kubernetes.
+ * As such this implementation is strictly when workers are configured to use
kubernetes runtime.
+ * We use kubernetes in built secrets and bind them as environment variables
within the function container
+ * to ensure that the secrets are availble to the function at runtime. Then we
plug in the
+ * EnvironmentBasedSecretsConfig as the secrets provider who knows how to read
these environment variables
+ */
+public class KubernetesSecretsProviderConfigurator implements
SecretsProviderConfigurator {
+ private static String ID_KEY = "id";
+ private static String KEY_KEY = "key";
+ @Override
+ public String getSecretsProviderClassName(Function.FunctionDetails
functionDetails) {
+ switch (functionDetails.getRuntime()) {
+ case JAVA:
+ return EnvironmentBasedSecretsProvider.class.getName();
+ case PYTHON:
+ return "secretsprovider.EnvironmentBasedSecretsProvider";
+ default:
+ throw new RuntimeException("Unknown function runtime " +
functionDetails.getRuntime());
+ }
+ }
+
+ @Override
+ public Map<String, String>
getSecretsProviderConfig(Function.FunctionDetails functionDetails) {
+ return null;
+ }
+
+ // Kubernetes secrets can be exposed as volume mounts or as environment
variables in the pods. We are currently using the
+ // environment variables way. Essentially the secretName/secretPath is
attached as secretRef to the environment variables
+ // of a pod and kubernetes magically makes the secret pointed to by this
combination available as a env variable.
+ @Override
+ public void configureKubernetesRuntimeSecretsProvider(V1Container
container, Function.FunctionDetails functionDetails) {
+ if (!StringUtils.isEmpty(functionDetails.getSecretsMap())) {
+ Type type = new TypeToken<Map<String, Object>>() {
+ }.getType();
+ Map<String, Object> secretsMap = new
Gson().fromJson(functionDetails.getSecretsMap(), type);
+ for (Map.Entry<String, Object> entry : secretsMap.entrySet()) {
+ final V1EnvVar secretEnv = new V1EnvVar();
+ Map<String, String> kv = (Map<String, String>)
entry.getValue();
+ secretEnv.name(entry.getKey())
+ .valueFrom(new V1EnvVarSource()
+ .secretKeyRef(new V1SecretKeySelector()
+ .name(kv.get(ID_KEY))
+ .key(kv.get(KEY_KEY))));
+ container.addEnvItem(secretEnv);
+ }
+ }
+ }
+
+ @Override
+ public void configureProcessRuntimeSecretsProvider(ProcessBuilder
processBuilder, Function.FunctionDetails functionDetails) {
+ throw new RuntimeException("KubernetesSecretsProviderConfigurator
should only be setup for Kubernetes Runtime");
+ }
+
+ @Override
+ public Type getSecretObjectType() {
+ return new TypeToken<Map<String, String>>() {}.getType();
+ }
+
+ // The secret object should be of type Map<String, String> and it should
contain "id" and "key"
+ @Override
+ public void validateSecretMap(Map<String, Object> secretMap) {
+ for (Object object : secretMap.values()) {
+ if (object instanceof Map) {
+ Map<String, String> kubernetesSecret = (Map<String, String>)
object;
+ if (kubernetesSecret.size() < 2) {
+ throw new IllegalArgumentException("Kubernetes Secret
should contain id and key");
+ }
+ if (!kubernetesSecret.containsKey(ID_KEY)) {
+ throw new IllegalArgumentException("Kubernetes Secret
should contain id information");
+ }
+ if (!kubernetesSecret.containsKey(KEY_KEY)) {
+ throw new IllegalArgumentException("Kubernetes Secret
should contain key information");
+ }
+ } else {
+ throw new IllegalArgumentException("Kubernetes Secret should
be a Map containing id/key pairs");
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git
a/pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsproviderconfigurator/SecretsProviderConfigurator.java
b/pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsproviderconfigurator/SecretsProviderConfigurator.java
new file mode 100644
index 0000000..a1792f1
--- /dev/null
+++
b/pulsar-functions/secrets/src/main/java/org/apache/pulsar/functions/secretsproviderconfigurator/SecretsProviderConfigurator.java
@@ -0,0 +1,72 @@
+/**
+ * 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.pulsar.functions.secretsproviderconfigurator;
+
+import io.kubernetes.client.models.V1Container;
+import org.apache.pulsar.functions.proto.Function;
+
+import java.lang.reflect.Type;
+import java.util.Map;
+
+/**
+ * This file defines the SecretsProviderConfigurator interface. This interface
is used by the function_workers
+ * to choose the SecretProvider class name(if any) and its associated config
at the time of starting
+ * the function instances.
+ */
+public interface SecretsProviderConfigurator {
+ /**
+ * Initialize the SecretsProviderConfigurator
+ * @return
+ */
+ default void init(Map<String, String> config) {}
+
+ /**
+ * Return the Secrets Provider Classname. This will be passed to the
cmdline
+ * of the instance and should contain the logic of connecting with the
secrets
+ * provider and obtaining secrets
+ */
+ String getSecretsProviderClassName(Function.FunctionDetails
functionDetails);
+
+ /**
+ * Return the secrets provider config
+ */
+ Map<String, String> getSecretsProviderConfig(Function.FunctionDetails
functionDetails);
+
+ /**
+ * Attaches any secrets specific stuff to the k8 container for kubernetes
runtime
+ */
+ void configureKubernetesRuntimeSecretsProvider(V1Container container,
Function.FunctionDetails functionDetails);
+
+ /**
+ * Attaches any secrets specific stuff to the ProcessBuilder for process
runtime
+ */
+ void configureProcessRuntimeSecretsProvider(ProcessBuilder processBuilder,
Function.FunctionDetails functionDetails);
+
+ /**
+ * What is the type of the object that should be in the user secret config
+ * @return
+ */
+ Type getSecretObjectType();
+
+ /**
+ * Do config checks to see whether the secrets provided are conforming
+ */
+ default void validateSecretMap(Map<String, Object> secretMap) {}
+
+}
\ No newline at end of file
diff --git
a/pulsar-functions/secrets/src/test/java/org/apache/pulsar/functions/secretsprovider/ClearTextSecretsProviderTest.java
b/pulsar-functions/secrets/src/test/java/org/apache/pulsar/functions/secretsprovider/ClearTextSecretsProviderTest.java
new file mode 100644
index 0000000..d6f0139
--- /dev/null
+++
b/pulsar-functions/secrets/src/test/java/org/apache/pulsar/functions/secretsprovider/ClearTextSecretsProviderTest.java
@@ -0,0 +1,37 @@
+/**
+ * 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.pulsar.functions.secretsprovider;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+/**
+ * Unit test of {@link Exceptions}.
+ */
+public class ClearTextSecretsProviderTest {
+
+ @Test
+ public void testConfigValidation() throws Exception {
+ ClearTextSecretsProvider provider = new ClearTextSecretsProvider();
+ Assert.assertEquals(provider.provideSecret("SecretName",
"SecretValue"), "SecretValue");
+ Assert.assertEquals(provider.provideSecret("SecretName", ""), "");
+ Assert.assertEquals(provider.provideSecret("SecretName", null), null);
+ }
+}
diff --git
a/pulsar-functions/secrets/src/test/java/org/apache/pulsar/functions/secretsprovider/EnvironmentBasedSecretsProviderTest.java
b/pulsar-functions/secrets/src/test/java/org/apache/pulsar/functions/secretsprovider/EnvironmentBasedSecretsProviderTest.java
new file mode 100644
index 0000000..fa80d70
--- /dev/null
+++
b/pulsar-functions/secrets/src/test/java/org/apache/pulsar/functions/secretsprovider/EnvironmentBasedSecretsProviderTest.java
@@ -0,0 +1,75 @@
+/**
+ * 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.pulsar.functions.secretsprovider;
+
+import org.mockito.Mockito;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.lang.reflect.Field;
+import java.util.Map;
+
+import static org.mockito.Matchers.anyString;
+
+/**
+ * Unit test of {@link Exceptions}.
+ */
+public class EnvironmentBasedSecretsProviderTest {
+ @Test
+ public void testConfigValidation() throws Exception {
+ EnvironmentBasedSecretsProvider provider = new
EnvironmentBasedSecretsProvider();
+ Assert.assertEquals(provider.provideSecret("mySecretName", "Ignored"),
null);
+ injectEnvironmentVariable("mySecretName", "SecretValue");
+ Assert.assertEquals(provider.provideSecret("mySecretName", "Ignored"),
"SecretValue");
+ }
+
+ private static void injectEnvironmentVariable(String key, String value)
+ throws Exception {
+
+ Class<?> processEnvironment =
Class.forName("java.lang.ProcessEnvironment");
+
+ Field unmodifiableMapField = getAccessibleField(processEnvironment,
"theUnmodifiableEnvironment");
+ Object unmodifiableMap = unmodifiableMapField.get(null);
+ injectIntoUnmodifiableMap(key, value, unmodifiableMap);
+
+ Field mapField = getAccessibleField(processEnvironment,
"theEnvironment");
+ Map<String, String> map = (Map<String, String>) mapField.get(null);
+ map.put(key, value);
+ }
+
+ private static Field getAccessibleField(Class<?> clazz, String fieldName)
+ throws NoSuchFieldException {
+
+ Field field = clazz.getDeclaredField(fieldName);
+ field.setAccessible(true);
+ return field;
+ }
+
+ private static void injectIntoUnmodifiableMap(String key, String value,
Object map)
+ throws ReflectiveOperationException {
+
+ Class unmodifiableMap =
Class.forName("java.util.Collections$UnmodifiableMap");
+ Field field = getAccessibleField(unmodifiableMap, "m");
+ Object obj = field.get(map);
+ ((Map<String, String>) obj).put(key, value);
+ }
+}
diff --git
a/pulsar-functions/secrets/src/test/java/org/apache/pulsar/functions/secretsproviderconfigurator/KubernetesSecretsProviderConfiguratorTest.java
b/pulsar-functions/secrets/src/test/java/org/apache/pulsar/functions/secretsproviderconfigurator/KubernetesSecretsProviderConfiguratorTest.java
new file mode 100644
index 0000000..6d64d5e
--- /dev/null
+++
b/pulsar-functions/secrets/src/test/java/org/apache/pulsar/functions/secretsproviderconfigurator/KubernetesSecretsProviderConfiguratorTest.java
@@ -0,0 +1,62 @@
+/**
+ * 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.pulsar.functions.secretsproviderconfigurator;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.util.HashMap;
+
+/**
+ * Unit test of {@link Exceptions}.
+ */
+public class KubernetesSecretsProviderConfiguratorTest {
+
+ @Test
+ public void testConfigValidation() throws Exception {
+ KubernetesSecretsProviderConfigurator provider = new
KubernetesSecretsProviderConfigurator();
+ try {
+ HashMap<String, Object> map = new HashMap<String, Object>();
+ map.put("secretname", "randomsecret");
+ provider.validateSecretMap(map);
+ Assert.fail("Non conforming secret object should not validate");
+ } catch (Exception e) {
+ }
+ try {
+ HashMap<String, Object> map = new HashMap<String, Object>();
+ HashMap<String, String> map1 = new HashMap<String, String>();
+ map1.put("secretname", "secretvalue");
+ map.put("secretname", map1);
+ provider.validateSecretMap(map);
+ Assert.fail("Non conforming secret object should not validate");
+ } catch (Exception e) {
+ }
+ try {
+ HashMap<String, Object> map = new HashMap<String, Object>();
+ HashMap<String, String> map1 = new HashMap<String, String>();
+ map1.put("id", "secretvalue");
+ map1.put("key", "secretvalue");
+ map.put("secretname", map1);
+ provider.validateSecretMap(map);
+ } catch (Exception e) {
+ Assert.fail("Conforming secret object should validate");
+ }
+ }
+}
