基于自定义org.apache.kafka.common.config.provider管理Vault密钥生命周期

我基于org.apache.kafka.common.config.provider.ConfigProvider接口开发用于动态数据库密钥的Vault配置提供者，其中configure方法负责配置Vault连接参数，get方法实现Vault密钥的检索逻辑。现在已经实现了Vault密钥修改/续期的检测方法，想知道如何将其集成到自定义ConfigProvider中，触发密钥更新并同步更新使用该提供者的Kafka Connect连接器配置，同时需要参考代码示例。

核心思路 #

Kafka Connect的配置刷新机制依赖于ConfigProvider的get方法返回的ConfigData对象中的expiration时间戳，以及Connect框架对配置的定期刷新检查。当密钥更新时，通过更新ConfigData的过期时间，让Connect主动触发重新拉取配置；同时结合密钥变更检测逻辑，可提前触发配置刷新。

集成步骤 #

• 维护密钥状态与检测逻辑：在ConfigProvider中新增后台定时任务，执行已实现的Vault密钥变更检测方法，缓存当前密钥版本/哈希值，检测到变更时标记缓存失效。
• 利用ConfigData expiration控制刷新：get方法中，若缓存失效则从Vault拉取最新密钥，设置下一次检测间隔为过期时间；缓存有效时返回当前密钥及剩余过期时间，到期后Connect自动重新调用get方法。
• 主动触发刷新（可选）：若需实时响应密钥变更，可在检测到变更时调用Kafka Connect的REST API（POST /connectors/{connector-name}/config/refresh）主动触发连接器配置刷新。

参考代码示例 #

自定义VaultConfigProvider核心实现 #

import org.apache.kafka.common.config.provider.ConfigData;
import org.apache.kafka.common.config.provider.ConfigProvider;
import org.apache.kafka.common.config.Configurable;
import java.io.Closeable;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

public class VaultConfigProvider implements ConfigProvider, Configurable, Closeable {
    private VaultClient vaultClient;
    private ScheduledExecutorService scheduler;
    private AtomicBoolean cacheInvalidated = new AtomicBoolean(false);
    private String currentSecretVersion;
    private long refreshIntervalMs = 300000; // 默认5分钟检测间隔

    @Override
    public void configure(Map<String, ?> configs) {
        // 初始化Vault连接参数
        String vaultUrl = (String) configs.get("vault.url");
        String token = (String) configs.get("vault.token");
        this.vaultClient = new VaultClient(vaultUrl, token);
        
        // 启动密钥变更检测定时任务
        this.refreshIntervalMs = Long.parseLong((String) configs.getOrDefault("refresh.interval.ms", "300000"));
        this.scheduler = Executors.newSingleThreadScheduledExecutor();
        this.scheduler.scheduleAtFixedRate(this::checkSecretChanges, 0, refreshIntervalMs, TimeUnit.MILLISECONDS);
    }

    @Override
    public ConfigData get(String path) {
        return get(path, null);
    }

    @Override
    public ConfigData get(String path, Set<String> keys) {
        // 缓存失效或首次加载时，拉取最新密钥
        if (cacheInvalidated.get() || currentSecretVersion == null) {
            Map<String, String> secretData = vaultClient.fetchSecret(path, keys);
            currentSecretVersion = vaultClient.getSecretVersion(path);
            cacheInvalidated.set(false);
            // 设置过期时间为下一次检测时间，让Connect自动刷新
            long expiration = System.currentTimeMillis() + refreshIntervalMs;
            return new ConfigData(secretData, expiration);
        }
        // 缓存有效时返回缓存数据
        Map<String, String> cachedData = vaultClient.getCachedSecret(path, keys);
        return new ConfigData(cachedData, System.currentTimeMillis() + refreshIntervalMs);
    }

    private void checkSecretChanges() {
        try {
            String latestVersion = vaultClient.getSecretVersion("path/to/db/secret");
            if (!latestVersion.equals(currentSecretVersion)) {
                cacheInvalidated.set(true);
                // 可选：主动调用Connect API触发刷新
                triggerConnectorConfigRefresh("my-db-connector");
            }
        } catch (Exception e) {
            // 异常处理：日志记录等
            e.printStackTrace();
        }
    }

    private void triggerConnectorConfigRefresh(String connectorName) {
        // 实现调用Kafka Connect REST API逻辑，示例简化
        // 实际需用HttpClient发送POST请求到http://connect-host:8083/connectors/{connectorName}/config/refresh
    }

    @Override
    public void close() {
        if (scheduler != null) {
            scheduler.shutdown();
        }
    }

    // 简化的Vault客户端实现
    private static class VaultClient {
        private final String vaultUrl;
        private final String token;
        private Map<String, String> cachedSecret;
        private String secretVersion;

        public VaultClient(String vaultUrl, String token) {
            this.vaultUrl = vaultUrl;
            this.token = token;
        }

        public Map<String, String> fetchSecret(String path, Set<String> keys) {
            // 调用Vault API获取密钥的逻辑
            return Map.of("db.password", "new-updated-password");
        }

        public String getSecretVersion(String path) {
            // 调用Vault API获取当前密钥版本
            return "v2";
        }

        public Map<String, String> getCachedSecret(String path, Set<String> keys) {
            return cachedSecret;
        }
    }
}

Kafka Connect配置示例 #

在连接器配置中引用自定义提供者：

# 配置Vault提供者
config.providers=vault
config.providers.vault.class=com.example.VaultConfigProvider
config.providers.vault.param.vault.url=http://vault-host:8200
config.providers.vault.param.vault.token=vault-root-token
config.providers.vault.param.refresh.interval.ms=60000

# 连接器配置引用Vault密钥
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
connection.url=jdbc:mysql://db-host:3306/mydb
connection.user=${vault:/path/to/db/secret:db.user}
connection.password=${vault:/path/to/db/secret:db.password}

关键注意事项 #

• 缓存策略：合理设置检测间隔，避免频繁调用Vault API引发性能问题。
• 异常处理：处理Vault连接失败、权限不足等异常，防止影响连接器运行。
• API权限：若使用主动刷新，确保调用方拥有Kafka Connect REST端点的访问权限。

内容的提问来源于stack exchange，提问作者user3783064