[CI] Speed up CI test and fix flaky test (#1925)

### Motivation

KoP test is too slow, and there are couple of test very flaky and always
timeout.

### Modifications

Parallelize KoP uni tests with multiple jobs after regroup the tests
into corresponding packages
Add a testing matrix for github workflow to spin up one job for each
group of tests
Increase timeout to fix some flaky tests

.github/workflows/pr-tests.yml

@@ -15,66 +15,200 @@ concurrency:
   cancel-in-progress: true
 
 jobs:
-  build:
-
+  basic-validation:
+    name: Style check and basic unit tests
     runs-on: ubuntu-latest
-
     steps:
-    - uses: actions/checkout@v1
-    - name: Set up JDK 17
-      uses: actions/setup-java@v1
-      with:
-        java-version: 17
+      - uses: actions/checkout@v1
+      - name: Set up JDK 17
+        uses: actions/setup-java@v1
+        with:
+          java-version: 17
 
-    - name: License check
-      run: mvn -ntp -B license:check
+      - name: License check
+        run: mvn -ntp -B license:check
 
-    - name: Style check
-      run: mvn -ntp -B checkstyle:check
+      - name: Style check
+        run: mvn -ntp -B checkstyle:check
 
-    - name: Build with Maven skipTests
-      run: mvn clean install -ntp -B -DskipTests
+      - name: Build with Maven skipTests
+        run: mvn clean install -ntp -B -DskipTests
 
-    - name: Building JavaDocs
-      run: mvn -ntp -B javadoc:jar
+      - name: Building JavaDocs
+        run: mvn -ntp -B javadoc:jar
 
-    - name: Spotbugs check
-      run: mvn -ntp -B spotbugs:check
+      - name: Spotbugs check
+        run: mvn -ntp -B spotbugs:check
 
-    - name: kafka-impl test after build
-      run: mvn test -ntp -B -DfailIfNoTests=false -pl kafka-impl
+      - name: kafka-impl test after build
+        run: mvn test -ntp -B -DfailIfNoTests=false -pl kafka-impl
 
-    - name: schema registry test after build
-      run: mvn test -ntp -B -DfailIfNoTests=false -pl schema-registry
+      - name: schema registry test after build
+        run: mvn test -ntp -B -DfailIfNoTests=false -pl schema-registry
 
-    - name: kafka-payload-processor test after build
-      run: mvn test -ntp -B -DfailIfNoTests=false -pl kafka-payload-processor-shaded-tests
+      - name: kafka-payload-processor test after build
+        run: mvn test -ntp -B -DfailIfNoTests=false -pl kafka-payload-processor-shaded-tests
 
-    - name: Start and init the oauth server
-      run: ./ci/init_hydra_oauth_server.sh
-      timeout-minutes: 5
+      - name: Start and init the oauth server
+        run: ./ci/init_hydra_oauth_server.sh
+        timeout-minutes: 5
 
-    - name: oauth-client test after build
-      run: mvn test -ntp -B -DfailIfNoTests=false -pl oauth-client
+      - name: oauth-client test after build
+        run: mvn test -ntp -B -DfailIfNoTests=false -pl oauth-client
 
-    - name: tests module
-      run: mvn test -ntp -B -DfailIfNoTests=false '-Dtest=!KafkaIntegration*Test' -pl tests
-      timeout-minutes: 60
+      - name: Upload to Codecov
+        uses: codecov/codecov-action@v3
 
-    - name: Upload to Codecov
-      uses: codecov/codecov-action@v3
+      - name: package surefire artifacts
+        if: failure()
+        run: |
+          rm -rf artifacts
+          mkdir artifacts
+          find . -type d -name "*surefire*" -exec cp --parents -R {} artifacts/ \;
+          zip -r artifacts.zip artifacts
 
-    - name: package surefire artifacts
-      if: failure()
-      run: |
-        rm -rf artifacts
-        mkdir artifacts
-        find . -type d -name "*surefire*" -exec cp --parents -R {} artifacts/ \;
-        zip -r artifacts.zip artifacts
+      - uses: actions/upload-artifact@master
+        name: upload surefire-artifacts
+        if: failure()
+        with:
+          name: surefire-artifacts
+          path: artifacts.zip
 
-    - uses: actions/upload-artifact@master
-      name: upload surefire-artifacts
-      if: failure()
-      with:
-        name: surefire-artifacts
-        path: artifacts.zip
+  kop-unit-tests:
+    name: Unit Test (${{ matrix.test.name }})
+    runs-on: ubuntu-latest
+    timeout-minutes: 60
+    strategy:
+      # other jobs should run even if one test fails
+      fail-fast: false
+      matrix:
+        test: [
+          {
+            name: "admin test",
+            scripts: [
+              "mvn test -ntp -B -DfailIfNoTests=false '-Dtest=io.streamnative.pulsar.handlers.kop.admin.*Test' -pl tests"
+            ]
+          },
+          {
+            name: "compatibility test",
+            scripts: [
+              "mvn test -ntp -B -DfailIfNoTests=false '-Dtest=io.streamnative.pulsar.handlers.kop.compatibility.*Test' -pl tests",
+              "mvn test -ntp -B -DfailIfNoTests=false '-Dtest=io.streamnative.pulsar.handlers.kop.compatibility..*.*Test' -pl tests"
+            ]
+          },
+          {
+            name: "coordinator test",
+            scripts: [
+              "mvn test -ntp -B -DfailIfNoTests=false '-Dtest=io.streamnative.pulsar.handlers.kop.coordinator.*Test' -pl tests",
+              "mvn test -ntp -B -DfailIfNoTests=false '-Dtest=io.streamnative.pulsar.handlers.kop.coordinator..*.*Test' -pl tests"
+            ]
+          },
+          {
+            name: "end to end test",
+            scripts: [
+              "mvn test -ntp -B -DfailIfNoTests=false '-Dtest=io.streamnative.pulsar.handlers.kop.e2e.*Test' -pl tests"
+            ]
+          },
+          {
+            name: "format test",
+            scripts: [
+              "mvn test -ntp -B -DfailIfNoTests=false '-Dtest=io.streamnative.pulsar.handlers.kop.format.*Test' -pl tests"
+            ]
+          },
+          {
+            name: "metadata test",
+            scripts: [
+              "mvn test -ntp -B -DfailIfNoTests=false '-Dtest=io.streamnative.pulsar.handlers.kop.metadata.*Test' -pl tests"
+            ]
+          },
+          {
+            name: "metrics test",
+            scripts: [
+              "mvn test -ntp -B -DfailIfNoTests=false '-Dtest=io.streamnative.pulsar.handlers.kop.metrics.*Test' -pl tests"
+            ]
+          },
+          {
+            name: "producer test",
+            scripts: [
+              "mvn test -ntp -B -DfailIfNoTests=false '-Dtest=io.streamnative.pulsar.handlers.kop.producer.*Test' -pl tests"
+            ]
+          },
+          {
+            name: "schema test",
+            scripts: [
+              "mvn test -ntp -B -DfailIfNoTests=false '-Dtest=io.streamnative.pulsar.handlers.kop.schemaregistry.model.impl.*Test' -pl tests"
+            ]
+          },
+          {
+            name: "security test",
+            scripts: [
+              "mvn test -ntp -B -DfailIfNoTests=false '-Dtest=io.streamnative.pulsar.handlers.kop.security.*Test' -pl tests",
+              "mvn test -ntp -B -DfailIfNoTests=false '-Dtest=io.streamnative.pulsar.handlers.kop.security.*.*Test' -pl tests"
+            ]
+          },
+          {
+            name: "storage test",
+            scripts: [
+              "mvn test -ntp -B -DfailIfNoTests=false '-Dtest=io.streamnative.pulsar.handlers.kop.storage.*Test' -pl tests"
+            ]
+          },
+          {
+            name: "streams test",
+            scripts: [
+              "mvn test -ntp -B -DfailIfNoTests=false '-Dtest=io.streamnative.pulsar.handlers.kop.streams.*Test' -pl tests"
+            ]
+          },
+          {
+            name: "util test",
+            scripts: [
+              "mvn test -ntp -B -DfailIfNoTests=false '-Dtest=io.streamnative.pulsar.handlers.kop.util.timer.*Test' -pl tests"
+            ]
+          },
+          {
+            name: "other test",
+            scripts: [
+              "mvn test -ntp -B -DfailIfNoTests=false '-Dtest=io.streamnative.pulsar.handlers.kop.*Test' -pl tests"
+            ]
+          },
+        ]
+    steps:
+      - uses: actions/checkout@v3
+
+      - name: Set up JDK 17
+        uses: actions/setup-java@v1
+        with:
+          java-version: 17
+
+      - uses: actions/setup-python@v4
+        with:
+          python-version: '3.10'
+
+      - name: Install jq
+        run: sudo apt-get install -y jq
+
+      - name: Build with Maven skipTests
+        run: mvn clean install -ntp -B -DskipTests
+
+      - name: Start and init the oauth server
+        run: ./ci/init_hydra_oauth_server.sh
+        timeout-minutes: 5
+
+      - name: ${{ matrix.test.name }}
+        run: |
+          echo '${{ matrix.test.scripts }}'
+          scripts=$(echo '${{ toJson(matrix.test.scripts) }}' | jq -r '.[]')
+          IFS=$'\n'   # change the internal field separator to newline
+          echo $scripts
+          for script in $scripts
+          do
+            bash -c "${script}"
+          done
+          unset IFS  # revert the internal field separator back to default
+
+  unit-test-check:
+    name: Unit Test Check
+    runs-on: ubuntu-latest
+    needs: kop-unit-tests # This job will only run if all 'kop-unit-tests' jobs have completed successfully
+    steps:
+      - name: Check
+        run: echo "All tests have passed!"

tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaAdminTest.java → tests/src/test/java/io/streamnative/pulsar/handlers/kop/admin/KafkaAdminTest.java

@@ -11,7 +11,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.streamnative.pulsar.handlers.kop;
+package io.streamnative.pulsar.handlers.kop.admin;
 
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
@@ -21,6 +21,8 @@
 
 import com.google.common.collect.Sets;
 import io.jsonwebtoken.lang.Maps;
+import io.streamnative.pulsar.handlers.kop.KafkaLogConfig;
+import io.streamnative.pulsar.handlers.kop.KopProtocolHandlerTestBase;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.ArrayList;

tests/src/test/java/io/streamnative/pulsar/handlers/kop/KopBrokerLookupManagerTest.java → tests/src/test/java/io/streamnative/pulsar/handlers/kop/admin/KopBrokerLookupManagerTest.java

@@ -11,11 +11,14 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.streamnative.pulsar.handlers.kop;
+package io.streamnative.pulsar.handlers.kop.admin;
 
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertTrue;
 
+import io.streamnative.pulsar.handlers.kop.KopBrokerLookupManager;
+import io.streamnative.pulsar.handlers.kop.KopProtocolHandlerTestBase;
+import io.streamnative.pulsar.handlers.kop.LookupClient;
 import java.util.Collections;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.common.policies.data.TenantInfo;

tests/src/test/java/io/streamnative/pulsar/handlers/kop/OffsetResetTest.java → tests/src/test/java/io/streamnative/pulsar/handlers/kop/admin/OffsetResetTest.java

@@ -11,12 +11,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.streamnative.pulsar.handlers.kop;
+package io.streamnative.pulsar.handlers.kop.admin;
 
 import static org.apache.pulsar.common.naming.TopicName.PARTITIONED_TOPIC_SUFFIX;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertTrue;
 
+import io.streamnative.pulsar.handlers.kop.KopProtocolHandlerTestBase;
 import io.streamnative.pulsar.handlers.kop.coordinator.group.GroupMetadataConstants;
 import io.streamnative.pulsar.handlers.kop.coordinator.group.GroupMetadataManager.BaseKey;
 import io.streamnative.pulsar.handlers.kop.coordinator.group.GroupMetadataManager.OffsetKey;

tests/src/test/java/io/streamnative/pulsar/handlers/kop/TransactionTest.java → tests/src/test/java/io/streamnative/pulsar/handlers/kop/coordinator/transaction/TransactionTest.java

@@ -11,7 +11,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.streamnative.pulsar.handlers.kop;
+package io.streamnative.pulsar.handlers.kop.coordinator.transaction;
 
 import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -22,9 +22,8 @@
 import static org.testng.Assert.fail;
 
 import com.google.common.collect.ImmutableMap;
-import io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionCoordinator;
-import io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionState;
-import io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionStateManager;
+import io.streamnative.pulsar.handlers.kop.KafkaProtocolHandler;
+import io.streamnative.pulsar.handlers.kop.KopProtocolHandlerTestBase;
 import io.streamnative.pulsar.handlers.kop.scala.Either;
 import io.streamnative.pulsar.handlers.kop.storage.PartitionLog;
 import java.time.Duration;
@@ -111,25 +110,25 @@ protected static Object[][] produceConfigProvider() {
         };
     }
 
