[codex] Add shared rich integration test suite

.github/workflows/scripts/pr-tests/.pinot_tests_integration.sh

@@ -50,23 +50,43 @@ print_surefire_dumps() {
   done <<< "${dump_files}"
 }
 
-# Integration Tests
-cd pinot-integration-tests || exit 1
-if [ "$RUN_TEST_SET" == "1" ]; then
-  mvn test \
-      -P github-actions,codecoverage,integration-tests-set-1 || {
+run_integration_profile() {
+  local profile="$1"
+  shift
+
+  mvn test "$@" \
+      -P "github-actions,codecoverage,${profile}" || {
     print_surefire_dumps
     exit 1
   }
+}
+
+# Integration Tests
+cd pinot-integration-tests || exit 1
+if [ "$RUN_TEST_SET" == "1" ]; then
+  run_integration_profile integration-tests-set-1
+  shared_profiles=(
+    shared-hybrid-cluster-integration-test-suite
+    shared-llc-realtime-cluster-integration-test-suite
+  )
+  for shared_profile in "${shared_profiles[@]}"; do
+    run_integration_profile "${shared_profile}"
+  done
   exit 0
 fi
 if [ "$RUN_TEST_SET" == "2" ]; then
-  mvn test \
-      -DargLine="-Xms1g -Xmx2g -Dlog4j2.configurationFile=log4j2.xml" \
-      -P github-actions,codecoverage,integration-tests-set-2 || {
-    print_surefire_dumps
-    exit 1
-  }
+  run_integration_profile integration-tests-set-2 \
+      -DargLine="-Xms1g -Xmx2g -Dlog4j2.configurationFile=log4j2.xml"
+  # Keep this lane to the shared profiles that have shown a local wall-clock win.
+  shared_profiles=(
+    shared-no-override-offline-cluster-integration-test-suite
+    shared-realtime-manager-cluster-integration-test-suite
+    shared-controller-only-cluster-integration-test-suite
+    shared-offline-cluster-integration-test-suite
+  )
+  for shared_profile in "${shared_profiles[@]}"; do
+    run_integration_profile "${shared_profile}"
+  done
   exit 0
 fi
 

pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/AdminConsoleIntegrationTest.java

@@ -36,7 +36,7 @@
 /**
  * Tests that the controller, broker and server admin consoles return the expected pages.
  */
-public class AdminConsoleIntegrationTest extends BaseClusterIntegrationTest {
+public class AdminConsoleIntegrationTest extends SharedRichClusterIntegrationTest {
 
   @BeforeClass
   public void setUp()

pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTestSet.java

@@ -75,7 +75,7 @@
  * Shared set of common tests for cluster integration tests.
  * <p>To enable the test, override it and add @Test annotation.
  */
-public abstract class BaseClusterIntegrationTestSet extends BaseClusterIntegrationTest {
+public abstract class BaseClusterIntegrationTestSet extends SharedRichClusterIntegrationTest {
   private static final Logger LOGGER = LoggerFactory.getLogger(BaseClusterIntegrationTestSet.class);
 
   // Default settings

pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BaseDedupIntegrationTest.java

@@ -184,6 +184,9 @@ protected TableConfig createDedupTableWithReplicas(File sampleAvroFile, String p
   public void tearDown()
       throws IOException {
     dropRealtimeTable(getTableName());
+    dropRealtimeTable(DEDUP_TABLE_WITH_REPLICAS);
+    deleteSchema(getTableName());
+    deleteSchema(DEDUP_TABLE_WITH_REPLICAS);
     stopServer();
     stopBroker();
     stopController();

pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/BasicAuthBatchIntegrationTest.java

@@ -24,6 +24,7 @@
 import java.io.FileInputStream;
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.io.IOUtils;
@@ -33,6 +34,7 @@
 import org.apache.pinot.minion.executor.PinotTaskExecutor;
 import org.apache.pinot.spi.env.PinotConfiguration;
 import org.apache.pinot.spi.utils.JsonUtils;
+import org.apache.pinot.spi.utils.builder.TableNameBuilder;
 import org.apache.pinot.tools.BootstrapTableTool;
 import org.testng.Assert;
 import org.testng.annotations.AfterClass;
@@ -48,13 +50,16 @@
  * Integration test that provides example of {@link PinotTaskGenerator} and {@link PinotTaskExecutor} and tests simple
  * minion functionality.
  */
-public class BasicAuthBatchIntegrationTest extends ClusterTest {
+public class BasicAuthBatchIntegrationTest extends SharedRichClusterIntegrationTest {
   private static final String BOOTSTRAP_DATA_DIR = "/examples/batch/baseballStats";
+  private static final String TABLE_NAME = "baseballStats";
   private static final String SCHEMA_FILE = "baseballStats_schema.json";
   private static final String CONFIG_FILE = "baseballStats_offline_table_config.json";
   private static final String DATA_FILE = "baseballStats_data.csv";
   private static final String JOB_FILE = "ingestionJobSpec.yaml";
 
+  private File _quickstartTmpDir;
+
   @BeforeClass
   public void setUp()
       throws Exception {
@@ -65,16 +70,24 @@ public void setUp()
     startBroker();
     startServer();
     startMinion();
+
+    cleanUpTableAndSchema();
   }
 
   @AfterClass(alwaysRun = true)
   public void tearDown()
       throws Exception {
-    stopMinion();
-    stopServer();
-    stopBroker();
-    stopController();
-    stopZk();
+    Exception exception = null;
+    exception = runCleanup(exception, this::cleanUpTableAndSchema);
+    exception = runCleanup(exception, this::cleanUpQuickstartTmpDir);
+    exception = runCleanup(exception, this::stopMinionIfStarted);
+    exception = runCleanup(exception, this::stopServerIfStarted);
+    exception = runCleanup(exception, this::stopBrokerIfStarted);
+    exception = runCleanup(exception, this::stopControllerIfStarted);
+    exception = runCleanup(exception, this::stopZk);
+    if (exception != null) {
+      throw exception;
+    }
   }
 
   @Override
@@ -97,6 +110,31 @@ protected void overrideMinionConf(PinotConfiguration minionConf) {
     BasicAuthTestUtils.addMinionConfiguration(minionConf);
   }
 
+  @Override
+  protected Map<String, String> getControllerRequestClientHeaders() {
+    return AUTH_HEADER;
+  }
+
+  @Override
+  protected boolean shouldStartSharedKafka() {
+    return false;
+  }
+
+  @Override
+  protected int getSharedNumBrokers() {
+    return 1;
+  }
+
+  @Override
+  protected int getSharedNumServers() {
+    return 1;
+  }
+
+  @Override
+  protected boolean shouldStartSharedMinion() {
+    return true;
+  }
+
   @Test
   public void testBrokerNoAuth() {
     try {
@@ -139,51 +177,129 @@ public void testControllerGetTablesNoAuth() {
   @Test
   public void testIngestionBatch()
       throws Exception {
-    File quickstartTmpDir = new File(FileUtils.getTempDirectory(), String.valueOf(System.currentTimeMillis()));
-    FileUtils.forceDeleteOnExit(quickstartTmpDir);
+    cleanUpQuickstartTmpDir();
+    _quickstartTmpDir =
+        new File(FileUtils.getTempDirectory(), getClass().getSimpleName() + "-" + System.currentTimeMillis());
+    FileUtils.forceDeleteOnExit(_quickstartTmpDir);
 
-    File baseDir = new File(quickstartTmpDir, "baseballStats");
+    File baseDir = new File(_quickstartTmpDir, TABLE_NAME);
     File dataDir = new File(baseDir, "rawdata");
     File schemaFile = new File(baseDir, SCHEMA_FILE);
     File configFile = new File(baseDir, CONFIG_FILE);
     File dataFile = new File(dataDir, DATA_FILE);
     File jobFile = new File(baseDir, JOB_FILE);
-    Preconditions.checkState(dataDir.mkdirs());
+    try {
+      Preconditions.checkState(dataDir.mkdirs());
 
-    FileUtils.copyURLToFile(getClass().getResource(BOOTSTRAP_DATA_DIR + "/" + SCHEMA_FILE), schemaFile);
-    FileUtils.copyURLToFile(getClass().getResource(BOOTSTRAP_DATA_DIR + "/" + CONFIG_FILE), configFile);
-    FileUtils.copyURLToFile(getClass().getResource(BOOTSTRAP_DATA_DIR + "/rawdata/" + DATA_FILE), dataFile);
-    FileUtils.copyURLToFile(getClass().getResource(BOOTSTRAP_DATA_DIR + "/" + JOB_FILE), jobFile);
+      FileUtils.copyURLToFile(getClass().getResource(BOOTSTRAP_DATA_DIR + "/" + SCHEMA_FILE), schemaFile);
+      FileUtils.copyURLToFile(getClass().getResource(BOOTSTRAP_DATA_DIR + "/" + CONFIG_FILE), configFile);
+      FileUtils.copyURLToFile(getClass().getResource(BOOTSTRAP_DATA_DIR + "/rawdata/" + DATA_FILE), dataFile);
+      FileUtils.copyURLToFile(getClass().getResource(BOOTSTRAP_DATA_DIR + "/" + JOB_FILE), jobFile);
 
-    // patch ingestion job file
-    String jobFileContents = IOUtils.toString(new FileInputStream(jobFile));
-    IOUtils.write(jobFileContents.replaceAll("9000", String.valueOf(getControllerPort())),
-        new FileOutputStream(jobFile));
+      // patch ingestion job file
+      String jobFileContents;
+      try (FileInputStream inputStream = new FileInputStream(jobFile)) {
+        jobFileContents = IOUtils.toString(inputStream, StandardCharsets.UTF_8);
+      }
+      try (FileOutputStream outputStream = new FileOutputStream(jobFile)) {
+        IOUtils.write(jobFileContents.replaceAll("9000", String.valueOf(getControllerPort())), outputStream,
+            StandardCharsets.UTF_8);
+      }
 
-    new BootstrapTableTool("http", "localhost", getControllerPort(), baseDir.getAbsolutePath(),
-        AuthProviderUtils.makeAuthProvider(AUTH_TOKEN)).execute();
+      new BootstrapTableTool("http", "localhost", getControllerPort(), baseDir.getAbsolutePath(),
+          AuthProviderUtils.makeAuthProvider(AUTH_TOKEN)).execute();
 
-    Thread.sleep(5000);
+      Thread.sleep(5000);
 
-    // admin with full access
-    JsonNode response = JsonUtils.stringToJsonNode(
+      // admin with full access
+      JsonNode response = JsonUtils.stringToJsonNode(
+          sendPostRequest("http://localhost:" + getRandomBrokerPort() + "/query/sql",
+              "{\"sql\":\"SELECT count(*) FROM baseballStats\"}", AUTH_HEADER));
+      Assert.assertEquals(response.get("resultTable").get("dataSchema").get("columnDataTypes").get(0).asText(), "LONG",
+          "must return result with LONG value");
+      Assert.assertEquals(response.get("resultTable").get("dataSchema").get("columnNames").get(0).asText(), "count(*)",
+          "must return column name 'count(*)");
+      Assert.assertEquals(response.get("resultTable").get("rows").get(0).get(0).asInt(), 97889,
+          "must return row count 97889");
+      Assert.assertTrue(response.get("exceptions").isEmpty(), "must not return exception");
+
+      // user with valid auth but no table access - must return 403
+      try {
         sendPostRequest("http://localhost:" + getRandomBrokerPort() + "/query/sql",
-            "{\"sql\":\"SELECT count(*) FROM baseballStats\"}", AUTH_HEADER));
-    Assert.assertEquals(response.get("resultTable").get("dataSchema").get("columnDataTypes").get(0).asText(), "LONG",
-        "must return result with LONG value");
-    Assert.assertEquals(response.get("resultTable").get("dataSchema").get("columnNames").get(0).asText(), "count(*)",
-        "must return column name 'count(*)");
-    Assert.assertEquals(response.get("resultTable").get("rows").get(0).get(0).asInt(), 97889,
-        "must return row count 97889");
-    Assert.assertTrue(response.get("exceptions").isEmpty(), "must not return exception");
+            "{\"sql\":\"SELECT count(*) FROM baseballStats\"}", AUTH_HEADER_USER);
+      } catch (IOException e) {
+        HttpErrorStatusException httpErrorStatusException = (HttpErrorStatusException) e.getCause();
+        Assert.assertEquals(httpErrorStatusException.getStatusCode(), 403, "must return 403");
+      }
+    } finally {
+      cleanUpQuickstartTmpDir();
+    }
+  }
 
-    // user with valid auth but no table access - must return 403
+  private void cleanUpTableAndSchema()
+      throws Exception {
+    if (_helixResourceManager == null) {
+      return;
+    }
+
+    String offlineTableName = TableNameBuilder.OFFLINE.tableNameWithType(TABLE_NAME);
+    if (_helixResourceManager.getAllTables().contains(offlineTableName) || _helixResourceManager.hasOfflineTable(
+        TABLE_NAME)) {
+      dropOfflineTable(TABLE_NAME);
+      waitForTableDataManagerRemoved(offlineTableName);
+      waitForEVToDisappear(offlineTableName);
+    }
+    if (_helixResourceManager.getSchema(TABLE_NAME) != null) {
+      deleteSchema(TABLE_NAME);
+    }
+  }
+
+  private void cleanUpQuickstartTmpDir()
+      throws IOException {
+    if (_quickstartTmpDir != null) {
+      FileUtils.deleteDirectory(_quickstartTmpDir);
+      _quickstartTmpDir = null;
+    }
+  }
+
+  private void stopMinionIfStarted() {
+    if (_minionStarter != null) {
+      stopMinion();
+    }
+  }
+
+  private void stopServerIfStarted() {
+    if (!_serverStarters.isEmpty()) {
+      stopServer();
+    }
+  }
+
+  private void stopBrokerIfStarted() {
+    if (!_brokerStarters.isEmpty()) {
+      stopBroker();
+    }
+  }
+
+  private void stopControllerIfStarted() {
+    if (_controllerStarter != null) {
+      stopController();
+    }
+  }
+
+  private Exception runCleanup(Exception firstException, Cleanup cleanup) {
     try {
-      sendPostRequest("http://localhost:" + getRandomBrokerPort() + "/query/sql",
-          "{\"sql\":\"SELECT count(*) FROM baseballStats\"}", AUTH_HEADER_USER);
-    } catch (IOException e) {
-      HttpErrorStatusException httpErrorStatusException = (HttpErrorStatusException) e.getCause();
-      Assert.assertEquals(httpErrorStatusException.getStatusCode(), 403, "must return 403");
+      cleanup.run();
+    } catch (Exception e) {
+      if (firstException == null) {
+        return e;
+      }
+      firstException.addSuppressed(e);
     }
+    return firstException;
+  }
+
+  private interface Cleanup {
+    void run()
+        throws Exception;
   }
 }