This is an automated email from the ASF dual-hosted git repository.
sarathsubramanian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/atlas.git
The following commit(s) were added to refs/heads/master by this push:
new 319a1d0d0 ATLAS-5338: Add Trino extractor integration and unit test
suite (#691)
319a1d0d0 is described below
commit 319a1d0d082042db90e333b61a76352808927936
Author: Ramachandran Krishnan <[email protected]>
AuthorDate: Thu Jul 9 23:29:08 2026 +0530
ATLAS-5338: Add Trino extractor integration and unit test suite (#691)
* ATLAS-5338: Add Trino extractor integration and unit test suite
Implement live integration tests that run the standalone extractor tarball
against Atlas + Trino, plus unit tests for cron validation and concurrent
job guards. Adds the trino-extractor-it Maven profile and README.
https://issues.apache.org/jira/browse/ATLAS-5338
* ATLAS-5338: Document components required for Trino extractor ITs
* ATLAS-5338: Skip Trino extractor ITs in default Maven verify.
Live-stack integration tests require Atlas and Trino; opt in with
-Ptrino-extractor-it so CI mvn verify does not invoke Failsafe here.
* ATLAS-5338: Use Java 8 compatible List helpers in TrinoExtractorITSupport
Replace List.of() with Arrays.asList() so test code compiles on JDK 8
(CI docker-build matrix).
* Harden atlas-hbase Docker startup for CI compose smoke test
RegionServer often lags Master on slow runners; the old healthcheck
(start_period 40s, 30s interval) and tail-on-HMaster-only entrypoint
caused flaky "atlas-hbase is unhealthy" failures during compose up --wait.
- Add shared atlas-hbase-healthcheck.sh (wget/curl) for Master + RS
- Wait up to 180s for both endpoints before marking container ready
- Keep container alive while Master or RegionServer JVMs are running
- Install wget/curl in atlas-base; tune healthcheck (120s start_period)
- CI: --wait-timeout 300 and dump logs when compose up fails
* Stabilize ImportTaskListenerImplTest async consumer tests.
Use per-test import request mocks and a synchronous executor so Mockito
stubs are not accessed concurrently, fixing the JDK 8 ClassCastException flake
in testStartImportConsumer_Successful.
Co-authored-by: Cursor <[email protected]>
---------
Co-authored-by: ramk <[email protected]>
Co-authored-by: Cursor <[email protected]>
---
.github/workflows/ci.yml | 9 +-
addons/trino-extractor/README.md | 185 +++++++
addons/trino-extractor/pom.xml | 116 +++-
.../test/assemblies/trino-extractor-tarball.xml | 59 ++
.../apache/atlas/trino/cli/TrinoExtractorIT.java | 329 +++++++++++-
.../atlas/trino/cli/TrinoExtractorITSupport.java | 598 +++++++++++++++++++++
.../apache/atlas/trino/cli/TrinoExtractorTest.java | 47 ++
dev-support/atlas-docker/Dockerfile.atlas-base | 2 +-
dev-support/atlas-docker/Dockerfile.atlas-hbase | 4 +-
.../atlas-docker/docker-compose.atlas-hbase.yml | 10 +-
.../{atlas-hbase.sh => atlas-hbase-healthcheck.sh} | 48 +-
dev-support/atlas-docker/scripts/atlas-hbase.sh | 33 +-
.../notification/ImportTaskListenerImplTest.java | 86 +--
13 files changed, 1415 insertions(+), 111 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index ededed192..909843307 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 000000000..0f6172320
--- /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 9a15f3edf..3ddce7de5 100644
--- a/addons/trino-extractor/pom.xml
+++ b/addons/trino-extractor/pom.xml
@@ -35,7 +35,7 @@
<properties>
<checkstyle.failOnViolation>true</checkstyle.failOnViolation>
<checkstyle.skip>false</checkstyle.skip>
- <!-- No live-stack ITs yet; avoid Failsafe/TestNG on default mvn
verify (see ATLAS-5338). -->
+ <!-- Live-stack ITs are opt-in via -Ptrino-extractor-it (default CI
verify skips them). -->
<skipITs>true</skipITs>
</properties>
@@ -64,6 +64,18 @@
<artifactId>quartz</artifactId>
<version>2.3.2</version>
</dependency>
+
+ <dependency>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.testng</groupId>
+ <artifactId>testng</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
<plugins>
@@ -90,4 +102,106 @@
</plugin>
</plugins>
</build>
+
+ <profiles>
+ <profile>
+ <id>trino-extractor-it</id>
+ <properties>
+ <skipITs>false</skipITs>
+ </properties>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-antrun-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>refresh-trino-extractor-lib</id>
+ <goals>
+ <goal>run</goal>
+ </goals>
+ <phase>pre-integration-test</phase>
+ <configuration>
+ <target>
+ <delete
dir="${project.build.directory}/dependency/trino" quiet="true" />
+ </target>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>copy-dependencies-for-it-tarball</id>
+ <goals>
+ <goal>copy-dependencies</goal>
+ </goals>
+ <phase>pre-integration-test</phase>
+ <configuration>
+ <excludeScope>test</excludeScope>
+ <includeScope>compile</includeScope>
+
<outputDirectory>${project.build.directory}/dependency/trino</outputDirectory>
+ <overWriteReleases>true</overWriteReleases>
+
<overWriteSnapshots>true</overWriteSnapshots>
+ <overWriteIfNewer>true</overWriteIfNewer>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>package-trino-extractor-tarball</id>
+ <goals>
+ <goal>single</goal>
+ </goals>
+ <phase>pre-integration-test</phase>
+ <configuration>
+ <attach>false</attach>
+ <descriptors>
+
<descriptor>src/test/assemblies/trino-extractor-tarball.xml</descriptor>
+ </descriptors>
+
<finalName>apache-atlas-${project.version}</finalName>
+
<outputDirectory>${project.build.directory}/trino-extractor-dist</outputDirectory>
+ <skipAssembly>false</skipAssembly>
+ <tarLongFileMode>gnu</tarLongFileMode>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-failsafe-plugin</artifactId>
+ <configuration>
+ <includes>
+ <include>**/*IT.java</include>
+ </includes>
+ </configuration>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.maven.surefire</groupId>
+ <artifactId>surefire-testng</artifactId>
+ <version>${failsafe.version}</version>
+ </dependency>
+ </dependencies>
+ <executions>
+ <execution>
+ <goals>
+ <goal>integration-test</goal>
+ <goal>verify</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
</project>
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 000000000..03f4debb5
--- /dev/null
+++ b/addons/trino-extractor/src/test/assemblies/trino-extractor-tarball.xml
@@ -0,0 +1,59 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+ -->
+<assembly xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+
xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.2";
+
xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.2
http://maven.apache.org/xsd/assembly-1.1.2.xsd";>
+ <formats>
+ <format>tar.gz</format>
+ </formats>
+ <id>trino-extractor</id>
+
<baseDirectory>apache-atlas-trino-extractor-${project.version}</baseDirectory>
+ <fileSets>
+ <fileSet>
+ <directory>src/main/conf</directory>
+ <outputDirectory>/conf</outputDirectory>
+ <includes>
+ <include>atlas-trino-extractor-logback.xml</include>
+ <include>atlas-trino-extractor.properties</include>
+ </includes>
+ <fileMode>0755</fileMode>
+ <directoryMode>0755</directoryMode>
+ </fileSet>
+ <fileSet>
+ <directory>src/main/bin</directory>
+ <outputDirectory>/bin</outputDirectory>
+ <includes>
+ <include>run-trino-extractor.sh</include>
+ </includes>
+ <fileMode>0755</fileMode>
+ <directoryMode>0755</directoryMode>
+ </fileSet>
+ <fileSet>
+ <directory>target/dependency/trino</directory>
+ <outputDirectory>/lib</outputDirectory>
+ </fileSet>
+ <fileSet>
+ <directory>target</directory>
+ <outputDirectory>/lib</outputDirectory>
+ <includes>
+ <include>atlas-trino-extractor-*.jar</include>
+ </includes>
+ </fileSet>
+ </fileSets>
+</assembly>
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 e0e74ff9b..b3474c300 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.
+ * <p>
+ * 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<String, String> 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<String, String> 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<String, String> 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<String, String> 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<String, String> 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 000000000..362ff46e7
--- /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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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<String> jerseyClientJars = new ArrayList<>();
+
+ try (DirectoryStream<Path> 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<String> 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<String> 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<String, String> 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<String> entries = new ArrayList<>();
+ entries.add(workDir.resolve("conf").toString());
+
+ try (DirectoryStream<Path> 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<Path> 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<Path> 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<String, String> 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<String, String> 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 000000000..d85e10387
--- /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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 abc2b68d3..a7af42844 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 a05bda3a1..c4e7e7910 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 9d46702c2..82bdf10dc 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.sh
b/dev-support/atlas-docker/scripts/atlas-hbase-healthcheck.sh
old mode 100755
new mode 100644
similarity index 50%
copy from dev-support/atlas-docker/scripts/atlas-hbase.sh
copy to dev-support/atlas-docker/scripts/atlas-hbase-healthcheck.sh
index e139a9377..ddab30595
--- a/dev-support/atlas-docker/scripts/atlas-hbase.sh
+++ b/dev-support/atlas-docker/scripts/atlas-hbase-healthcheck.sh
@@ -1,5 +1,4 @@
#!/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
@@ -16,38 +15,19 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-service ssh start
-
-if [ ! -e ${HBASE_HOME}/.setupDone ]
-then
- su -c "ssh-keygen -t rsa -P '' -f ~/.ssh/id_rsa" hbase
- su -c "cat ~/.ssh/id_rsa.pub >> ~/.ssh/authorized_keys" hbase
- su -c "chmod 0600 ~/.ssh/authorized_keys" hbase
-
- echo "ssh" > /etc/pdsh/rcmd_default
-
- ${ATLAS_SCRIPTS}/atlas-hbase-setup.sh
-
- touch ${HBASE_HOME}/.setupDone
-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
- break
+# 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
+}
- sleep 2
-done
-
-if [ -z "${HBASE_MASTER_PID}" ]; then
- echo "HBase HMaster failed to start" >&2
- exit 1
-fi
-
-# prevent the container from exiting
-tail --pid=$HBASE_MASTER_PID -f /dev/null
+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 e139a9377..3c986c80c 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 c0fdfddf5..422f7d4cf 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 class ImportTaskListenerImplTest {
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 class ImportTaskListenerImplTest {
@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 class ImportTaskListenerImplTest {
@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 class ImportTaskListenerImplTest {
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 class ImportTaskListenerImplTest {
@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 class ImportTaskListenerImplTest {
@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 class ImportTaskListenerImplTest {
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 class ImportTaskListenerImplTest {
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;
