diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index ededed192eb..909843307a5 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -126,7 +126,14 @@ jobs:
export DOCKER_BUILDKIT=1
export COMPOSE_DOCKER_CLI_BUILD=1
docker compose -f docker-compose.atlas-base.yml build
- docker compose -f docker-compose.atlas.yml -f docker-compose.atlas-hadoop.yml -f docker-compose.atlas-hive.yml up -d --wait
+ if ! docker compose -f docker-compose.atlas.yml -f docker-compose.atlas-hadoop.yml -f docker-compose.atlas-hive.yml up -d --wait --wait-timeout 300; then
+ echo "=== compose up --wait failed; dumping service logs ==="
+ for container in atlas-hbase atlas-hadoop atlas-zk atlas-kafka atlas-db atlas-hive atlas; do
+ echo "--- logs: ${container} ---"
+ docker logs --tail 200 "${container}" 2>&1 || true
+ done
+ exit 1
+ fi
- name: Check status of containers and remove them
run: |
diff --git a/addons/trino-extractor/README.md b/addons/trino-extractor/README.md
new file mode 100644
index 00000000000..0f617232028
--- /dev/null
+++ b/addons/trino-extractor/README.md
@@ -0,0 +1,185 @@
+# Trino Extractor — integration tests
+
+## Live integration test (`TrinoExtractorIT`)
+
+`TrinoExtractorIT` runs `./bin/run-trino-extractor.sh` from a **standalone tarball** (same `lib/` layout as distro), not the Maven test classpath. It skips automatically when no live stack is configured.
+
+**CI / default `mvn verify`:** integration tests are **skipped** (`skipITs=true`). Enable them with the `trino-extractor-it` profile below.
+
+**Run against Atlas + Trino lab** (e.g. ranger-docker Trino on `:8080`, Atlas on `:21000`):
+
+```bash
+export ATLAS_REST_URL=http://localhost:21000
+export TRINO_JDBC_URL=jdbc:trino://localhost:8080/
+export ATLAS_USERNAME=admin
+export ATLAS_PASSWORD=atlasR0cks!
+export TRINO_EXTRACTOR_SCHEMA=hr
+export TRINO_EXTRACTOR_TABLE=trino_pii_lab
+export TRINO_EXTRACTOR_COLUMN=ssn
+
+mvn -pl addons/trino-extractor -Ptrino-extractor-it verify
+```
+
+The `-Ptrino-extractor-it` profile builds the tarball in `pre-integration-test`, then runs `TrinoExtractorIT`. Optional: `TRINO_EXTRACTOR_TARBALL=/path/to/apache-atlas-*-trino-extractor.tar.gz` to reuse a pre-built distro artifact.
+
+### Test coverage map
+
+| Original testcase | Test class | Method | Default run |
+|-------------------|------------|--------|-------------|
+| Invalid arguments | `TrinoExtractorIT` | `testInvalidArguments` | Live (Atlas up) |
+| Invalid cron expression | `TrinoExtractorIT` | `testInvalidCronExpression` | Live |
+| Valid catalog run | `TrinoExtractorIT` | `testValidRegisteredCatalogRun` | Live |
+| Instance creation | `TrinoExtractorIT` | `testInstanceCreation` | Live |
+| Catalog creation | `TrinoExtractorIT` | `testCatalogCreation` | Live |
+| Schema creation | `TrinoExtractorIT` | `testSchemaCreation` | Live |
+| Table creation | `TrinoExtractorIT` | `testTableCreation` | Live |
+| Without cron expression | `TrinoExtractorIT` | `testWithoutCronExpression` | Live |
+| Unregistered catalog via CLI `-c` | `TrinoExtractorIT` | `testUnregisteredCatalogViaCommandLine` | Live |
+| Hook entity linked | `TrinoExtractorIT` | `testHookEntityLinkedToTrinoColumn` | Skip unless `TRINO_IT_HOOK_ENABLED=1` |
+| Tag propagated | `TrinoExtractorIT` | `testTagPropagatedToTrinoColumn` | Skip unless `TRINO_IT_TAG_PROPAGATION=1` |
+| Deleted table / schema / catalog | `TrinoExtractorIT` | `testDeleted*` | Skip unless `TRINO_IT_DELETE_SYNC=1` |
+| Rename catalog / schema (stale cleanup) | `TrinoExtractorIT` | `testRename*` | Skip unless `TRINO_IT_DELETE_SYNC=1` |
+| Deleted column | `TrinoExtractorIT` | `testDeletedColumnNotSupportedYet` | Skip (not implemented in extractor) |
+| Cron overlap guard | `TrinoExtractorTest` | `testMetadataJobDisallowConcurrentExecution` | Unit (`mvn test`) |
+| Cron validation | `TrinoExtractorTest` | `testInvalidCronExpressionRejected` | Unit |
+| Tarball Jersey classpath | `TrinoExtractorIT` | `testTarballJerseyClasspath` | Live |
+
+### Unit tests (no Atlas/Trino required)
+
+```bash
+mvn -pl addons/trino-extractor test -Dtest=TrinoExtractorTest
+```
+
+No live services are needed — only JDK and Maven.
+
+---
+
+## Components required to run tests
+
+What must be up depends on which tests you run.
+
+### Unit tests only
+
+| Required | Not required |
+|----------|----------------|
+| JDK + Maven | Atlas, Trino, Kafka, Hive, Ranger |
+
+```bash
+mvn -pl addons/trino-extractor test -Dtest=TrinoExtractorTest
+```
+
+### Default integration tests (`TrinoExtractorIT`)
+
+Set `ATLAS_REST_URL` and run:
+
+```bash
+mvn -pl addons/trino-extractor -Ptrino-extractor-it verify
+```
+
+#### Must be running
+
+| Component | Role | Default |
+|-----------|------|---------|
+| **Atlas server** | REST API — create/update/assert `trino_*` entities | `http://localhost:21000` |
+| **Atlas backend store** | Postgres or HBase (started with Atlas) | — |
+| **Trino coordinator** | JDBC metadata source for the extractor | `jdbc:trino://localhost:8080/` |
+| **Hive metastore** | Backing store for the Trino `hive` catalog | e.g. `ranger-hive` in docker lab |
+| **Trino `hive` catalog** | Catalog the extractor queries | `hive` |
+| **Schema / table / column** | Objects imported into Atlas | `hr.trino_pii_lab.ssn` |
+
+```text
+Trino Extractor IT --REST--> Atlas (:21000)
+ |
+ +--JDBC--> Trino (:8080) --> hive catalog --> Hive metastore
+```
+
+#### Minimum environment variables
+
+```bash
+export ATLAS_REST_URL=http://localhost:21000
+export TRINO_JDBC_URL=jdbc:trino://localhost:8080/
+export ATLAS_USERNAME=admin
+export ATLAS_PASSWORD=atlasR0cks!
+
+# Optional — defaults match ranger-docker lab
+export TRINO_EXTRACTOR_CATALOG=hive
+export TRINO_EXTRACTOR_SCHEMA=hr
+export TRINO_EXTRACTOR_TABLE=trino_pii_lab
+export TRINO_EXTRACTOR_COLUMN=ssn
+export ATLAS_TRINO_NAMESPACE=dev
+```
+
+#### Not required for default ITs
+
+| Component | Why not |
+|-----------|---------|
+| **Kafka** | Extractor uses Trino JDBC + Atlas REST only |
+| **Hive hook** | Default tests only create `trino_*` entities |
+| **Ranger** | Not used by extractor ITs |
+| **TagSync** | Only for optional tag propagation test |
+| **Pre-built distro tarball** | Maven profile `trino-extractor-it` builds the IT tarball |
+
+If `ATLAS_REST_URL` is unset or Atlas is unreachable, `TrinoExtractorIT` **skips** (suite passes without live tests).
+
+### Optional integration tests (extra flags)
+
+| Flag | Extra components | What it verifies |
+|------|------------------|------------------|
+| `TRINO_IT_HOOK_ENABLED=1` | HiveServer2 with **Atlas Hive hook**; `hive_column` in Atlas (default namespace `cm`) | `trino_column` links to `hive_column` |
+| `TRINO_IT_TAG_PROPAGATION=1` | Hook linkage + Atlas classification API | PII tag propagates to `trino_column` |
+| `TRINO_IT_DELETE_SYNC=1` | Atlas only (stale entities seeded via REST) | Stale `trino_table` / `trino_schema` / `trino_catalog` removal |
+
+TagSync and Ranger are **not** required even for tag propagation ITs (Atlas REST only).
+
+### Test tier matrix
+
+| Tier | Atlas | Trino | Hive MS + table | Hive hook | Kafka | Ranger | TagSync |
+|------|-------|-------|-----------------|-----------|-------|--------|---------|
+| Unit (`TrinoExtractorTest`) | — | — | — | — | — | — | — |
+| Default IT | Yes | Yes | Yes (`hr.trino_pii_lab`) | — | — | — | — |
+| Hook IT | Yes | Yes | Yes | Yes | —* | — | — |
+| Tag propagation IT | Yes | Yes | Yes | Yes | —* | — | — |
+| Delete sync IT | Yes | Yes** | — | — | — | — | — |
+
+\* Kafka only if your Hive hook setup uses it; not required by the extractor itself.
+\*\* Trino still needed for extractor JDBC; stale entities are seeded via Atlas REST.
+
+### Recommended docker lab (default ITs)
+
+Atlas coexist + ranger-docker stack:
+
+| Service | Port | Purpose |
+|---------|------|---------|
+| `atlas` | 21000 | Atlas REST + graph store |
+| Atlas Postgres (or HBase) | — | Atlas persistence |
+| `ranger-trino` | 8080 | Trino JDBC |
+| `ranger-hive` | — | Hive metastore for `hive` catalog |
+
+Create the test table if missing:
+
+```sql
+CREATE TABLE hr.trino_pii_lab (id INT, ssn STRING, address STRING);
+```
+
+Verify Trino sees it:
+
+```bash
+docker exec ranger-trino trino --user admin \
+ --execute "SELECT ssn FROM hive.hr.trino_pii_lab LIMIT 1"
+```
+
+### Pre-flight checks
+
+```bash
+# Atlas up
+curl -u admin:atlasR0cks! http://localhost:21000/api/atlas/v2/types/trino_column
+
+# Trino up
+docker exec ranger-trino trino --user admin --execute "SHOW SCHEMAS FROM hive"
+
+# Table exists
+docker exec ranger-trino trino --user admin \
+ --execute "DESCRIBE hive.hr.trino_pii_lab"
+```
+
+For full Atlas + Ranger + TagSync E2E (beyond these ITs), see `dev-support/atlas-docker/README-TRINO-ATLAS-RANGER-E2E.md`.
diff --git a/addons/trino-extractor/pom.xml b/addons/trino-extractor/pom.xml
index 9a15f3edf08..3ddce7de561 100644
--- a/addons/trino-extractor/pom.xml
+++ b/addons/trino-extractor/pom.xml
@@ -35,7 +35,7 @@
true
false
-
+
true
@@ -64,6 +64,18 @@
quartz
2.3.2
+
+
+ commons-io
+ commons-io
+ test
+
+
+
+ org.testng
+ testng
+ test
+
@@ -90,4 +102,106 @@
+
+
+
+ trino-extractor-it
+
+ false
+
+
+
+
+ org.apache.maven.plugins
+ maven-antrun-plugin
+
+
+ refresh-trino-extractor-lib
+
+ run
+
+ pre-integration-test
+
+
+
+
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-dependency-plugin
+
+
+ copy-dependencies-for-it-tarball
+
+ copy-dependencies
+
+ pre-integration-test
+
+ test
+ compile
+ ${project.build.directory}/dependency/trino
+ true
+ true
+ true
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-assembly-plugin
+
+
+ package-trino-extractor-tarball
+
+ single
+
+ pre-integration-test
+
+ false
+
+ src/test/assemblies/trino-extractor-tarball.xml
+
+ apache-atlas-${project.version}
+ ${project.build.directory}/trino-extractor-dist
+ false
+ gnu
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-failsafe-plugin
+
+
+ **/*IT.java
+
+
+
+
+ org.apache.maven.surefire
+ surefire-testng
+ ${failsafe.version}
+
+
+
+
+
+ integration-test
+ verify
+
+
+
+
+
+
+
+
diff --git a/addons/trino-extractor/src/test/assemblies/trino-extractor-tarball.xml b/addons/trino-extractor/src/test/assemblies/trino-extractor-tarball.xml
new file mode 100644
index 00000000000..03f4debb5e7
--- /dev/null
+++ b/addons/trino-extractor/src/test/assemblies/trino-extractor-tarball.xml
@@ -0,0 +1,59 @@
+
+
+
+
+ tar.gz
+
+ trino-extractor
+ apache-atlas-trino-extractor-${project.version}
+
+
+ src/main/conf
+ /conf
+
+ atlas-trino-extractor-logback.xml
+ atlas-trino-extractor.properties
+
+ 0755
+ 0755
+
+
+ src/main/bin
+ /bin
+
+ run-trino-extractor.sh
+
+ 0755
+ 0755
+
+
+ target/dependency/trino
+ /lib
+
+
+ target
+ /lib
+
+ atlas-trino-extractor-*.jar
+
+
+
+
diff --git a/addons/trino-extractor/src/test/java/org/apache/atlas/trino/cli/TrinoExtractorIT.java b/addons/trino-extractor/src/test/java/org/apache/atlas/trino/cli/TrinoExtractorIT.java
index e0e74ff9b80..b3474c300f4 100644
--- a/addons/trino-extractor/src/test/java/org/apache/atlas/trino/cli/TrinoExtractorIT.java
+++ b/addons/trino-extractor/src/test/java/org/apache/atlas/trino/cli/TrinoExtractorIT.java
@@ -17,24 +17,315 @@
*/
package org.apache.atlas.trino.cli;
+import org.testng.SkipException;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.io.IOException;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.Base64;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotEquals;
+import static org.testng.Assert.assertTrue;
+
+/**
+ * Live integration tests for the standalone Trino extractor tarball against Atlas + Trino.
+ *
+ * Enable with {@code mvn -pl addons/trino-extractor -Ptrino-extractor-it verify} and set
+ * {@code ATLAS_REST_URL}. Optional flags: {@code TRINO_IT_HOOK_ENABLED=1},
+ * {@code TRINO_IT_TAG_PROPAGATION=1}.
+ */
public class TrinoExtractorIT {
- /* List of testcases
- Invalid Arguments
- Invalid cron expression
- Test valid Catalog to be run
- Test Instance creation
- Test catalog creation
- Test schema creation
- Test table creation
- Test of hook is enabled, hook entity if created, is connected to Trino entity
- Test cron doesn't trigger new job, before earlier thread completes
- Test without cron expression
- Test even if catalog is not registered, it should run if passed from commandLine
- Deleted table
- Deleted catalog
- Deleted column
- Deleted schema
- Rename catalog
- Rename schema
- Tag propagated*/
+ private static TrinoExtractorITSupport.LiveStackConfig config;
+ private static Path workDir;
+
+ @BeforeClass
+ public static void setUpClass() throws Exception {
+ config = TrinoExtractorITSupport.loadLiveStackConfig();
+ workDir = TrinoExtractorITSupport.prepareExtractorWorkDir(config);
+ }
+
+ @AfterClass
+ public static void tearDownClass() throws Exception {
+ TrinoExtractorITSupport.deleteWorkDir(workDir);
+ workDir = null;
+ }
+
+ @BeforeMethod
+ public void resetExtractorProperties() throws Exception {
+ TrinoExtractorITSupport.rewriteProperties(workDir, config, null);
+ }
+
+ @Test(priority = 1)
+ public void testTarballJerseyClasspath() throws Exception {
+ TrinoExtractorITSupport.assertTarballJerseyClasspath(workDir);
+ }
+
+ @Test(priority = 2)
+ public void testInvalidArguments() throws Exception {
+ int exitCode = TrinoExtractorITSupport.runExtractorJava(workDir, config, "-c", config.catalog, "unexpected-arg");
+ assertNotEquals(exitCode, 0, "unrecognized Java argument should fail");
+ }
+
+ @Test(priority = 2)
+ public void testInvalidCronExpression() throws Exception {
+ long startMs = System.currentTimeMillis();
+ int exitCode = TrinoExtractorITSupport.runExtractorJava(workDir, config, "--cronExpression", "not-a-valid-cron", "-c", config.catalog, "-s", config.schema, "-t", config.table);
+ long elapsedSec = TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis() - startMs);
+
+ assertNotEquals(exitCode, 0, "invalid cron expression should fail");
+ assertTrue(elapsedSec < 120, "invalid cron should fail quickly, took " + elapsedSec + "s");
+ }
+
+ @Test(priority = 3)
+ public void testWithoutCronExpression() throws Exception {
+ assertEquals(runSingleTableExtract(), 0);
+ }
+
+ @Test(priority = 3)
+ public void testTableCreation() throws Exception {
+ assertEquals(runSingleTableExtract(), 0);
+
+ String columnQn = TrinoExtractorITSupport.trinoQualifiedName(config, "trino_column", config.catalog, config.schema, config.table, config.column);
+ TrinoExtractorITSupport.assertEntityExists(config, "trino_column", columnQn);
+ }
+
+ @Test(priority = 3, dependsOnMethods = "testTableCreation")
+ public void testSchemaCreation() throws Exception {
+ String schemaQn = TrinoExtractorITSupport.trinoQualifiedName(config, "trino_schema", config.catalog, config.schema);
+ TrinoExtractorITSupport.assertEntityExists(config, "trino_schema", schemaQn);
+ }
+
+ @Test(priority = 3, dependsOnMethods = "testTableCreation")
+ public void testCatalogCreation() throws Exception {
+ String catalogQn = TrinoExtractorITSupport.trinoQualifiedName(config, "trino_catalog", config.catalog);
+ TrinoExtractorITSupport.assertEntityExists(config, "trino_catalog", catalogQn);
+ }
+
+ @Test(priority = 3, dependsOnMethods = "testTableCreation")
+ public void testInstanceCreation() throws Exception {
+ TrinoExtractorITSupport.assertEntityExists(config, "trino_instance", config.trinoNamespace);
+ }
+
+ @Test(priority = 4)
+ public void testValidRegisteredCatalogRun() throws Exception {
+ Map overrides = new HashMap<>();
+ overrides.put("atlas.trino.catalogs.registered", config.catalog);
+ overrides.remove("atlas.trino.extractor.catalog");
+ overrides.remove("atlas.trino.extractor.schema");
+ overrides.remove("atlas.trino.extractor.table");
+
+ TrinoExtractorITSupport.rewriteProperties(workDir, config, overrides);
+
+ int exitCode = TrinoExtractorITSupport.runExtractorScript(workDir, config, "-c", config.catalog, "-s", config.schema, "-t", config.table);
+ assertEquals(exitCode, 0);
+ }
+
+ @Test(priority = 4)
+ public void testUnregisteredCatalogViaCommandLine() throws Exception {
+ Map overrides = new HashMap<>();
+ overrides.put("atlas.trino.catalogs.registered", "not_registered_in_trino");
+ TrinoExtractorITSupport.rewriteProperties(workDir, config, overrides);
+
+ int exitCode = TrinoExtractorITSupport.runExtractorScript(workDir, config, "-c", config.catalog, "-s", config.schema, "-t", config.table);
+ assertEquals(exitCode, 0, "CLI -c should work even when catalog is not in catalogs.registered");
+ }
+
+ @Test(priority = 5)
+ public void testDeletedTable() throws Exception {
+ if (!TrinoExtractorITSupport.isEnabled("TRINO_IT_DELETE_SYNC")) {
+ throw new SkipException("Set TRINO_IT_DELETE_SYNC=1 to verify stale trino_table removal (REST-seeded stale entities are not removed in current runs)");
+ }
+
+ assertEquals(runSingleTableExtract(), 0);
+
+ String staleTableName = "stale_trino_table_it_" + System.currentTimeMillis();
+ TrinoExtractorITSupport.seedStaleTrinoTable(config, staleTableName);
+
+ int exitCode = TrinoExtractorITSupport.runExtractorScript(workDir, config, "-c", config.catalog, "-s", config.schema);
+ assertEquals(exitCode, 0);
+
+ String staleTableQn = TrinoExtractorITSupport.trinoQualifiedName(config, "trino_table", config.catalog, config.schema, staleTableName);
+ TrinoExtractorITSupport.assertEntityAbsent(config, "trino_table", staleTableQn);
+ }
+
+ @Test(priority = 5)
+ public void testDeletedSchema() throws Exception {
+ if (!TrinoExtractorITSupport.isEnabled("TRINO_IT_DELETE_SYNC")) {
+ throw new SkipException("Set TRINO_IT_DELETE_SYNC=1 to verify stale trino_schema removal");
+ }
+
+ assertEquals(runSingleTableExtract(), 0);
+
+ String staleSchemaName = "stale_trino_schema_it_" + System.currentTimeMillis();
+ TrinoExtractorITSupport.seedStaleTrinoSchema(config, staleSchemaName);
+
+ int exitCode = TrinoExtractorITSupport.runExtractorScript(workDir, config, "-c", config.catalog);
+ assertEquals(exitCode, 0);
+
+ String staleSchemaQn = TrinoExtractorITSupport.trinoQualifiedName(config, "trino_schema", config.catalog, staleSchemaName);
+ TrinoExtractorITSupport.assertEntityAbsent(config, "trino_schema", staleSchemaQn);
+ }
+
+ @Test(priority = 5)
+ public void testDeletedCatalog() throws Exception {
+ if (!TrinoExtractorITSupport.isEnabled("TRINO_IT_DELETE_SYNC")) {
+ throw new SkipException("Set TRINO_IT_DELETE_SYNC=1 to verify stale trino_catalog removal");
+ }
+
+ assertEquals(runSingleTableExtract(), 0);
+
+ String staleCatalogName = "stale_trino_catalog_it_" + System.currentTimeMillis();
+ TrinoExtractorITSupport.seedStaleTrinoCatalog(config, staleCatalogName);
+
+ Map overrides = new HashMap<>();
+ overrides.put("atlas.trino.catalogs.registered", config.catalog);
+ TrinoExtractorITSupport.rewriteProperties(workDir, config, overrides);
+
+ int exitCode = TrinoExtractorITSupport.runExtractorScript(workDir, config);
+ assertEquals(exitCode, 0);
+
+ String staleCatalogQn = TrinoExtractorITSupport.trinoQualifiedName(config, "trino_catalog", staleCatalogName);
+ TrinoExtractorITSupport.assertEntityAbsent(config, "trino_catalog", staleCatalogQn);
+ }
+
+ @Test(priority = 6)
+ public void testRenameSchemaCreatesNewEntityAndDropsStale() throws Exception {
+ if (!TrinoExtractorITSupport.isEnabled("TRINO_IT_DELETE_SYNC")) {
+ throw new SkipException("Set TRINO_IT_DELETE_SYNC=1 to verify stale trino_schema cleanup after Trino rename");
+ }
+
+ assertEquals(runSingleTableExtract(), 0);
+
+ String staleSchemaName = "stale_trino_schema_it_" + System.currentTimeMillis();
+ TrinoExtractorITSupport.seedStaleTrinoSchema(config, staleSchemaName);
+
+ int exitCode = TrinoExtractorITSupport.runExtractorScript(workDir, config, "-c", config.catalog);
+ assertEquals(exitCode, 0);
+
+ String staleSchemaQn = TrinoExtractorITSupport.trinoQualifiedName(config, "trino_schema", config.catalog, staleSchemaName);
+ TrinoExtractorITSupport.assertEntityAbsent(config, "trino_schema", staleSchemaQn);
+
+ String liveSchemaQn = TrinoExtractorITSupport.trinoQualifiedName(config, "trino_schema", config.catalog, config.schema);
+ TrinoExtractorITSupport.assertEntityExists(config, "trino_schema", liveSchemaQn);
+ }
+
+ @Test(priority = 6)
+ public void testRenameCatalogCreatesNewEntityAndDropsStale() throws Exception {
+ if (!TrinoExtractorITSupport.isEnabled("TRINO_IT_DELETE_SYNC")) {
+ throw new SkipException("Set TRINO_IT_DELETE_SYNC=1 to verify stale trino_catalog cleanup after Trino rename");
+ }
+
+ assertEquals(runSingleTableExtract(), 0);
+
+ String staleCatalogName = "stale_trino_catalog_it_" + System.currentTimeMillis();
+ TrinoExtractorITSupport.seedStaleTrinoCatalog(config, staleCatalogName);
+
+ Map overrides = new HashMap<>();
+ overrides.put("atlas.trino.catalogs.registered", config.catalog);
+ TrinoExtractorITSupport.rewriteProperties(workDir, config, overrides);
+
+ int exitCode = TrinoExtractorITSupport.runExtractorScript(workDir, config);
+ assertEquals(exitCode, 0);
+
+ String staleCatalogQn = TrinoExtractorITSupport.trinoQualifiedName(config, "trino_catalog", staleCatalogName);
+ TrinoExtractorITSupport.assertEntityAbsent(config, "trino_catalog", staleCatalogQn);
+
+ String liveCatalogQn = TrinoExtractorITSupport.trinoQualifiedName(config, "trino_catalog", config.catalog);
+ TrinoExtractorITSupport.assertEntityExists(config, "trino_catalog", liveCatalogQn);
+ }
+
+ @Test(priority = 7)
+ public void testHookEntityLinkedToTrinoColumn() throws Exception {
+ if (!TrinoExtractorITSupport.isEnabled("TRINO_IT_HOOK_ENABLED")) {
+ throw new SkipException("Set TRINO_IT_HOOK_ENABLED=1 when Hive hook metadata exists in Atlas (namespace " + config.hiveNamespace + ")");
+ }
+
+ runHookEnabledExtract();
+
+ String columnQn = TrinoExtractorITSupport.trinoQualifiedName(config, "trino_column", config.catalog, config.schema, config.table, config.column);
+ String body = TrinoExtractorITSupport.getEntityBody(config, "trino_column", columnQn);
+
+ assertTrue(body != null && body.contains("hive_column"), "trino_column should link to hive_column when hook is enabled");
+ }
+
+ @Test(priority = 7)
+ public void testTagPropagatedToTrinoColumn() throws Exception {
+ if (!TrinoExtractorITSupport.isEnabled("TRINO_IT_TAG_PROPAGATION")) {
+ throw new SkipException("Set TRINO_IT_TAG_PROPAGATION=1 when hive_column PII tag propagation is configured");
+ }
+
+ runHookEnabledExtract();
+
+ String hiveColumnQn = config.schema + "." + config.table + "." + config.column + "@" + config.hiveNamespace;
+ String hiveGuid = TrinoExtractorITSupport.getEntityGuid(config, "hive_column", hiveColumnQn);
+
+ if (hiveGuid == null) {
+ throw new SkipException("hive_column not found for tag propagation test: " + hiveColumnQn);
+ }
+
+ classifyEntity(config, hiveGuid, "PII");
+
+ assertEquals(runSingleTableExtract(), 0);
+
+ String trinoColumnQn = TrinoExtractorITSupport.trinoQualifiedName(config, "trino_column", config.catalog, config.schema, config.table, config.column);
+ String body = TrinoExtractorITSupport.getEntityBody(config, "trino_column", trinoColumnQn);
+
+ assertTrue(body != null && body.contains("PII"), "PII tag should propagate to trino_column when hook link exists");
+ }
+
+ @Test(priority = 8)
+ public void testDeletedColumnNotSupportedYet() {
+ throw new SkipException("Extractor does not delete orphan trino_column entities when a column is dropped in Trino");
+ }
+
+ @Test(priority = 8)
+ public void testCronDoesNotOverlapConcurrentRuns() {
+ throw new SkipException("Concurrent cron scheduling is guarded by @DisallowConcurrentExecution; covered in TrinoExtractorTest");
+ }
+
+ private static int runSingleTableExtract() throws Exception {
+ return TrinoExtractorITSupport.runExtractorScript(workDir, config, "-c", config.catalog, "-s", config.schema, "-t", config.table);
+ }
+
+ private void runHookEnabledExtract() throws Exception {
+ Map overrides = new HashMap<>();
+ overrides.put("atlas.trino.catalog.hook.enabled." + config.catalog, "true");
+ overrides.put("atlas.trino.catalog.hook.enabled." + config.catalog + ".namespace", config.hiveNamespace);
+ TrinoExtractorITSupport.rewriteProperties(workDir, config, overrides);
+ assertEquals(runSingleTableExtract(), 0);
+ }
+
+ private static void classifyEntity(TrinoExtractorITSupport.LiveStackConfig cfg, String guid, String classification) throws IOException {
+ String payload = "[{\"typeName\":\"" + classification + "\"}]";
+ HttpURLConnection connection = openClassificationsPost(cfg, guid, payload);
+
+ try {
+ int status = connection.getResponseCode();
+ assertTrue(status >= 200 && status < 300, "classification POST failed: HTTP " + status);
+ } finally {
+ connection.disconnect();
+ }
+ }
+
+ private static HttpURLConnection openClassificationsPost(TrinoExtractorITSupport.LiveStackConfig cfg, String guid, String payload) throws IOException {
+ HttpURLConnection connection = (HttpURLConnection) new URL(cfg.atlasRestUrl + "api/atlas/v2/entity/guid/" + guid + "/classifications").openConnection();
+ connection.setRequestMethod("POST");
+ connection.setDoOutput(true);
+ connection.setRequestProperty("Content-Type", "application/json");
+ String credentials = cfg.atlasUsername + ":" + cfg.atlasPassword;
+ String encoded = Base64.getEncoder().encodeToString(credentials.getBytes(StandardCharsets.UTF_8));
+ connection.setRequestProperty("Authorization", "Basic " + encoded);
+ connection.getOutputStream().write(payload.getBytes(StandardCharsets.UTF_8));
+ return connection;
+ }
}
diff --git a/addons/trino-extractor/src/test/java/org/apache/atlas/trino/cli/TrinoExtractorITSupport.java b/addons/trino-extractor/src/test/java/org/apache/atlas/trino/cli/TrinoExtractorITSupport.java
new file mode 100644
index 00000000000..362ff46e7a0
--- /dev/null
+++ b/addons/trino-extractor/src/test/java/org/apache/atlas/trino/cli/TrinoExtractorITSupport.java
@@ -0,0 +1,598 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+package org.apache.atlas.trino.cli;
+
+import org.apache.commons.io.FileUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.SkipException;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.DirectoryStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.attribute.PosixFilePermission;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Properties;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+
+/**
+ * Shared helpers for live Trino extractor integration tests (tarball subprocess + Atlas REST).
+ */
+final class TrinoExtractorITSupport {
+ private static final Logger LOG = LoggerFactory.getLogger(TrinoExtractorITSupport.class);
+
+ static final String DEFAULT_TRINO_JDBC_URL = "jdbc:trino://localhost:8080/";
+ static final String DEFAULT_ATLAS_USER = "admin";
+ static final String DEFAULT_ATLAS_PASS = "atlasR0cks!";
+ static final String DEFAULT_NAMESPACE = "dev";
+ static final String DEFAULT_CATALOG = "hive";
+ static final String DEFAULT_SCHEMA = "hr";
+ static final String DEFAULT_TABLE = "trino_pii_lab";
+ static final String DEFAULT_COLUMN = "ssn";
+ static final String DEFAULT_HIVE_NAMESPACE = "cm";
+
+ private TrinoExtractorITSupport() {
+ }
+
+ static LiveStackConfig loadLiveStackConfig() {
+ String atlasRestUrl = firstNonBlank(System.getenv("ATLAS_REST_URL"), System.getenv("ATLAS_URL"));
+
+ if (atlasRestUrl == null) {
+ throw new SkipException("Set ATLAS_REST_URL (or ATLAS_URL) to run TrinoExtractorIT against a live Atlas");
+ }
+
+ LiveStackConfig config = new LiveStackConfig();
+ config.atlasRestUrl = normalizeAtlasUrl(atlasRestUrl);
+ config.atlasUsername = firstNonBlank(System.getenv("ATLAS_USERNAME"), System.getenv("ATLAS_USER"), DEFAULT_ATLAS_USER);
+ config.atlasPassword = firstNonBlank(System.getenv("ATLAS_PASSWORD"), System.getenv("ATLAS_PASS"), DEFAULT_ATLAS_PASS);
+ config.trinoJdbcUrl = firstNonBlank(System.getenv("TRINO_JDBC_URL"), DEFAULT_TRINO_JDBC_URL);
+ config.trinoNamespace = firstNonBlank(System.getenv("ATLAS_TRINO_NAMESPACE"), DEFAULT_NAMESPACE);
+ config.catalog = firstNonBlank(System.getenv("TRINO_EXTRACTOR_CATALOG"), DEFAULT_CATALOG);
+ config.schema = firstNonBlank(System.getenv("TRINO_EXTRACTOR_SCHEMA"), DEFAULT_SCHEMA);
+ config.table = firstNonBlank(System.getenv("TRINO_EXTRACTOR_TABLE"), DEFAULT_TABLE);
+ config.column = firstNonBlank(System.getenv("TRINO_EXTRACTOR_COLUMN"), DEFAULT_COLUMN);
+ config.hiveNamespace = firstNonBlank(System.getenv("ATLAS_HIVE_NAMESPACE"), DEFAULT_HIVE_NAMESPACE);
+
+ if (!isAtlasReachable(config)) {
+ throw new SkipException("Atlas is not reachable at " + config.atlasRestUrl);
+ }
+
+ return config;
+ }
+
+ static Path prepareExtractorWorkDir(LiveStackConfig config) throws Exception {
+ Path tarball = resolveTarball();
+ Path workDir = Files.createTempDirectory("atlas-trino-extractor-it-");
+ unpackTarball(tarball, workDir);
+ writeExtractorProperties(workDir, config, null);
+ makeExecutable(workDir.resolve("bin").resolve("run-trino-extractor.sh"));
+ LOG.info("Prepared extractor workdir={} tarball={}", workDir, tarball);
+ return workDir;
+ }
+
+ static void deleteWorkDir(Path workDir) throws IOException {
+ if (workDir != null) {
+ FileUtils.deleteDirectory(workDir.toFile());
+ }
+ }
+
+ static void assertTarballJerseyClasspath(Path workDir) throws IOException {
+ Path libDir = workDir.resolve("lib");
+ assertTrue(Files.isDirectory(libDir), "tarball lib/ directory missing");
+
+ List jerseyClientJars = new ArrayList<>();
+
+ try (DirectoryStream stream = Files.newDirectoryStream(libDir, "jersey-client*.jar")) {
+ for (Path jar : stream) {
+ jerseyClientJars.add(jar.getFileName().toString());
+ }
+ }
+
+ assertFalse(jerseyClientJars.isEmpty(), "jersey-client jar missing from tarball lib/");
+ assertTrue(jerseyClientJars.stream().anyMatch(name -> name.contains("jersey-client-1.19")),
+ "expected jersey-client 1.19 in tarball lib/, found: " + jerseyClientJars);
+ assertFalse(jerseyClientJars.stream().anyMatch(name -> name.matches("jersey-client-1\\.9\\.jar")),
+ "jersey-client 1.9 must not be in tarball lib/: " + jerseyClientJars);
+ }
+
+ static int runExtractorJava(Path workDir, LiveStackConfig config, String... javaArgs) throws IOException, InterruptedException {
+ Files.createDirectories(workDir.resolve("log"));
+
+ List command = new ArrayList<>();
+ command.add("java");
+ command.add("-Datlas.log.dir=" + workDir.resolve("log"));
+ command.add("-Datlas.log.file=atlas-trino-extractor.log");
+ command.add("-Dlogback.configurationFile=atlas-trino-extractor-logback.xml");
+ command.add("-Datlas.properties=atlas-trino-extractor.properties");
+ command.add("-cp");
+ command.add(buildClasspath(workDir));
+ command.add("org.apache.atlas.trino.cli.TrinoExtractor");
+ command.addAll(Arrays.asList(javaArgs));
+
+ ProcessBuilder processBuilder = new ProcessBuilder(command);
+ processBuilder.directory(workDir.toFile());
+ processBuilder.redirectErrorStream(true);
+ processBuilder.environment().put("ATLAS_USERNAME", config.atlasUsername);
+ processBuilder.environment().put("ATLAS_PASSWORD", config.atlasPassword);
+
+ Process process = processBuilder.start();
+
+ try (BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream(), StandardCharsets.UTF_8))) {
+ String output = reader.lines().collect(Collectors.joining(System.lineSeparator()));
+
+ if (!output.isEmpty()) {
+ LOG.debug("TrinoExtractor java output:\n{}", output);
+ }
+ }
+
+ return process.waitFor();
+ }
+
+ static int runExtractorScript(Path workDir, LiveStackConfig config, String... extraArgs) throws IOException, InterruptedException {
+ List command = new ArrayList<>();
+ command.add(workDir.resolve("bin").resolve("run-trino-extractor.sh").toString());
+ command.addAll(Arrays.asList(extraArgs));
+
+ ProcessBuilder processBuilder = new ProcessBuilder(command);
+ processBuilder.directory(workDir.toFile());
+ processBuilder.redirectErrorStream(true);
+ processBuilder.environment().put("ATLAS_USERNAME", config.atlasUsername);
+ processBuilder.environment().put("ATLAS_PASSWORD", config.atlasPassword);
+
+ Process process = processBuilder.start();
+ String output;
+
+ try (BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream(), StandardCharsets.UTF_8))) {
+ output = reader.lines().collect(Collectors.joining(System.lineSeparator()));
+ }
+
+ int exitCode = process.waitFor();
+
+ if (exitCode != 0) {
+ LOG.error("run-trino-extractor.sh {} exit={} output:\n{}", extraArgs, exitCode, output);
+ Path logFile = workDir.resolve("log").resolve("atlas-trino-extractor.log");
+
+ if (Files.isRegularFile(logFile)) {
+ LOG.error("extractor log tail:\n{}", tailFile(logFile, 8000));
+ }
+ } else {
+ LOG.info("run-trino-extractor.sh {} completed successfully", Arrays.toString(extraArgs));
+ }
+
+ return exitCode;
+ }
+
+ static void rewriteProperties(Path workDir, LiveStackConfig config, Map overrides) throws IOException {
+ writeExtractorProperties(workDir, config, overrides);
+ }
+
+ static String trinoQualifiedName(LiveStackConfig config, String type, String... parts) {
+ switch (type) {
+ case "trino_instance":
+ return config.trinoNamespace;
+ case "trino_catalog":
+ return String.format(Locale.ROOT, "%s@%s", parts[0], config.trinoNamespace);
+ case "trino_schema":
+ return String.format(Locale.ROOT, "%s.%s@%s", parts[0], parts[1], config.trinoNamespace);
+ case "trino_table":
+ return String.format(Locale.ROOT, "%s.%s.%s@%s", parts[0], parts[1], parts[2], config.trinoNamespace);
+ case "trino_column":
+ return String.format(Locale.ROOT, "%s.%s.%s.%s@%s", parts[0], parts[1], parts[2], parts[3], config.trinoNamespace);
+ default:
+ throw new IllegalArgumentException("unsupported type: " + type);
+ }
+ }
+
+ static void assertEntityExists(LiveStackConfig config, String typeName, String qualifiedName) throws IOException {
+ assertEquals(getEntityStatus(config, typeName, qualifiedName), 200,
+ typeName + " missing for qualifiedName=" + qualifiedName);
+ }
+
+ static void assertEntityAbsent(LiveStackConfig config, String typeName, String qualifiedName) throws IOException {
+ assertEquals(getEntityStatus(config, typeName, qualifiedName), 404,
+ typeName + " still present for qualifiedName=" + qualifiedName);
+ }
+
+ static String getEntityBody(LiveStackConfig config, String typeName, String qualifiedName) throws IOException {
+ String requestUrl = config.atlasRestUrl + "api/atlas/v2/entity/uniqueAttribute/type/" + typeName
+ + "?attr:qualifiedName=" + urlEncode(qualifiedName);
+ HttpURLConnection connection = openGet(requestUrl, config);
+
+ try {
+ int status = connection.getResponseCode();
+
+ if (status != 200) {
+ return null;
+ }
+
+ return readBody(connection);
+ } finally {
+ connection.disconnect();
+ }
+ }
+
+ static String getEntityGuid(LiveStackConfig config, String typeName, String qualifiedName) throws IOException {
+ String body = getEntityBody(config, typeName, qualifiedName);
+ return parseEntityGuid(body);
+ }
+
+ private static String parseEntityGuid(String body) {
+ if (body == null) {
+ return null;
+ }
+
+ int entityIdx = body.indexOf("\"entity\"");
+ int searchFrom = entityIdx >= 0 ? entityIdx : 0;
+ int guidIdx = body.indexOf("\"guid\":\"", searchFrom);
+
+ if (guidIdx < 0) {
+ return null;
+ }
+
+ int start = guidIdx + 8;
+ int end = body.indexOf('"', start);
+
+ return end > start ? body.substring(start, end) : null;
+ }
+
+ static void deleteEntityByGuid(LiveStackConfig config, String guid) throws IOException {
+ if (guid == null) {
+ return;
+ }
+
+ HttpURLConnection connection = openDelete(config.atlasRestUrl + "api/atlas/v2/entity/guid/" + guid, config);
+
+ try {
+ connection.getResponseCode();
+ } finally {
+ connection.disconnect();
+ }
+ }
+
+ static void seedStaleTrinoTable(LiveStackConfig config, String tableName) throws IOException {
+ String schemaQn = trinoQualifiedName(config, "trino_schema", config.catalog, config.schema);
+ String tableQn = trinoQualifiedName(config, "trino_table", config.catalog, config.schema, tableName);
+
+ deleteIfExists(config, "trino_table", tableQn);
+
+ String schemaGuid = getEntityGuid(config, "trino_schema", schemaQn);
+ assertTrue(schemaGuid != null, "schema must exist before stale trino_table test: " + schemaQn);
+
+ String payload = "{\"entity\":{\"typeName\":\"trino_table\",\"attributes\":{\"qualifiedName\":\"" + tableQn + "\",\"name\":\"" + tableName + "\"},\"relationshipAttributes\":{\"trinoschema\":{\"guid\":\"" + schemaGuid + "\",\"typeName\":\"trino_schema\"}}}}";
+
+ postJson(config, "api/atlas/v2/entity", payload);
+ assertEntityExists(config, "trino_table", tableQn);
+ }
+
+ static void seedStaleTrinoSchema(LiveStackConfig config, String schemaName) throws IOException {
+ String catalogQn = trinoQualifiedName(config, "trino_catalog", config.catalog);
+ String schemaQn = trinoQualifiedName(config, "trino_schema", config.catalog, schemaName);
+
+ deleteIfExists(config, "trino_schema", schemaQn);
+
+ String catalogGuid = getEntityGuid(config, "trino_catalog", catalogQn);
+ assertTrue(catalogGuid != null, "catalog must exist before stale trino_schema test: " + catalogQn);
+
+ String payload = "{\"entity\":{\"typeName\":\"trino_schema\",\"attributes\":{\"qualifiedName\":\"" + schemaQn + "\",\"name\":\"" + schemaName + "\"},\"relationshipAttributes\":{\"catalog\":{\"guid\":\"" + catalogGuid + "\",\"typeName\":\"trino_catalog\"}}}}";
+
+ postJson(config, "api/atlas/v2/entity", payload);
+ assertEntityExists(config, "trino_schema", schemaQn);
+ }
+
+ static void seedStaleTrinoCatalog(LiveStackConfig config, String catalogName) throws IOException {
+ String catalogQn = trinoQualifiedName(config, "trino_catalog", catalogName);
+
+ deleteIfExists(config, "trino_catalog", catalogQn);
+
+ String instanceGuid = getEntityGuid(config, "trino_instance", config.trinoNamespace);
+ assertTrue(instanceGuid != null, "trino_instance must exist before stale trino_catalog test");
+
+ String payload = "{\"entity\":{\"typeName\":\"trino_catalog\",\"attributes\":{\"qualifiedName\":\"" + catalogQn + "\",\"name\":\"" + catalogName + "\",\"connectorType\":\"hive\"},\"relationshipAttributes\":{\"instance\":{\"guid\":\"" + instanceGuid + "\",\"typeName\":\"trino_instance\"}}}}";
+
+ postJson(config, "api/atlas/v2/entity", payload);
+ assertEntityExists(config, "trino_catalog", catalogQn);
+ }
+
+ static boolean isEnabled(String envName) {
+ return "1".equals(System.getenv(envName)) || "true".equalsIgnoreCase(System.getenv(envName));
+ }
+
+ private static void deleteIfExists(LiveStackConfig config, String typeName, String qualifiedName) throws IOException {
+ if (getEntityStatus(config, typeName, qualifiedName) == 200) {
+ deleteEntityByGuid(config, getEntityGuid(config, typeName, qualifiedName));
+ }
+ }
+
+ private static String buildClasspath(Path workDir) throws IOException {
+ List entries = new ArrayList<>();
+ entries.add(workDir.resolve("conf").toString());
+
+ try (DirectoryStream stream = Files.newDirectoryStream(workDir.resolve("lib"), "*.jar")) {
+ for (Path jar : stream) {
+ entries.add(jar.toString());
+ }
+ }
+
+ return String.join(":", entries);
+ }
+
+ private static int getEntityStatus(LiveStackConfig config, String typeName, String qualifiedName) throws IOException {
+ String requestUrl = config.atlasRestUrl + "api/atlas/v2/entity/uniqueAttribute/type/" + typeName
+ + "?attr:qualifiedName=" + urlEncode(qualifiedName);
+ HttpURLConnection connection = openGet(requestUrl, config);
+
+ try {
+ return connection.getResponseCode();
+ } finally {
+ connection.disconnect();
+ }
+ }
+
+ private static void postJson(LiveStackConfig config, String path, String json) throws IOException {
+ HttpURLConnection connection = openPost(config.atlasRestUrl + path, config);
+
+ try {
+ connection.getOutputStream().write(json.getBytes(StandardCharsets.UTF_8));
+ int status = connection.getResponseCode();
+ String body = readResponseBody(connection, status);
+
+ if (status < 200 || status >= 300) {
+ throw new IOException("POST " + path + " failed: HTTP " + status + " body=" + body);
+ }
+ } finally {
+ connection.disconnect();
+ }
+ }
+
+ private static String readResponseBody(HttpURLConnection connection, int status) throws IOException {
+ java.io.InputStream stream = status >= 400 ? connection.getErrorStream() : connection.getInputStream();
+
+ if (stream == null) {
+ return "";
+ }
+
+ try (BufferedReader reader = new BufferedReader(new InputStreamReader(stream, StandardCharsets.UTF_8))) {
+ return reader.lines().collect(Collectors.joining());
+ }
+ }
+
+ private static boolean isAtlasReachable(LiveStackConfig config) {
+ try {
+ HttpURLConnection connection = openGet(config.atlasRestUrl + "api/atlas/admin/version", config);
+
+ try {
+ return connection.getResponseCode() == 200;
+ } finally {
+ connection.disconnect();
+ }
+ } catch (IOException e) {
+ LOG.warn("Atlas reachability check failed: {}", e.toString());
+ return false;
+ }
+ }
+
+ private static HttpURLConnection openGet(String requestUrl, LiveStackConfig config) throws IOException {
+ HttpURLConnection connection = (HttpURLConnection) new URL(requestUrl).openConnection();
+ connection.setRequestMethod("GET");
+ connection.setConnectTimeout(10_000);
+ connection.setReadTimeout(30_000);
+ addBasicAuth(connection, config);
+ return connection;
+ }
+
+ private static HttpURLConnection openPost(String requestUrl, LiveStackConfig config) throws IOException {
+ HttpURLConnection connection = (HttpURLConnection) new URL(requestUrl).openConnection();
+ connection.setRequestMethod("POST");
+ connection.setDoOutput(true);
+ connection.setConnectTimeout(10_000);
+ connection.setReadTimeout(30_000);
+ connection.setRequestProperty("Content-Type", "application/json");
+ addBasicAuth(connection, config);
+ return connection;
+ }
+
+ private static HttpURLConnection openDelete(String requestUrl, LiveStackConfig config) throws IOException {
+ HttpURLConnection connection = (HttpURLConnection) new URL(requestUrl).openConnection();
+ connection.setRequestMethod("DELETE");
+ connection.setConnectTimeout(10_000);
+ connection.setReadTimeout(30_000);
+ addBasicAuth(connection, config);
+ return connection;
+ }
+
+ private static void addBasicAuth(HttpURLConnection connection, LiveStackConfig config) {
+ String credentials = config.atlasUsername + ":" + config.atlasPassword;
+ String encoded = Base64.getEncoder().encodeToString(credentials.getBytes(StandardCharsets.UTF_8));
+ connection.setRequestProperty("Authorization", "Basic " + encoded);
+ }
+
+ private static String readBody(HttpURLConnection connection) throws IOException {
+ try (BufferedReader reader = new BufferedReader(new InputStreamReader(connection.getInputStream(), StandardCharsets.UTF_8))) {
+ return reader.lines().collect(Collectors.joining());
+ }
+ }
+
+ private static Path resolveTarball() throws IOException {
+ String explicitTarball = System.getenv("TRINO_EXTRACTOR_TARBALL");
+
+ if (explicitTarball != null && !explicitTarball.isEmpty()) {
+ Path path = Paths.get(explicitTarball);
+
+ if (!Files.isRegularFile(path)) {
+ throw new SkipException("TRINO_EXTRACTOR_TARBALL does not exist: " + explicitTarball);
+ }
+
+ return path;
+ }
+
+ Path moduleDir = Paths.get(System.getProperty("user.dir")).toAbsolutePath().normalize();
+ List candidates = Arrays.asList(
+ moduleDir.resolve("target/trino-extractor-dist"),
+ moduleDir.resolve("../../distro/target").normalize());
+
+ for (Path dir : candidates) {
+ Path match = findTarballInDirectory(dir);
+
+ if (match != null) {
+ return match;
+ }
+ }
+
+ throw new SkipException("Trino extractor tarball not found. Build with "
+ + "mvn -pl addons/trino-extractor -Ptrino-extractor-it package "
+ + "or set TRINO_EXTRACTOR_TARBALL");
+ }
+
+ private static Path findTarballInDirectory(Path directory) throws IOException {
+ if (!Files.isDirectory(directory)) {
+ return null;
+ }
+
+ try (Stream paths = Files.list(directory)) {
+ return paths
+ .filter(Files::isRegularFile)
+ .filter(path -> path.getFileName().toString().contains("trino-extractor"))
+ .filter(path -> path.getFileName().toString().endsWith(".tar.gz"))
+ .sorted()
+ .findFirst()
+ .orElse(null);
+ }
+ }
+
+ private static void unpackTarball(Path tarball, Path destination) throws IOException, InterruptedException {
+ Files.createDirectories(destination);
+
+ ProcessBuilder processBuilder = new ProcessBuilder(
+ "tar", "xzf", tarball.toAbsolutePath().toString(),
+ "-C", destination.toAbsolutePath().toString(),
+ "--strip-components=1");
+ processBuilder.redirectErrorStream(true);
+
+ Process process = processBuilder.start();
+ process.waitFor();
+ assertEquals(process.exitValue(), 0, "failed to unpack tarball " + tarball);
+ }
+
+ private static void writeExtractorProperties(Path workDir, LiveStackConfig config, Map overrides) throws IOException {
+ Properties properties = new Properties();
+ properties.setProperty("atlas.rest.address", config.atlasRestUrl);
+ properties.setProperty("atlas.trino.jdbc.address", config.trinoJdbcUrl);
+ properties.setProperty("atlas.trino.jdbc.user", firstNonBlank(System.getenv("TRINO_JDBC_USER"), "admin"));
+ properties.setProperty("atlas.trino.namespace", config.trinoNamespace);
+ properties.setProperty("atlas.trino.catalogs.registered", config.catalog);
+
+ if (overrides != null) {
+ for (Map.Entry entry : overrides.entrySet()) {
+ if (entry.getValue() == null) {
+ properties.remove(entry.getKey());
+ } else {
+ properties.setProperty(entry.getKey(), entry.getValue());
+ }
+ }
+ }
+
+ Path confDir = workDir.resolve("conf");
+ Files.createDirectories(confDir);
+
+ try (OutputStream out = Files.newOutputStream(confDir.resolve("atlas-trino-extractor.properties"))) {
+ properties.store(out, "TrinoExtractorIT");
+ }
+ }
+
+ private static void makeExecutable(Path script) throws IOException {
+ if (!Files.exists(script)) {
+ return;
+ }
+
+ try {
+ Files.setPosixFilePermissions(script, EnumSet.of(
+ PosixFilePermission.OWNER_READ,
+ PosixFilePermission.OWNER_WRITE,
+ PosixFilePermission.OWNER_EXECUTE,
+ PosixFilePermission.GROUP_READ,
+ PosixFilePermission.GROUP_EXECUTE,
+ PosixFilePermission.OTHERS_READ,
+ PosixFilePermission.OTHERS_EXECUTE));
+ } catch (UnsupportedOperationException e) {
+ LOG.debug("POSIX permissions not supported for {}", script);
+ }
+ }
+
+ private static String tailFile(Path file, int maxChars) throws IOException {
+ String content = new String(Files.readAllBytes(file), StandardCharsets.UTF_8);
+ return content.length() <= maxChars ? content : content.substring(content.length() - maxChars);
+ }
+
+ private static String normalizeAtlasUrl(String url) {
+ return url.endsWith("/") ? url : url + "/";
+ }
+
+ private static String urlEncode(String value) {
+ try {
+ return java.net.URLEncoder.encode(value, "UTF-8");
+ } catch (java.io.UnsupportedEncodingException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private static String firstNonBlank(String... values) {
+ if (values == null) {
+ return null;
+ }
+
+ for (String value : values) {
+ if (value != null && !value.trim().isEmpty()) {
+ return value.trim();
+ }
+ }
+
+ return null;
+ }
+
+ static final class LiveStackConfig {
+ String atlasRestUrl;
+ String atlasUsername;
+ String atlasPassword;
+ String trinoJdbcUrl;
+ String trinoNamespace;
+ String catalog;
+ String schema;
+ String table;
+ String column;
+ String hiveNamespace;
+ }
+}
diff --git a/addons/trino-extractor/src/test/java/org/apache/atlas/trino/cli/TrinoExtractorTest.java b/addons/trino-extractor/src/test/java/org/apache/atlas/trino/cli/TrinoExtractorTest.java
new file mode 100644
index 00000000000..d85e1038768
--- /dev/null
+++ b/addons/trino-extractor/src/test/java/org/apache/atlas/trino/cli/TrinoExtractorTest.java
@@ -0,0 +1,47 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+package org.apache.atlas.trino.cli;
+
+import org.quartz.CronExpression;
+import org.quartz.DisallowConcurrentExecution;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+
+/**
+ * Unit tests for Trino extractor CLI behaviour that do not require a live Atlas/Trino stack.
+ */
+public class TrinoExtractorTest {
+ @Test
+ public void testInvalidCronExpressionRejected() {
+ assertFalse(CronExpression.isValidExpression("not-a-valid-cron"));
+ }
+
+ @Test
+ public void testValidCronExpressionAccepted() {
+ assertTrue(CronExpression.isValidExpression("0 0 2 * * ?"));
+ }
+
+ @Test
+ public void testMetadataJobDisallowConcurrentExecution() throws Exception {
+ DisallowConcurrentExecution annotation = TrinoExtractor.MetadataJob.class.getAnnotation(DisallowConcurrentExecution.class);
+ assertNotNull(annotation, "MetadataJob must disallow overlapping cron executions");
+ }
+}
diff --git a/dev-support/atlas-docker/Dockerfile.atlas-base b/dev-support/atlas-docker/Dockerfile.atlas-base
index abc2b68d39d..a7af4284450 100644
--- a/dev-support/atlas-docker/Dockerfile.atlas-base
+++ b/dev-support/atlas-docker/Dockerfile.atlas-base
@@ -22,7 +22,7 @@ ARG ATLAS_BASE_JAVA_VERSION
# Install tzdata, Python, Java
RUN apt-get update && \
- DEBIAN_FRONTEND="noninteractive" apt-get -y install tzdata vim\
+ DEBIAN_FRONTEND="noninteractive" apt-get -y install tzdata vim wget curl \
python3 python3-pip openjdk-8-jdk openjdk-11-jdk openjdk-17-jdk bc iputils-ping ssh pdsh
# Set environment variables
diff --git a/dev-support/atlas-docker/Dockerfile.atlas-hbase b/dev-support/atlas-docker/Dockerfile.atlas-hbase
index a05bda3a197..c4e7e791079 100644
--- a/dev-support/atlas-docker/Dockerfile.atlas-hbase
+++ b/dev-support/atlas-docker/Dockerfile.atlas-hbase
@@ -24,6 +24,7 @@ COPY ./dist/apache-atlas-${ATLAS_VERSION}-hbase-hook.tar.gz /home/atlas/dist/
COPY ./downloads/hbase-${HBASE_VERSION}-bin.tar.gz /home/atlas/dist/
COPY ./scripts/atlas-hbase-setup.sh /home/atlas/scripts/
+COPY ./scripts/atlas-hbase-healthcheck.sh /home/atlas/scripts/
COPY ./scripts/atlas-hbase.sh /home/atlas/scripts/
COPY ./scripts/hbase-site.xml /home/atlas/scripts/
COPY ./scripts/atlas-hbase-application.properties /home/atlas/scripts/
@@ -38,7 +39,8 @@ RUN tar xvfz /home/atlas/dist/hbase-${HBASE_VERSION}-bin.tar.gz --directory=/opt
ln -s /opt/apache-atlas-hbase-hook/hook/hbase/hbase-bridge-shim-${ATLAS_VERSION}.jar /opt/hbase/lib/ && \
ln -s /opt/apache-atlas-hbase-hook/hook/hbase/atlas-hbase-plugin-impl /opt/hbase/lib/atlas-hbase-plugin-impl && \
cp /home/atlas/scripts/hbase-site.xml /opt/hbase/conf/hbase-site.xml && \
- cp -f /home/atlas/scripts/atlas-hbase-application.properties /opt/hbase/conf/atlas-application.properties
+ cp -f /home/atlas/scripts/atlas-hbase-application.properties /opt/hbase/conf/atlas-application.properties && \
+ chmod +x /home/atlas/scripts/atlas-hbase-healthcheck.sh /home/atlas/scripts/atlas-hbase.sh
ENV HBASE_HOME=/opt/hbase
ENV PATH=/usr/java/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/opt/hbase/bin
diff --git a/dev-support/atlas-docker/docker-compose.atlas-hbase.yml b/dev-support/atlas-docker/docker-compose.atlas-hbase.yml
index 9d46702c289..82bdf10dcef 100644
--- a/dev-support/atlas-docker/docker-compose.atlas-hbase.yml
+++ b/dev-support/atlas-docker/docker-compose.atlas-hbase.yml
@@ -27,11 +27,11 @@ services:
condition: service_started
restart: unless-stopped
healthcheck:
- test: [ "CMD-SHELL", "wget -q --spider http://localhost:16030/rs-status && wget -q --spider http://localhost:16010/master-status" ]
- interval: 30s
- timeout: 10s
- retries: 30
- start_period: 40s
+ test: [ "CMD-SHELL", "/home/atlas/scripts/atlas-hbase-healthcheck.sh" ]
+ interval: 10s
+ timeout: 15s
+ retries: 24
+ start_period: 120s
environment:
- HBASE_VERSION
- ATLAS_VERSION
diff --git a/dev-support/atlas-docker/scripts/atlas-hbase-healthcheck.sh b/dev-support/atlas-docker/scripts/atlas-hbase-healthcheck.sh
new file mode 100644
index 00000000000..ddab3059563
--- /dev/null
+++ b/dev-support/atlas-docker/scripts/atlas-hbase-healthcheck.sh
@@ -0,0 +1,33 @@
+#!/bin/bash
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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.
+
+# Docker healthcheck: HBase Master (16010) and RegionServer (16030) info pages.
+check_url() {
+ local url="$1"
+ if command -v wget >/dev/null 2>&1; then
+ wget -q --spider "${url}"
+ return $?
+ fi
+ if command -v curl >/dev/null 2>&1; then
+ curl -sf -o /dev/null "${url}"
+ return $?
+ fi
+ return 1
+}
+
+check_url "http://localhost:16010/master-status" \
+ && check_url "http://localhost:16030/rs-status"
diff --git a/dev-support/atlas-docker/scripts/atlas-hbase.sh b/dev-support/atlas-docker/scripts/atlas-hbase.sh
index e139a937756..3c986c80c28 100755
--- a/dev-support/atlas-docker/scripts/atlas-hbase.sh
+++ b/dev-support/atlas-docker/scripts/atlas-hbase.sh
@@ -33,21 +33,34 @@ fi
su -c "${HBASE_HOME}/bin/start-hbase.sh" hbase
-HBASE_MASTER_PID=""
-for attempt in $(seq 1 60); do
- HBASE_MASTER_PID=`ps -ef | grep -v grep | grep -i "org.apache.hadoop.hbase.master.HMaster" | awk '{print $2}'`
-
- if [ -n "${HBASE_MASTER_PID}" ]; then
+echo "Waiting for HBase Master and RegionServer (up to 180s)..."
+READY=false
+for attempt in $(seq 1 90); do
+ if ${ATLAS_SCRIPTS}/atlas-hbase-healthcheck.sh 2>/dev/null; then
+ echo "HBase master and regionserver ready (~$((attempt * 2))s)"
+ READY=true
break
fi
-
sleep 2
done
-if [ -z "${HBASE_MASTER_PID}" ]; then
- echo "HBase HMaster failed to start" >&2
+if [ "${READY}" != "true" ]; then
+ echo "HBase health endpoints not ready within 180s" >&2
+ ls -la ${HBASE_HOME}/logs/ 2>/dev/null || true
+ tail -80 ${HBASE_HOME}/logs/* 2>/dev/null || true
exit 1
fi
-# prevent the container from exiting
-tail --pid=$HBASE_MASTER_PID -f /dev/null
+# Keep container alive while HBase JVMs are running (do not tail only HMaster PID).
+while true; do
+ MASTER_PID=$(ps -ef | grep -v grep | grep -i "org.apache.hadoop.hbase.master.HMaster" | awk '{print $2}')
+ RS_PID=$(ps -ef | grep -v grep | grep -i "org.apache.hadoop.hbase.regionserver.HRegionServer" | awk '{print $2}')
+
+ if [ -z "${MASTER_PID}" ] && [ -z "${RS_PID}" ]; then
+ echo "HBase master and regionserver processes exited" >&2
+ tail -80 ${HBASE_HOME}/logs/* 2>/dev/null || true
+ exit 1
+ fi
+
+ sleep 30
+done
diff --git a/webapp/src/test/java/org/apache/atlas/notification/ImportTaskListenerImplTest.java b/webapp/src/test/java/org/apache/atlas/notification/ImportTaskListenerImplTest.java
index c0fdfddf574..422f7d4cf5b 100644
--- a/webapp/src/test/java/org/apache/atlas/notification/ImportTaskListenerImplTest.java
+++ b/webapp/src/test/java/org/apache/atlas/notification/ImportTaskListenerImplTest.java
@@ -91,10 +91,7 @@ public class ImportTaskListenerImplTest {
public void setup() throws Exception {
MockitoAnnotations.openMocks(this);
- importRequest = mock(AtlasAsyncImportRequest.class);
-
- when(importRequest.getImportId()).thenReturn("import123");
- when(importRequest.getTopicName()).thenReturn("topic1");
+ importRequest = createImportRequestMock("import123", "topic1");
requestQueue = mock(BlockingDeque.class);
asyncImportService = mock(AsyncImportService.class);
@@ -109,8 +106,7 @@ public void setup() throws Exception {
public void resetMocks() throws AtlasException {
MockitoAnnotations.openMocks(this);
- when(importRequest.getImportId()).thenReturn("import123");
- when(importRequest.getTopicName()).thenReturn("topic1");
+ importRequest = createImportRequestMock("import123", "topic1");
when(asyncImportService.fetchImportRequestByImportId(any(String.class))).thenReturn(importRequest);
importTaskListener = new ImportTaskListenerImpl(asyncImportService, notificationHookConsumer, requestQueue);
@@ -119,7 +115,7 @@ public void resetMocks() throws AtlasException {
@AfterMethod
public void teardown() throws Exception {
shutdownImportExecutor(importTaskListener);
- Mockito.reset(asyncImportService, notificationHookConsumer, requestQueue, importRequest);
+ Mockito.reset(asyncImportService, notificationHookConsumer, requestQueue);
}
@Test
@@ -293,10 +289,11 @@ public void testStartAsyncImportIfAvailable_WithInvalidStatus() throws Exception
@Test
public void testStartImportConsumer_Successful() throws Exception {
- Mockito.doReturn("import123").when(importRequest).getImportId();
- when(importRequest.getStatus()).thenReturn(WAITING);
- when(importRequest.getTopicName()).thenReturn("topic1");
- when(asyncImportService.fetchImportRequestByImportId("import123")).thenReturn(importRequest);
+ AtlasAsyncImportRequest request = createImportRequestMock("import123", "topic1");
+
+ when(request.getStatus()).thenReturn(WAITING);
+ when(asyncImportService.fetchImportRequestByImportId("import123")).thenReturn(request);
+ when(requestQueue.poll(anyLong(), any(TimeUnit.class))).thenReturn("import123");
CountDownLatch consumerStarted = new CountDownLatch(1);
@@ -305,14 +302,9 @@ public void testStartImportConsumer_Successful() throws Exception {
return null;
}).when(notificationHookConsumer).startAsyncImportConsumer(any(), anyString(), anyString());
- ExecutorService realExecutor = java.util.concurrent.Executors.newSingleThreadExecutor();
- Field executorField = ImportTaskListenerImpl.class.getDeclaredField("executorService");
+ setExecutorService(importTaskListener, synchronousExecutor());
- executorField.setAccessible(true);
- executorField.set(importTaskListener, realExecutor);
- when(requestQueue.poll(anyLong(), any(TimeUnit.class))).thenReturn("import123");
-
- importTaskListener.onReceiveImportRequest(importRequest);
+ importTaskListener.onReceiveImportRequest(request);
assertTrue(consumerStarted.await(5, TimeUnit.SECONDS), "startAsyncImportConsumer was not invoked");
@@ -321,35 +313,29 @@ public void testStartImportConsumer_Successful() throws Exception {
@Test
public void testStartImportConsumer_Failure() throws Exception {
- when(importRequest.getImportId()).thenReturn("import123");
- when(importRequest.getStatus()).thenReturn(WAITING);
- when(importRequest.getTopicName()).thenReturn("topic1");
- when(asyncImportService.fetchImportRequestByImportId("import123")).thenReturn(importRequest);
+ AtlasAsyncImportRequest request = createImportRequestMock("import123", "topic1");
+
+ when(request.getStatus()).thenReturn(WAITING);
+ when(asyncImportService.fetchImportRequestByImportId("import123")).thenReturn(request);
+ when(requestQueue.poll(anyLong(), any(TimeUnit.class))).thenReturn("import123").thenReturn(null);
CountDownLatch consumerClosed = new CountDownLatch(1);
doThrow(new RuntimeException("Consumer failed")).when(notificationHookConsumer).startAsyncImportConsumer(NotificationInterface.NotificationType.ASYNC_IMPORT, "import123", "topic1");
doAnswer(invocation -> {
- Object newStatus = invocation.getArgument(0);
- when(importRequest.getStatus()).thenReturn((AtlasAsyncImportRequest.ImportStatus) newStatus);
+ when(request.getStatus()).thenReturn(invocation.getArgument(0));
return null;
- }).when(importRequest).setStatus(any());
+ }).when(request).setStatus(any());
doAnswer(invocation -> {
consumerClosed.countDown();
return null;
}).when(notificationHookConsumer).closeImportConsumer(anyString(), anyString());
- ExecutorService realExecutor = java.util.concurrent.Executors.newSingleThreadExecutor();
- Field executorField = ImportTaskListenerImpl.class.getDeclaredField("executorService");
+ setExecutorService(importTaskListener, synchronousExecutor());
- executorField.setAccessible(true);
- executorField.set(importTaskListener, realExecutor);
-
- when(requestQueue.poll(anyLong(), any(TimeUnit.class))).thenReturn("import123");
-
- importTaskListener.onReceiveImportRequest(importRequest);
+ importTaskListener.onReceiveImportRequest(request);
assertTrue(consumerClosed.await(5, TimeUnit.SECONDS), "closeImportConsumer was not invoked");
@@ -584,20 +570,15 @@ public void testStartInternalIsNonBlocking() throws InterruptedException {
@Test
public void testImportNotProcessedWhenPassive() throws Exception {
- Mockito.doReturn("import123").when(importRequest).getImportId();
when(importRequest.getStatus()).thenReturn(WAITING);
when(requestQueue.poll(anyLong(), any(TimeUnit.class))).thenReturn("import123");
importTaskListener.instanceIsPassive();
importTaskListener.onReceiveImportRequest(importRequest);
- Thread.sleep(200);
verify(notificationHookConsumer, never()).startAsyncImportConsumer(any(), anyString(), anyString());
}
@Test
public void testExecutorNotRecreatedWhenPassive() throws Exception {
- Mockito.doReturn("import123").when(importRequest).getImportId();
- when(importRequest.getStatus()).thenReturn(WAITING);
- when(requestQueue.poll(anyLong(), any(TimeUnit.class))).thenReturn("import123");
when(importRequest.getStatus()).thenReturn(WAITING);
when(requestQueue.poll(anyLong(), any(TimeUnit.class))).thenReturn("import123");
importTaskListener.instanceIsPassive();
@@ -608,7 +589,6 @@ public void testExecutorNotRecreatedWhenPassive() throws Exception {
exec.shutdownNow();
}
importTaskListener.onReceiveImportRequest(importRequest);
- Thread.sleep(200);
ExecutorService execAfter = (ExecutorService) executorField.get(importTaskListener);
// Should remain null when passive
assertTrue(execAfter == null);
@@ -744,6 +724,34 @@ public void ensureExecutorAliveReturnsNullWhenPassiveEvenUnderConcurrency() thro
callers.shutdownNow();
}
+ private AtlasAsyncImportRequest createImportRequestMock(String importId, String topicName) {
+ AtlasAsyncImportRequest request = mock(AtlasAsyncImportRequest.class);
+
+ when(request.getImportId()).thenReturn(importId);
+ when(request.getTopicName()).thenReturn(topicName);
+
+ return request;
+ }
+
+ private ExecutorService synchronousExecutor() {
+ ExecutorService executor = mock(ExecutorService.class);
+
+ doAnswer(invocation -> {
+ Runnable task = invocation.getArgument(0);
+ task.run();
+ return null;
+ }).when(executor).submit(any(Runnable.class));
+
+ return executor;
+ }
+
+ private void setExecutorService(ImportTaskListenerImpl listener, ExecutorService executor) throws Exception {
+ Field executorField = ImportTaskListenerImpl.class.getDeclaredField("executorService");
+
+ executorField.setAccessible(true);
+ executorField.set(listener, executor);
+ }
+
private void shutdownImportExecutor(ImportTaskListenerImpl listener) throws Exception {
if (listener == null) {
return;