[FLINK-38263][table] Add SecretStore related interfaces

This closes #27394.

Author: lihaosky

flink-python/pyflink/table/tests/test_environment_settings_completeness.py

@@ -38,7 +38,7 @@ def java_class(cls):
     def excluded_methods(cls):
         # internal interfaces, no need to expose to users.
         return {'getPlanner', 'getExecutor', 'getUserClassLoader', 'getCatalogStore',
-                'toConfiguration', 'fromConfiguration', 'getSqlFactory'}
+                'getSecretStore', 'toConfiguration', 'fromConfiguration', 'getSqlFactory'}
 
 
 class EnvironmentSettingsBuilderCompletenessTests(PythonAPICompletenessTestCase, PyFlinkTestCase):
@@ -59,7 +59,7 @@ def java_class(cls):
     def excluded_methods(cls):
         # internal interfaces, no need to expose to users.
         # withSqlFactory - needs to be implemented
-        return {'withClassLoader', 'withCatalogStore', 'withSqlFactory'}
+        return {'withClassLoader', 'withCatalogStore', 'withSecretStore', 'withSqlFactory'}
 
 if __name__ == '__main__':
     import unittest

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/EnvironmentSettings.java

@@ -25,6 +25,7 @@
 import org.apache.flink.table.catalog.CatalogStore;
 import org.apache.flink.table.expressions.SqlFactory;
 import org.apache.flink.table.functions.UserDefinedFunction;
+import org.apache.flink.table.secret.SecretStore;
 
 import javax.annotation.Nullable;
 
