merge master

snowflakedb · Oct 20, 2023 · 1aeaaf0 · 1aeaaf0
2 parents 58f67e1 + 055dfca
commit 1aeaaf0
Show file tree

Hide file tree

Showing 73 changed files with 3,369 additions and 725 deletions.
diff --git a/.github/scripts/profile.json.gpg b/.github/scripts/profile.json.gpg
diff --git a/.github/scripts/profile_azure.json.gpg b/.github/scripts/profile_azure.json.gpg
diff --git a/.github/scripts/profile_gcs.json.gpg b/.github/scripts/profile_gcs.json.gpg
diff --git a/.github/workflows/IntegrationTest.yml b/.github/workflows/IntegrationTest.yml
@@ -28,9 +28,9 @@ jobs:
         restore-keys: |
           ${{ runner.os }}-maven-
     - name: Install Python
-      uses: actions/setup-python@v1
+      uses: actions/setup-python@v4
       with:
-        python-version: '3.6'
+        python-version: '3.9'
         architecture: 'x64'
     - name: Decrypt profile.json in Snowflake Cloud ${{ matrix.snowflake_cloud }}
       run: ./.github/scripts/decrypt_secret.sh ${{ matrix.snowflake_cloud }}

diff --git a/.github/workflows/PerfTest.yml b/.github/workflows/PerfTest.yml
@@ -22,9 +22,9 @@ jobs:
         restore-keys: |
           ${{ runner.os }}-maven-
     - name: Install Python
-      uses: actions/setup-python@v1
+      uses: actions/setup-python@v4
       with:
-        python-version: '3.6'
+        python-version: '3.9'
         architecture: 'x64'
     - name: Decrypt snowflake.json
       run: ./.github/scripts/perf_test_decrypt_secret.sh

diff --git a/.github/workflows/snyk-issue.yml b/.github/workflows/snyk-issue.yml
@@ -4,10 +4,15 @@ on:
   schedule:
     - cron: '* */12 * * *'
 
+permissions:
+  contents: read
+  issues: write
+  pull-requests: write
+
 concurrency: snyk-issue
 
 jobs:
-  whitesource:
+  snyk:
     runs-on: ubuntu-latest
     steps:
     - name: checkout action

diff --git a/.github/workflows/snyk-pr.yml b/.github/workflows/snyk-pr.yml
@@ -3,8 +3,14 @@ on:
   pull_request:
     branches:
       - master
+
+permissions:
+  contents: read
+  issues: write
+  pull-requests: write
+
 jobs:
-  whitesource:
+  snyk:
     runs-on: ubuntu-latest
     if: ${{ github.event.pull_request.user.login == 'sfc-gh-snyk-sca-sa' }}
     steps:

diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
@@ -1,5 +1,5 @@
 repos:
 - repo: [email protected]:snowflakedb/casec_precommit.git
-  rev: v1.3
+  rev: v1.29
   hooks:
   - id: secret-scanner
diff --git a/pom.xml b/pom.xml
@@ -12,7 +12,7 @@
 
     <groupId>com.snowflake</groupId>
     <artifactId>snowflake-kafka-connector</artifactId>
-    <version>1.9.4</version>
+    <version>2.1.0</version>
     <packaging>jar</packaging>
     <name>Snowflake Kafka Connector</name>
     <description>Snowflake Kafka Connect Sink Connector</description>
@@ -342,7 +342,7 @@
         <dependency>
             <groupId>net.snowflake</groupId>
             <artifactId>snowflake-ingest-sdk</artifactId>
-            <version>2.0.2</version>
+            <version>2.0.3</version>
             <exclusions>
                 <exclusion>
                     <groupId>net.snowflake</groupId>
@@ -354,7 +354,7 @@
         <dependency>
             <groupId>org.apache.avro</groupId>
             <artifactId>avro</artifactId>
-            <version>1.11.1</version>
+            <version>1.11.3</version>
             <exclusions>
                 <exclusion>
                     <groupId>com.fasterxml.jackson.core</groupId>
