Implementation:Risingwavelabs Risingwave SnowflakeJDBCSinkConfig

Property	Value
Component	risingwave-sink-jdbc
Language	Java
Package	com.risingwave.connector
Extends	JDBCSinkConfig
Lines	242
Source	SnowflakeJDBCSinkConfig.java

Overview

SnowflakeJDBCSinkConfig extends JDBCSinkConfig with Snowflake-specific authentication support, including key-pair authentication via PEM private keys. While the base JDBCSinkConfig supports only password-based authentication, this subclass adds three authentication methods:

1. Password authentication: Standard username/password (inherited from the base class).
2. Key-pair file authentication (key_pair_file): References a private key file path on the filesystem.
3. Key-pair object authentication (key_pair_object): Accepts PEM-encoded private key content directly, which is parsed and converted to a PrivateKey object using BouncyCastle.

The class uses the BouncyCastle cryptographic library for PEM parsing and supports both unencrypted (PrivateKeyInfo) and encrypted (PKCS8EncryptedPrivateKeyInfo) private keys. The BouncyCastle provider is registered in a static initializer block.

Code Reference

Source Location

java/connector-node/risingwave-sink-jdbc/src/main/java/com/risingwave/connector/SnowflakeJDBCSinkConfig.java

Signature

public class SnowflakeJDBCSinkConfig extends JDBCSinkConfig {
    @JsonCreator
    public SnowflakeJDBCSinkConfig(
            @JsonProperty(value = "jdbc.url") String jdbcUrl,
            @JsonProperty(value = "table.name") String tableName,
            @JsonProperty(value = "type") String sinkType);

    @Override
    public Connection getConnection() throws SQLException;

    public static void handleSnowflakeAuth(Properties props)
            throws IOException, GeneralSecurityException, OperatorCreationException, PKCSException, SQLException;

    public static PrivateKey loadPrivateKeyFromPem(String pemContent, String passphrase)
            throws IOException, GeneralSecurityException, OperatorCreationException, PKCSException;

    public static PrivateKey convertToPrivateKey(Object parsed, String passphrase)
            throws GeneralSecurityException, IOException, OperatorCreationException, PKCSException;
}

Imports

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import java.io.IOException;
import java.io.StringReader;
import java.security.GeneralSecurityException;
import java.security.PrivateKey;
import java.security.Security;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.Properties;
import org.bouncycastle.asn1.pkcs.PrivateKeyInfo;
import org.bouncycastle.jce.provider.BouncyCastleProvider;
import org.bouncycastle.openssl.PEMParser;
import org.bouncycastle.openssl.jcajce.JcaPEMKeyConverter;
import org.bouncycastle.openssl.jcajce.JceOpenSSLPKCS8DecryptorProviderBuilder;
import org.bouncycastle.operator.InputDecryptorProvider;
import org.bouncycastle.operator.OperatorCreationException;
import org.bouncycastle.pkcs.PKCS8EncryptedPrivateKeyInfo;
import org.bouncycastle.pkcs.PKCSException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

I/O Contract

Input (additional properties beyond JDBCSinkConfig)

Property Key	Field	Type	Description
auth.method	authMethod	String	Authentication method: "password", "key_pair_file", or "key_pair_object"
private_key_file	privateKeyFile	String	Path to the private key file (for key_pair_file auth)
private_key_file_pwd	privateKeyFilePwd	String	Passphrase for encrypted private keys
private_key_pem	privateKeyPem	String	PEM-encoded private key content (for key_pair_object auth)

Output

• getConnection(): Returns a java.sql.Connection to Snowflake configured with the appropriate authentication method, auto-commit setting, and READ_COMMITTED transaction isolation.

Authentication Flow

The handleSnowflakeAuth method processes the auth.method property:

auth.method	Behavior
null	Falls back to default password authentication (no extra processing)
password	Uses password already set in properties
key_pair_file	Sets authenticator=snowflake_jwt; expects private_key_file and optional private_key_file_pwd
key_pair_object	Parses PEM content from private_key_pem, converts to PrivateKey object, sets authenticator=snowflake_jwt

Private Key Parsing

The convertToPrivateKey method handles three PEM object types:

1. PrivateKeyInfo (unencrypted PKCS#8): Directly converts to PrivateKey.
2. PKCS8EncryptedPrivateKeyInfo (encrypted PKCS#8): Decrypts using the provided passphrase, then converts.
3. byte[] (raw DER): Attempts to parse as DER-encoded PKCS#8 private key info.

Usage Examples

Snowflake Sink with Key-Pair Object Authentication

CREATE SINK snowflake_sink FROM my_mv WITH (
    connector = 'jdbc',
    jdbc.url = 'jdbc:snowflake://account.snowflakecomputing.com',
    user = 'my_user',
    table.name = 'target_table',
    type = 'append-only',
    auth.method = 'key_pair_object',
    private_key_pem = '-----BEGIN PRIVATE KEY-----\nMIIEvg...\n-----END PRIVATE KEY-----'
);

Snowflake Sink with Key-Pair File Authentication

CREATE SINK snowflake_sink FROM my_mv WITH (
    connector = 'jdbc',
    jdbc.url = 'jdbc:snowflake://account.snowflakecomputing.com',
    user = 'my_user',
    table.name = 'target_table',
    type = 'append-only',
    auth.method = 'key_pair_file',
    private_key_file = '/path/to/rsa_key.p8',
    private_key_file_pwd = 'my_passphrase'
);

Programmatic PEM Key Loading

String pemContent = "-----BEGIN PRIVATE KEY-----\n...";
PrivateKey key = SnowflakeJDBCSinkConfig.loadPrivateKeyFromPem(pemContent, null);

Related Pages

• Risingwavelabs_Risingwave_JDBCSinkConfig -- Base configuration class extended by this class
• Risingwavelabs_Risingwave_JDBCSinkFactory -- Factory that routes Snowflake JDBC URLs to this config class
• Risingwavelabs_Risingwave_JdbcUtils -- Provides createBaseProperties used by getConnection()
• Risingwavelabs_Risingwave_SnowflakeDialect -- Snowflake SQL dialect used alongside this configuration
• Risingwavelabs_Risingwave_BatchAppendOnlyJDBCSink -- Sink writer used for Snowflake targets