-    @Test(timeOut = 1000 * 10, dataProvider = "produceConfigProvider")
+    @Test(timeOut = 1000 * 30, dataProvider = "produceConfigProvider")
     public void readCommittedTest(boolean isBatch) throws Exception {
         basicProduceAndConsumeTest("read-committed-test", "txn-11", "read_committed", isBatch);
     }
 
-    @Test(timeOut = 1000 * 10, dataProvider = "produceConfigProvider")
+    @Test(timeOut = 1000 * 30, dataProvider = "produceConfigProvider")
     public void readUncommittedTest(boolean isBatch) throws Exception {
         basicProduceAndConsumeTest("read-uncommitted-test", "txn-12", "read_uncommitted", isBatch);
     }
 
-    @Test(timeOut = 1000 * 10)
+    @Test(timeOut = 1000 * 30)
     public void testInitTransaction() {
         final KafkaProducer<Integer, String> producer = buildTransactionProducer("prod-1");
 
         producer.initTransactions();
         producer.close();
     }
 
-    @Test(timeOut = 1000 * 10)
+    @Test(timeOut = 1000 * 30)
     public void testMultiCommits() throws Exception {
         final String topic = "test-multi-commits";
         final KafkaProducer<Integer, String> producer1 = buildTransactionProducer("X1");
@@ -278,7 +277,7 @@ public void offsetCommitTest() throws Exception {
         txnOffsetTest("txn-offset-commit-test", 10, true);
     }
 
-    @Test(timeOut = 1000 * 10)
+    @Test(timeOut = 3000 * 10)
     public void offsetAbortTest() throws Exception {
         txnOffsetTest("txn-offset-abort-test", 10, false);
     }

tests/src/test/java/io/streamnative/pulsar/handlers/kop/TransactionWithOAuthBearerAuthTest.java → tests/src/test/java/io/streamnative/pulsar/handlers/kop/coordinator/transaction/TransactionWithOAuthBearerAuthTest.java

@@ -11,11 +11,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.streamnative.pulsar.handlers.kop;
+package io.streamnative.pulsar.handlers.kop.coordinator.transaction;
 
 import com.google.common.collect.Sets;
+import io.streamnative.pulsar.handlers.kop.security.oauth.HydraOAuthUtils;
 import io.streamnative.pulsar.handlers.kop.security.oauth.OauthLoginCallbackHandler;
 import io.streamnative.pulsar.handlers.kop.security.oauth.OauthValidatorCallbackHandler;
+import io.streamnative.pulsar.handlers.kop.security.oauth.SaslOAuthKopHandlersTest;
 import java.net.URL;
 import java.util.Properties;
 import lombok.extern.slf4j.Slf4j;