@@ -377,12 +377,12 @@
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-core</artifactId>
-            <version>2.13.1</version>
+            <version>2.15.2</version>
         </dependency>
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
-            <version>2.13.4.2</version>
+            <version>2.15.2</version>
         </dependency>
         <dependency>
             <groupId>io.confluent</groupId>

diff --git a/pom_confluent.xml b/pom_confluent.xml
@@ -12,7 +12,7 @@
 
     <groupId>com.snowflake</groupId>
     <artifactId>snowflake-kafka-connector</artifactId>
-    <version>1.9.4</version>
+    <version>2.1.0</version>
     <packaging>jar</packaging>
     <name>Snowflake Kafka Connector</name>
     <description>Snowflake Kafka Connect Sink Connector</description>
@@ -386,7 +386,7 @@
         <dependency>
             <groupId>net.snowflake</groupId>
             <artifactId>snowflake-ingest-sdk</artifactId>
-            <version>2.0.1</version>
+            <version>2.0.3</version>
             <exclusions>
                 <exclusion>
                     <groupId>net.snowflake</groupId>
@@ -398,7 +398,7 @@
         <dependency>
             <groupId>org.apache.avro</groupId>
             <artifactId>avro</artifactId>
-            <version>1.11.1</version>
+            <version>1.11.3</version>
             <exclusions>
                 <exclusion>
                     <groupId>com.fasterxml.jackson.core</groupId>
@@ -420,13 +420,13 @@
         </dependency>
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
-            <artifactId>jackson-core</artifactId>
-            <version>2.13.1</version>
+            <artifactId>jackson-core</artifactId>            
+            <version>2.15.2</version>
         </dependency>
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
-            <version>2.13.4.2</version>
+            <version>2.15.2</version>
         </dependency>
         <dependency>
             <groupId>io.confluent</groupId>

diff --git a/src/main/java/com/snowflake/kafka/connector/SnowflakeSinkConnector.java b/src/main/java/com/snowflake/kafka/connector/SnowflakeSinkConnector.java
@@ -209,9 +209,13 @@ public Config validate(Map<String, String> connectorConfigs) {
       Utils.updateConfigErrorMessage(result, invalidKey, invalidProxyParams.get(invalidKey));
     }
 
-    // If private key or private key passphrase is provided through file, skip validation
-    if (connectorConfigs.getOrDefault(Utils.SF_PRIVATE_KEY, "").contains("${file:")
-        || connectorConfigs.getOrDefault(Utils.PRIVATE_KEY_PASSPHRASE, "").contains("${file:"))
+    // If using snowflake_jwt and authentication, and private key or private key passphrase is
+    // provided through file, skip validation
+    if (connectorConfigs
+            .getOrDefault(Utils.SF_AUTHENTICATOR, Utils.SNOWFLAKE_JWT)
+            .equals(Utils.SNOWFLAKE_JWT)
+        && (connectorConfigs.getOrDefault(Utils.SF_PRIVATE_KEY, "").contains("${file:")
+            || connectorConfigs.getOrDefault(Utils.PRIVATE_KEY_PASSPHRASE, "").contains("${file:")))
       return result;
 
     // We don't validate name, since it is not included in the return value
