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:
- Password authentication: Standard username/password (inherited from the base class).
- Key-pair file authentication (
key_pair_file): References a private key file path on the filesystem. - Key-pair object authentication (
key_pair_object): Accepts PEM-encoded private key content directly, which is parsed and converted to aPrivateKeyobject 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.Connectionto Snowflake configured with the appropriate authentication method, auto-commit setting, andREAD_COMMITTEDtransaction 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:
- PrivateKeyInfo (unencrypted PKCS#8): Directly converts to
PrivateKey. - PKCS8EncryptedPrivateKeyInfo (encrypted PKCS#8): Decrypts using the provided passphrase, then converts.
- 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
createBasePropertiesused bygetConnection() - Risingwavelabs_Risingwave_SnowflakeDialect -- Snowflake SQL dialect used alongside this configuration
- Risingwavelabs_Risingwave_BatchAppendOnlyJDBCSink -- Sink writer used for Snowflake targets