Parallelize integration tests, retain test output as separate artifacts

.github/workflows/prs_and_commits.yml

@@ -27,9 +27,9 @@ jobs:
           java-version: 8
           cache: maven
       - name: Build (Maven)
-        run: mvn --batch-mode clean package -DskipTests
+        run: mvn -P ci --batch-mode clean package -DskipTests
       - name: Unit tests (Maven)
-        run: mvn --batch-mode test
+        run: mvn -P ci --batch-mode test
       - name: Integration tests (Maven)
         env:
           # Necessary for client builder integration tests that run with
@@ -43,4 +43,11 @@ jobs:
           KCBQ_TEST_BUCKET:     ${{ secrets.KCBQ_TEST_BUCKET }}
         run: |
           echo "$CREDENTIALS_JSON" > /tmp/creds.json
-          mvn -Dskip.unit.tests=true verify
+          export KCBQ_TEST_TABLE_SUFFIX=_$(date +%s)_$RANDOM
+          mvn -P ci -Dskip.unit.tests=true verify
+      - name: Upload integration test results (Maven)
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          path: kcbq-connector/target/failsafe-reports/*
+          name: integration-test-results

kcbq-connector/pom.xml

@@ -46,6 +46,28 @@
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-server</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-servlet</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-servlets</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-client</artifactId>
+            <scope>test</scope>
         </dependency>
 
         <dependency>
@@ -63,8 +85,19 @@
         </dependency>
 
         <dependency>
-            <groupId>junit</groupId>
-            <artifactId>junit</artifactId>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-engine</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-params</artifactId>
+            <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>org.mockito</groupId>

kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/exception/BigQueryStorageWriteApiErrorResponses.java

@@ -23,6 +23,7 @@ public class BigQueryStorageWriteApiErrorResponses {
   private static final String NOT_FOUND = "Not found: table";
   private static final String TABLE_IS_DELETED = "Table is deleted";
   private static final String MESSAGE_TOO_LARGE = "MessageSize is too large";
+  private static final String APPEND_ROWS_REQUEST_TOO_LARGE = "AppendRows request too large";
   private static final String[] retriableCodes = {Code.INTERNAL.name(), Code.ABORTED.name(), Code.CANCELLED.name()};
   /*
    Below list is taken from :
@@ -126,7 +127,9 @@ public static boolean isNonRetriableStorageError(Exception exception) {
 
   public static boolean isMessageTooLargeError(String errorMessage) {
     return isMalformedRequest(errorMessage)
-        && errorMessage.contains(MESSAGE_TOO_LARGE);
+        && (errorMessage.contains(MESSAGE_TOO_LARGE)
+        || errorMessage.contains(APPEND_ROWS_REQUEST_TOO_LARGE)
+      );
   }
 
 }

kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/write/storage/StorageWriteApiRetryHandler.java

@@ -110,6 +110,7 @@ public void attemptTableOperation(BiConsumer<TableId, List<SinkRecord>> tableOpe
       if (BigQueryErrorResponses.isRateLimitExceededError(exception)) {
         // Can happen if several tasks try to create a table all at once; should be fine
         logger.info("Table appears to have been created by a different task");
+        setAdditionalRetriesAndWait();
         return;
       }
       throw new BigQueryStorageWriteApiConnectException(

kcbq-connector/src/main/resources/META-INF/services/org.apache.kafka.connect.sink.SinkConnector

@@ -0,0 +1 @@
+com.wepay.kafka.connect.bigquery.BigQuerySinkConnector

kcbq-connector/src/test/java/com/wepay/kafka/connect/bigquery/BigQuerySinkConnectorTest.java

@@ -19,9 +19,9 @@
 
 package com.wepay.kafka.connect.bigquery;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
 
 import com.wepay.kafka.connect.bigquery.api.SchemaRetriever;
 import com.wepay.kafka.connect.bigquery.config.BigQuerySinkTaskConfig;
@@ -30,13 +30,13 @@
 import java.util.Map;
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.sink.SinkRecord;
-import org.junit.BeforeClass;
-import org.junit.Test;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
 
 public class BigQuerySinkConnectorTest {
   private static SinkPropertiesFactory propertiesFactory;
 
-  @BeforeClass
+  @BeforeAll
   public static void initializePropertiesFactory() {
     propertiesFactory = new SinkPropertiesFactory();
   }
@@ -61,22 +61,22 @@ public void testTaskConfigs() {
       for (int j = 0; j < i; j++) {
         expectedProperties.put(BigQuerySinkTaskConfig.TASK_ID_CONFIG, Integer.toString(j));
         assertEquals(
-            "Connector properties should match task configs",
             expectedProperties,
-            taskConfigs.get(j)
+            taskConfigs.get(j),
+            "Connector properties should match task configs"
         );
         assertNotSame(
-            "Properties should not be referentially equal to task config",
             properties,
-            taskConfigs.get(j)
+            taskConfigs.get(j),
+            "Properties should not be referentially equal to task config"
         );
         // A little overboard, sure, but since it's only in the ballpark of 10,000 iterations this
         // should be fine
         for (int k = j + 1; k < i; k++) {
           assertNotSame(
-              "Task configs should not be referentially equal to each other",
               taskConfigs.get(j),
-              taskConfigs.get(k)
+              taskConfigs.get(k),
+              "Task configs should not be referentially equal to each other"
           );
         }
       }