@@ -244,6 +248,28 @@ public Config validate(Map<String, String> connectorConfigs) {
         case "0013":
           Utils.updateConfigErrorMessage(result, Utils.SF_PRIVATE_KEY, " must be non-empty");
           break;
+        case "0026":
+          Utils.updateConfigErrorMessage(
+              result,
+              Utils.SF_OAUTH_CLIENT_ID,
+              " must be non-empty when using oauth authenticator");
+          break;
+        case "0027":
+          Utils.updateConfigErrorMessage(
+              result,
+              Utils.SF_OAUTH_CLIENT_SECRET,
+              " must be non-empty when using oauth authenticator");
+          break;
+        case "0028":
+          Utils.updateConfigErrorMessage(
+              result,
+              Utils.SF_OAUTH_REFRESH_TOKEN,
+              " must be non-empty when using oauth authenticator");
+          break;
+        case "0029":
+          Utils.updateConfigErrorMessage(
+              result, Utils.SF_AUTHENTICATOR, " is not a valid authenticator");
+          break;
         case "0002":
           Utils.updateConfigErrorMessage(
               result, Utils.SF_PRIVATE_KEY, " must be a valid PEM RSA private key");

diff --git a/src/main/java/com/snowflake/kafka/connector/SnowflakeSinkConnectorConfig.java b/src/main/java/com/snowflake/kafka/connector/SnowflakeSinkConnectorConfig.java
@@ -71,6 +71,10 @@ public class SnowflakeSinkConnectorConfig {
   static final String SNOWFLAKE_DATABASE = Utils.SF_DATABASE;
   static final String SNOWFLAKE_SCHEMA = Utils.SF_SCHEMA;
   static final String SNOWFLAKE_PRIVATE_KEY_PASSPHRASE = Utils.PRIVATE_KEY_PASSPHRASE;
+  static final String AUTHENTICATOR_TYPE = Utils.SF_AUTHENTICATOR;
+  static final String OAUTH_CLIENT_ID = Utils.SF_OAUTH_CLIENT_ID;
+  static final String OAUTH_CLIENT_SECRET = Utils.SF_OAUTH_CLIENT_SECRET;
+  static final String OAUTH_REFRESH_TOKEN = Utils.SF_OAUTH_REFRESH_TOKEN;
 
   // For Snowpipe Streaming client
   public static final String SNOWFLAKE_ROLE = Utils.SF_ROLE;
@@ -301,6 +305,46 @@ static ConfigDef newConfigDef() {
             6,
             ConfigDef.Width.NONE,
             SNOWFLAKE_ROLE)
+        .define(
+            AUTHENTICATOR_TYPE,
+            Type.STRING, // TODO: SNOW-889748 change to enum and add validator
+            Utils.SNOWFLAKE_JWT,
+            Importance.LOW,
+            "Authenticator for JDBC and streaming ingest sdk",
+            SNOWFLAKE_LOGIN_INFO,
+            7,
+            ConfigDef.Width.NONE,
+            AUTHENTICATOR_TYPE)
+        .define(
+            OAUTH_CLIENT_ID,
+            Type.STRING,
+            "",
+            Importance.HIGH,
+            "Client id of target OAuth integration",
+            SNOWFLAKE_LOGIN_INFO,
+            8,
+            ConfigDef.Width.NONE,
+            OAUTH_CLIENT_ID)
+        .define(
+            OAUTH_CLIENT_SECRET,
+            Type.STRING,
+            "",
+            Importance.HIGH,
+            "Client secret of target OAuth integration",
+            SNOWFLAKE_LOGIN_INFO,
+            9,
+            ConfigDef.Width.NONE,
+            OAUTH_CLIENT_SECRET)
+        .define(
+            OAUTH_REFRESH_TOKEN,
+            Type.STRING,
+            "",
+            Importance.HIGH,
+            "Refresh token for OAuth",
+            SNOWFLAKE_LOGIN_INFO,
+            10,
+            ConfigDef.Width.NONE,
+            OAUTH_REFRESH_TOKEN)
         // proxy
         .define(
             JVM_PROXY_HOST,

diff --git a/src/main/java/com/snowflake/kafka/connector/SnowflakeSinkTask.java b/src/main/java/com/snowflake/kafka/connector/SnowflakeSinkTask.java
@@ -256,8 +256,7 @@ public void stop() {
   @Override
   public void open(final Collection<TopicPartition> partitions) {
     long startTime = System.currentTimeMillis();
-    partitions.forEach(
-        tp -> this.sink.startTask(Utils.tableName(tp.topic(), this.topic2table), tp));
+    this.sink.startPartitions(partitions, this.topic2table);
     this.DYNAMIC_LOGGER.info(
         "task opened with {} partitions, execution time: {} milliseconds",
         partitions.size(),