@@ -66,16 +67,19 @@ public class EnvironmentSettings {
 
     private final @Nullable CatalogStore catalogStore;
     private final @Nullable SqlFactory sqlFactory;
+    private final @Nullable SecretStore secretStore;
 
     private EnvironmentSettings(
             Configuration configuration,
             ClassLoader classLoader,
             CatalogStore catalogStore,
-            SqlFactory sqlFactory) {
+            SqlFactory sqlFactory,
+            SecretStore secretStore) {
         this.configuration = configuration;
         this.classLoader = classLoader;
         this.catalogStore = catalogStore;
         this.sqlFactory = sqlFactory;
+        this.secretStore = secretStore;
     }
 
     /**
@@ -153,6 +157,11 @@ public Optional<SqlFactory> getSqlFactory() {
         return Optional.ofNullable(sqlFactory);
     }
 
+    @Internal
+    public Optional<SecretStore> getSecretStore() {
+        return Optional.ofNullable(secretStore);
+    }
+
     /** A builder for {@link EnvironmentSettings}. */
     @PublicEvolving
     public static class Builder {
@@ -162,6 +171,7 @@ public static class Builder {
 
         private @Nullable CatalogStore catalogStore;
         private @Nullable SqlFactory sqlFactory;
+        private @Nullable SecretStore secretStore;
 
         public Builder() {}
 
@@ -250,12 +260,28 @@ public Builder withSqlFactory(SqlFactory sqlFactory) {
             return this;
         }
 
+        /**
+         * Specifies the {@link SecretStore} to be used for managing secrets in the {@link
+         * TableEnvironment}.
+         *
+         * <p>The secret store allows for secure storage and retrieval of sensitive configuration
+         * data such as credentials, tokens, and passwords.
+         *
+         * @param secretStore the secret store instance to use
+         * @return this builder
+         */
+        public Builder withSecretStore(SecretStore secretStore) {
+            this.secretStore = secretStore;
+            return this;
+        }
+
         /** Returns an immutable instance of {@link EnvironmentSettings}. */
         public EnvironmentSettings build() {
             if (classLoader == null) {
                 classLoader = Thread.currentThread().getContextClassLoader();
             }
-            return new EnvironmentSettings(configuration, classLoader, catalogStore, sqlFactory);
+            return new EnvironmentSettings(
+                    configuration, classLoader, catalogStore, sqlFactory, secretStore);
         }
     }
 }

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java

@@ -120,6 +120,8 @@
 import org.apache.flink.table.resource.ResourceManager;
 import org.apache.flink.table.resource.ResourceType;
 import org.apache.flink.table.resource.ResourceUri;
+import org.apache.flink.table.secret.SecretStore;
+import org.apache.flink.table.secret.SecretStoreFactory;
 import org.apache.flink.table.types.AbstractDataType;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.utils.DataTypeUtils;
@@ -258,6 +260,13 @@ public static TableEnvironmentImpl create(EnvironmentSettings settings) {
         final CatalogStore catalogStore = catalogStoreResult.getCatalogStore();
         final CatalogStoreFactory catalogStoreFactory = catalogStoreResult.getCatalogStoreFactory();
 
+        final ApiFactoryUtil.SecretStoreResult secretStoreResult =
+                ApiFactoryUtil.getOrCreateSecretStore(
+                        settings.getSecretStore(), settings.getConfiguration(), userClassLoader);
+        final SecretStore secretStore = secretStoreResult.getSecretStore();
+        final SecretStoreFactory secretStoreFactory = secretStoreResult.getSecretStoreFactory();
+        // TODO (FLINK-38261): pass secret store to catalog manager for encryption/decryption
+
         // use configuration to init table config
         final TableConfig tableConfig = TableConfig.getDefault();
         tableConfig.setRootConfiguration(executor.getConfiguration());

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/factories/ApiFactoryUtil.java

@@ -19,10 +19,14 @@
 package org.apache.flink.table.factories;
 
 import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.DelegatingConfiguration;
 import org.apache.flink.table.catalog.CatalogStore;
 import org.apache.flink.table.catalog.CommonCatalogOptions;
+import org.apache.flink.table.secret.CommonSecretOptions;
+import org.apache.flink.table.secret.SecretStore;
+import org.apache.flink.table.secret.SecretStoreFactory;
 
 import javax.annotation.Nullable;
 
@@ -127,4 +131,101 @@ public static CatalogStoreFactory.Context buildCatalogStoreFactoryContext(
 
         return context;
     }
+
+    /** Result holder for secret store and factory. */
+    @Internal
+    public static class SecretStoreResult {
+        private final SecretStore secretStore;
+        @Nullable private final SecretStoreFactory secretStoreFactory;
+
+        public SecretStoreResult(
+                SecretStore secretStore, @Nullable SecretStoreFactory secretStoreFactory) {
+            this.secretStore = secretStore;
+            this.secretStoreFactory = secretStoreFactory;
+        }
+
+        public SecretStore getSecretStore() {
+            return secretStore;
+        }
+
+        @Nullable
+        public SecretStoreFactory getSecretStoreFactory() {
+            return secretStoreFactory;
+        }
+    }
+
+    /**
+     * Gets or creates a {@link SecretStore}. If a secret store is provided in settings, it will be
+     * used directly. Otherwise, a new secret store will be created using the factory.
+     *
+     * @param providedSecretStore the secret store from settings, if present
+     * @param configuration the configuration
+     * @param classLoader the user classloader
+     * @return a result containing the secret store and factory (factory is null if store was
+     *     provided)
+     */
+    public static SecretStoreResult getOrCreateSecretStore(
+            Optional<SecretStore> providedSecretStore,
+            Configuration configuration,
+            ClassLoader classLoader) {
+        if (providedSecretStore.isPresent()) {
+            return new SecretStoreResult(providedSecretStore.get(), null);
+        } else {
+            SecretStoreFactory secretStoreFactory =
+                    findAndCreateSecretStoreFactory(configuration, classLoader);
+            SecretStoreFactory.Context secretStoreFactoryContext =
+                    buildSecretStoreFactoryContext(configuration, classLoader);
+            secretStoreFactory.open(secretStoreFactoryContext);
+            SecretStore secretStore = secretStoreFactory.createSecretStore();
+            return new SecretStoreResult(secretStore, secretStoreFactory);
+        }
+    }
+
+    /**
+     * Finds and creates a {@link SecretStoreFactory} using the provided {@link Configuration} and
+     * user classloader.
+     *
+     * <p>The configuration format should be as follows:
+     *
+     * <pre>{@code
+     * table.secret-store.kind: {identifier}
+     * table.secret-store.{identifier}.{param1}: xxx
+     * table.secret-store.{identifier}.{param2}: xxx
+     * }</pre>
+     */
+    @VisibleForTesting
+    static SecretStoreFactory findAndCreateSecretStoreFactory(
+            Configuration configuration, ClassLoader classLoader) {
+        String identifier = configuration.get(CommonSecretOptions.TABLE_SECRET_STORE_KIND);
+
+        SecretStoreFactory secretStoreFactory =
+                FactoryUtil.discoverFactory(classLoader, SecretStoreFactory.class, identifier);
+
+        return secretStoreFactory;
+    }
+
+    /**
+     * Build a {@link SecretStoreFactory.Context} for opening the {@link SecretStoreFactory}.
+     *
+     * <p>The configuration format should be as follows:
+     *
+     * <pre>{@code
+     * table.secret-store.kind: {identifier}
+     * table.secret-store.{identifier}.{param1}: xxx
+     * table.secret-store.{identifier}.{param2}: xxx
+     * }</pre>
+     */
+    @VisibleForTesting
+    static SecretStoreFactory.Context buildSecretStoreFactoryContext(
+            Configuration configuration, ClassLoader classLoader) {
+        String identifier = configuration.get(CommonSecretOptions.TABLE_SECRET_STORE_KIND);
+        String secretStoreOptionPrefix =
+                CommonSecretOptions.TABLE_SECRET_STORE_OPTION_PREFIX + identifier + ".";
+        Map<String, String> options =
+                new DelegatingConfiguration(configuration, secretStoreOptionPrefix).toMap();
+        SecretStoreFactory.Context context =
+                new FactoryUtil.DefaultSecretStoreContext(options, configuration, classLoader);
+
+        return context;
+    }
 }

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/secret/GenericInMemorySecretStore.java

@@ -0,0 +1,94 @@
+/*
+ * 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.flink.table.secret;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.secret.exceptions.SecretNotFoundException;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A generic in-memory implementation of both {@link ReadableSecretStore} and {@link
+ * WritableSecretStore}.
+ *
+ * <p>This implementation stores secrets in memory as immutable Map objects. It is suitable for
+ * testing and development purposes but should not be used in production environments as secrets are
+ * not encrypted.
+ */
+@Internal
+public class GenericInMemorySecretStore implements ReadableSecretStore, WritableSecretStore {
+
+    private final Map<String, Map<String, String>> secrets;
+
+    public GenericInMemorySecretStore() {
+        this.secrets = new HashMap<>();
+    }
+
+    @Override
+    public Map<String, String> getSecret(String secretId) throws SecretNotFoundException {
+        checkNotNull(secretId, "Secret ID cannot be null");
+
+        Map<String, String> secretData = secrets.get(secretId);
+        if (secretData == null) {
+            throw new SecretNotFoundException(
+                    String.format("Secret with ID '%s' not found", secretId));
+        }
+
+        return secretData;
+    }
+
+    @Override
+    public String storeSecret(Map<String, String> secretData) {
+        checkNotNull(secretData, "Secret data cannot be null");
+
+        String secretId = UUID.randomUUID().toString();
+        secrets.put(secretId, Collections.unmodifiableMap(new HashMap<>(secretData)));
+        return secretId;
+    }
+
+    @Override
+    public void removeSecret(String secretId) {
+        checkNotNull(secretId, "Secret ID cannot be null");
+        secrets.remove(secretId);
+    }
+
+    @Override
+    public void updateSecret(String secretId, Map<String, String> newSecretData)
+            throws SecretNotFoundException {
+        checkNotNull(secretId, "Secret ID cannot be null");
+        checkNotNull(newSecretData, "New secret data cannot be null");
+
+        if (!secrets.containsKey(secretId)) {
+            throw new SecretNotFoundException(
+                    String.format("Secret with ID '%s' not found", secretId));
+        }
+
+        secrets.put(secretId, Collections.unmodifiableMap(new HashMap<>(newSecretData)));
+    }
+
+    /** Clears all secrets from the store (for testing purposes). */
+    void clear() {
+        secrets.clear();
+    }
+}

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/secret/GenericInMemorySecretStoreFactory.java

@@ -0,0 +1,63 @@
+/*
+ * 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.flink.table.secret;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+
+import java.util.Set;
+
+/**
+ * Factory for creating {@link GenericInMemorySecretStore} instances.
+ *
+ * <p>This factory creates in-memory secret stores that are suitable for testing and development
+ * purposes. Secrets are stored in plaintext JSON format in memory and are not persisted.
+ */
+@Internal
+public class GenericInMemorySecretStoreFactory implements SecretStoreFactory {
+
+    public static final String IDENTIFIER = "generic_in_memory";
+
+    @Override
+    public String factoryIdentifier() {
+        return IDENTIFIER;
+    }
+
+    @Override
+    public Set<ConfigOption<?>> requiredOptions() {
+        return Set.of();
+    }
+
+    @Override
+    public Set<ConfigOption<?>> optionalOptions() {
+        return Set.of();
+    }
+
+    @Override
+    public SecretStore createSecretStore() {
+        return new GenericInMemorySecretStore();
+    }
+
+    @Override
+    public void open(Context context) {}
+
+    @Override
+    public void close() throws CatalogException {}
+}