Merge branch '2.6.1-RC' into dependabot/maven/org.apache.kafka-kafka-clients-5.3.0-ccs

Author: Claudenw

.github/workflows/manual.yml

@@ -0,0 +1,110 @@
+#
+# Copyright 2024 Copyright 2022 Aiven Oy and
+# bigquery-connector-for-apache-kafka project contributors
+#
+# This software contains code derived from the Confluent BigQuery
+# Kafka Connector, Copyright Confluent, Inc, which in turn
+# contains code derived from the WePay BigQuery Kafka Connector,
+# Copyright WePay, Inc.
+#
+# Licensed 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.
+#
+
+# Workflow to check pull requests and new commits to main branches
+# This checks the source in the state as if after the merge.
+name: Manual workflow
+on:
+  workflow_dispatch:
+
+permissions:
+  contents: write
+  pull-requests: write
+  issues: write
+
+
+# Disallow concurrent runs for the same PR by cancelling in-progress runs
+# when new commits are pushed
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  build:
+    name: Build
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v2
+      - name: Dump GitHub context
+        env:
+          GITHUB_CONTEXT: ${{ toJson(github) }}
+        run: echo "$GITHUB_CONTEXT"
+      - name: Test
+        env:
+          TEST_GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          GCP_CREDENTIALS: ${{ secrets.GCP_CREDENTIALS }}
+        run: |
+          echo ${#TEST_GITHUB_TOKEN}
+          echo ${#GCP_CREDENTIALS}
+#      - name: Set up JDK 17
+#        uses: actions/setup-java@v4
+#        with:
+#          distribution: 'adopt'
+#          java-version: 17
+#          cache: maven
+#      - name: License header check
+#        run: |
+#          mvn -ntp license:remove license:format
+#          if [[ -n $(git status -s) ]]; then
+#            echo 1>&2 'Some files do not have the correct license header:'
+#            git diff --name-only 1>&2
+#            echo 1>&2 'Please update the license headers for these files by running `mvn license:remove license:format`'
+#            exit 1
+#          fi
+#      - name: Build (Maven)
+#        run: mvn -P ci --batch-mode clean package -DskipTests
+#      - name: Unit tests (Maven)
+#        run: mvn -ntp -P ci --batch-mode test
+#      - name: "upload build failure reports"
+#        uses: actions/upload-artifact@v4
+#        if: failure()
+#        with:
+#          name: unit-test-results
+#          path: |
+#            **/target/*-reports/**
+#          retention-days: 1
+#      - name: Integration tests (Maven)
+#        env:
+#          # Necessary for client builder integration tests that run with
+#          # default application credentials
+#          CREDENTIALS_JSON: ${{ secrets.GCP_CREDENTIALS }}
+#          GOOGLE_APPLICATION_CREDENTIALS: /tmp/creds.json
+#          KCBQ_TEST_KEYFILE:    ${{ secrets.GCP_CREDENTIALS }}
+#          KCBQ_TEST_KEY_SOURCE: JSON
+#          KCBQ_TEST_PROJECT:    ${{ secrets.KCBQ_TEST_PROJECT }}
+#          KCBQ_TEST_DATASET:    ${{ secrets.KCBQ_TEST_DATASET }}
+#          KCBQ_TEST_BUCKET:     ${{ secrets.KCBQ_TEST_BUCKET }}
+#        run: |
+#          export
+#          echo "$CREDENTIALS_JSON" > /tmp/creds.json
+#          export KCBQ_TEST_TABLE_SUFFIX=_$(date +%s)_$RANDOM
+#          mvn -ntp -P ci -Dskip.unit.tests=true verify
+#      - name: Upload integration test results (Maven)
+#        if: always()
+#        uses: actions/upload-artifact@v4
+#        with:
+#          path: |
+#            **/target/failsafe-reports/*
+#          name: integration-test-results
+#          retention-days: 3

kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/config/BigQuerySinkConfig.java

@@ -35,11 +35,14 @@
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -51,11 +54,22 @@
 import org.apache.kafka.common.config.types.Password;
 import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.sink.SinkConnector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Base class for connector and task configs; contains properties shared between the two of them.
  */
 public class BigQuerySinkConfig extends AbstractConfig {
+
+  private static final Logger logger = LoggerFactory.getLogger(BigQuerySinkConfig.class);
+
+  public static final String DEPRECATED_DOC = "(DEPRECATED)";
+  public static final String GCS_LOAD_DEPRECATION_NOTICE =
+      "GCS batch loading has been deprecated and will be removed in a future major release.";
+  public static final String DECORATOR_SYNTAX_DEPRECATION_NOTICE =
+      "Use of partition decorator syntax has been deprecated and will be removed in a future release.";
+
   // Values taken from https://github.com/apache/kafka/blob/1.1.1/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SinkConnectorConfig.java#L33
   public static final String TOPICS_CONFIG = SinkConnector.TOPICS_CONFIG;
   public static final String TOPICS_DEFAULT = "";
@@ -174,7 +188,7 @@ public class BigQuerySinkConfig extends AbstractConfig {
   private static final List<String> ENABLE_BATCH_DEFAULT = Collections.emptyList();
   private static final ConfigDef.Importance ENABLE_BATCH_IMPORTANCE = ConfigDef.Importance.LOW;
   private static final String ENABLE_BATCH_DOC =
-      "Beta Feature; use with caution: The sublist of topics to be batch loaded through GCS";
+      "The sublist of topics to be batch loaded through GCS.";
   private static final ConfigDef.Type BATCH_LOAD_INTERVAL_SEC_TYPE = ConfigDef.Type.INT;
   private static final Integer BATCH_LOAD_INTERVAL_SEC_DEFAULT = 120;
   private static final ConfigDef.Importance BATCH_LOAD_INTERVAL_SEC_IMPORTANCE =
@@ -541,12 +555,31 @@ public class BigQuerySinkConfig extends AbstractConfig {
 
   protected BigQuerySinkConfig(ConfigDef config, Map<String, String> properties) {
     super(config, properties);
+    logDeprecationWarnings();
   }
 
   public BigQuerySinkConfig(Map<String, String> properties) {
     this(getConfig(), properties);
   }
 
+  private void logDeprecationWarnings() {
+    if (!getList(ENABLE_BATCH_CONFIG).isEmpty()) {
+      logger.warn(
+          GCS_LOAD_DEPRECATION_NOTICE
+              + " To disable this feature, remove the {} property from the connector configuration",
+          ENABLE_BATCH_CONFIG
+      );
+    }
+
+    if (getBoolean(BIGQUERY_PARTITION_DECORATOR_CONFIG)) {
+      logger.warn(
+          DECORATOR_SYNTAX_DEPRECATION_NOTICE
+            + " To disable this feature, set the {} property to false in the connector configuration",
+          BIGQUERY_PARTITION_DECORATOR_CONFIG
+      );
+    }
+  }
+
   /**
    * Return the ConfigDef object used to define this config's fields.
    *
@@ -579,25 +612,25 @@ public static ConfigDef getConfig() {
             ENABLE_BATCH_TYPE,
             ENABLE_BATCH_DEFAULT,
             ENABLE_BATCH_IMPORTANCE,
-            ENABLE_BATCH_DOC
+            deprecatedGcsLoadDoc(ENABLE_BATCH_DOC)
         ).define(
             BATCH_LOAD_INTERVAL_SEC_CONFIG,
             BATCH_LOAD_INTERVAL_SEC_TYPE,
             BATCH_LOAD_INTERVAL_SEC_DEFAULT,
             BATCH_LOAD_INTERVAL_SEC_IMPORTANCE,
-            BATCH_LOAD_INTERVAL_SEC_DOC
+            deprecatedGcsLoadDoc(BATCH_LOAD_INTERVAL_SEC_DOC)
         ).define(
             GCS_BUCKET_NAME_CONFIG,
             GCS_BUCKET_NAME_TYPE,
             GCS_BUCKET_NAME_DEFAULT,
             GCS_BUCKET_NAME_IMPORTANCE,
-            GCS_BUCKET_NAME_DOC
+            deprecatedGcsLoadDoc(GCS_BUCKET_NAME_DOC)
         ).define(
             GCS_FOLDER_NAME_CONFIG,
             GCS_FOLDER_NAME_TYPE,
             GCS_FOLDER_NAME_DEFAULT,
             GCS_FOLDER_NAME_IMPORTANCE,
-            GCS_FOLDER_NAME_DOC
+            deprecatedGcsLoadDoc(GCS_FOLDER_NAME_DOC)
         ).define(
             PROJECT_CONFIG,
             PROJECT_TYPE,
@@ -691,7 +724,7 @@ public static ConfigDef getConfig() {
             AUTO_CREATE_BUCKET_TYPE,
             AUTO_CREATE_BUCKET_DEFAULT,
             AUTO_CREATE_BUCKET_IMPORTANCE,
-            AUTO_CREATE_BUCKET_DOC
+            deprecatedGcsLoadDoc(AUTO_CREATE_BUCKET_DOC)
         ).define(
             ALLOW_NEW_BIGQUERY_FIELDS_CONFIG,
             ALLOW_NEW_BIGQUERY_FIELDS_TYPE,
@@ -776,13 +809,13 @@ public static ConfigDef getConfig() {
             BIGQUERY_MESSAGE_TIME_PARTITIONING_CONFIG_TYPE,
             BIGQUERY_MESSAGE_TIME_PARTITIONING_DEFAULT,
             BIGQUERY_MESSAGE_TIME_PARTITIONING_IMPORTANCE,
-            BIGQUERY_MESSAGE_TIME_PARTITIONING_DOC
+            deprecatedPartitionSyntaxDoc(BIGQUERY_MESSAGE_TIME_PARTITIONING_DOC)
         ).define(
             BIGQUERY_PARTITION_DECORATOR_CONFIG,
             BIGQUERY_PARTITION_DECORATOR_CONFIG_TYPE,
             BIGQUERY_PARTITION_DECORATOR_DEFAULT,
             BIGQUERY_PARTITION_DECORATOR_IMPORTANCE,
-            BIGQUERY_PARTITION_DECORATOR_DOC
+            deprecatedPartitionSyntaxDoc(BIGQUERY_PARTITION_DECORATOR_DOC)
         ).define(
             BIGQUERY_TIMESTAMP_PARTITION_FIELD_NAME_CONFIG,
             BIGQUERY_TIMESTAMP_PARTITION_FIELD_NAME_TYPE,
@@ -879,18 +912,6 @@ public boolean visible(String s, Map<String, Object> map) {
         );
   }
 
-  public static boolean upsertDeleteEnabled(Map<String, String> props) {
-    String upsertStr = props.get(UPSERT_ENABLED_CONFIG);
-    String deleteStr = props.get(DELETE_ENABLED_CONFIG);
-    return Boolean.TRUE.toString().equalsIgnoreCase(upsertStr)
-        || Boolean.TRUE.toString().equalsIgnoreCase(deleteStr);
-  }
-
-  public static boolean gcsBatchLoadingEnabled(Map<String, String> props) {
-    String batchLoadStr = props.get(ENABLE_BATCH_CONFIG);
-    return batchLoadStr != null && !batchLoadStr.isEmpty();
-  }
-
   /**
    * Used in conjunction with {@link com.wepay.kafka.connect.bigquery.BigQuerySinkConnector#validate(Map)} to perform
    * preflight configuration checks. Simple validations that only require a single property value at a time (such as
@@ -936,13 +957,6 @@ public GcpClientBuilder.KeySource getKeySource() {
     }
   }
 
-  /**
-   * Returns the keyfile
-   */
-  public String getKeyFile() {
-    return Optional.ofNullable(getPassword(KEYFILE_CONFIG)).map(Password::value).orElse(null);
-  }
-
   /**
    * Return a new instance of the configured Schema Converter.
    *
@@ -1134,4 +1148,17 @@ private static String header(String text) {
     return wrapper + "\n" + text + "\n" + wrapper + "\n";
   }
 
+  private static String deprecatedGcsLoadDoc(String doc) {
+
+    return deprecatedDoc(doc, GCS_LOAD_DEPRECATION_NOTICE);
+  }
+
+  public static String deprecatedPartitionSyntaxDoc(String doc) {
+    return deprecatedDoc(doc, DECORATOR_SYNTAX_DEPRECATION_NOTICE);
+  }
+
+  private static String deprecatedDoc(String doc, String notice) {
+    return DEPRECATED_DOC + " " + doc + " Warning: " + notice;
+  }
+
 }

kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/write/row/GcsToBqWriter.java

@@ -128,8 +128,21 @@ public void writeRows(SortedMap<SinkRecord, RowToInsert> rows,
 
     // Check if the table specified exists
     // This error shouldn't be thrown. All tables should be created by the connector at startup
-    if (autoCreateTables && bigQuery.getTable(tableId) == null) {
-      attemptTableCreate(tableId, new ArrayList<>(rows.keySet()));
+    int lookupAttempts = 0;
+    boolean lookupSuccess = bigQuery.getTable(tableId) != null;
+    BigQueryException lookupException = null;
+
+    if (autoCreateTables && !lookupSuccess) {
+      logger.info("Table {} was not found. Creating the table automatically.", tableId);
+      schemaManager.createTable(tableId, new ArrayList<>(rows.keySet()));
+      while (!lookupSuccess && lookupAttempts <= retries) {
+        waitRandomTime();
+        lookupSuccess = bigQuery.getTable(tableId) != null;
+        lookupAttempts++;
+      }
+    }
+    if (!lookupSuccess) {
+      throw new BigQueryConnectException("Failed to lookup table " + tableId, lookupException);
     }
 
     int attemptCount = 0;
@@ -200,13 +213,4 @@ private void waitRandomTime() throws InterruptedException {
     time.sleep(retryWaitMs + random.nextInt(WAIT_MAX_JITTER));
   }
 
-  private void attemptTableCreate(TableId tableId, List<SinkRecord> records) {
-    try {
-      logger.info("Table {} does not exist, auto-creating table ", tableId);
-      schemaManager.createTable(tableId, records);
-    } catch (BigQueryException exception) {
-      throw new BigQueryConnectException(
-          "Failed to create table " + tableId, exception);
-    }
-  }
 }

pom.xml

@@ -44,7 +44,7 @@
         <confluent.version>7.6.0</confluent.version>
         <debezium.version>0.6.2</debezium.version>
         <jackson.version>2.14.2</jackson.version>
-        <jetty.version>9.4.53.v20231009</jetty.version>
+        <jetty.version>9.4.56.v20240826</jetty.version>
         <kafka.version>3.9.1</kafka.version>
         <kafka.scala.version>2.12</kafka.scala.version>
         <slf4j.version>1.7.26</slf4j.version>