ATLAS-5362: Atlas Modular Runtime Architecture — Service Decomposition, Graph-Native Coordination, Active-Active HA, and RunMode-Driven Execution

addons/hbase-testing-util/src/test/java/org/apache/atlas/hbase/TestHBaseTestingUtilSpinup.java

@@ -21,9 +21,10 @@
 import org.apache.hadoop.hbase.MiniHBaseCluster;
 import org.testng.annotations.Test;
 
-import java.io.IOException;
-import java.net.ServerSocket;
+import java.io.File;
+import java.nio.file.Files;
 import java.util.List;
+import java.util.UUID;
 
 import static org.testng.AssertJUnit.assertFalse;
 
@@ -34,12 +35,22 @@ public class TestHBaseTestingUtilSpinup {
     private final HBaseTestingUtility hBaseTestingUtility = new HBaseTestingUtility();
 
     public TestHBaseTestingUtilSpinup() throws Exception {
-        hBaseTestingUtility.getConfiguration().set("test.hbase.zookeeper.property.clientPort", String.valueOf(getFreePort()));
-        hBaseTestingUtility.getConfiguration().set("hbase.master.port", String.valueOf(getFreePort()));
-        hBaseTestingUtility.getConfiguration().set("hbase.master.info.port", String.valueOf(getFreePort()));
-        hBaseTestingUtility.getConfiguration().set("hbase.regionserver.port", String.valueOf(getFreePort()));
-        hBaseTestingUtility.getConfiguration().set("hbase.regionserver.info.port", String.valueOf(getFreePort()));
-        hBaseTestingUtility.getConfiguration().set("zookeeper.znode.parent", "/hbase-unsecure");
+        String runId = UUID.randomUUID().toString();
+        File   baseDir = Files.createTempDirectory("atlas-hbase-test-" + runId).toFile();
+
+        // Keep each test run isolated from stale local state and avoid fixed-port races.
+        hBaseTestingUtility.getConfiguration().set("hadoop.tmp.dir", new File(baseDir, "hadoop-tmp").getAbsolutePath());
+        hBaseTestingUtility.getConfiguration().set("hbase.rootdir", new File(baseDir, "hbase-root").toURI().toString());
+        hBaseTestingUtility.getConfiguration().set("hbase.zookeeper.property.dataDir", new File(baseDir, "zk-data").getAbsolutePath());
+        hBaseTestingUtility.getConfiguration().set("zookeeper.znode.parent", "/hbase-unsecure-" + runId);
+        hBaseTestingUtility.getConfiguration().set("test.hbase.zookeeper.property.clientPort", "0");
+        hBaseTestingUtility.getConfiguration().set("hbase.master.port", "0");
+        hBaseTestingUtility.getConfiguration().set("hbase.master.info.port", "0");
+        hBaseTestingUtility.getConfiguration().set("hbase.regionserver.port", "0");
+        hBaseTestingUtility.getConfiguration().set("hbase.regionserver.info.port", "0");
+        hBaseTestingUtility.getConfiguration().set("hbase.master.hostname", "localhost");
+        hBaseTestingUtility.getConfiguration().set("hbase.regionserver.hostname", "localhost");
+        hBaseTestingUtility.getConfiguration().set("hbase.regionserver.hostname.seen.by.master", "localhost");
         hBaseTestingUtility.getConfiguration().set("hbase.table.sanity.checks", "false");
     }
 
@@ -57,13 +68,4 @@ public void testGetMetaTableRows() throws Exception {
             hBaseTestingUtility.shutdownMiniCluster();
         }
     }
-
-    private static int getFreePort() throws IOException {
-        ServerSocket serverSocket = new ServerSocket(0);
-        int          port         = serverSocket.getLocalPort();
-
-        serverSocket.close();
-
-        return port;
-    }
 }

client/client-v1/src/test/java/org/apache/atlas/AtlasClientTest.java

@@ -233,10 +233,10 @@ public void shouldSelectActiveAmongMultipleServersIfHAIsEnabled() {
 
         when(firstResponse.getStatus()).thenReturn(Response.Status.OK.getStatusCode());
 
-        String passiveStatus = "{\"Status\":\"PASSIVE\"}";
+        String becomingActiveStatus = "{\"Status\":\"BECOMING_ACTIVE\"}";
 
-        when(firstResponse.getEntity(String.class)).thenReturn(passiveStatus);
-        when(firstResponse.getLength()).thenReturn(passiveStatus.length());
+        when(firstResponse.getEntity(String.class)).thenReturn(becomingActiveStatus);
+        when(firstResponse.getLength()).thenReturn(becomingActiveStatus.length());
 
         ClientResponse secondResponse = mock(ClientResponse.class);
 

common/src/main/java/org/apache/atlas/ha/HAConfiguration.java

@@ -21,6 +21,8 @@
 import org.apache.atlas.security.SecurityProperties;
 import org.apache.commons.configuration2.Configuration;
 import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -30,6 +32,8 @@
  * A wrapper for getting configuration entries related to HighAvailability.
  */
 public final class HAConfiguration {
+    private static final Logger LOG = LoggerFactory.getLogger(HAConfiguration.class);
+
     public static final String ATLAS_SERVER_ZK_ROOT_DEFAULT               = "/apache_atlas";
     public static final String ATLAS_SERVER_HA_PREFIX                     = "atlas.server.ha.";
     public static final String ZOOKEEPER_PREFIX                           = "zookeeper.";
@@ -57,14 +61,13 @@ private HAConfiguration() {
      * @return
      */
     public static boolean isHAEnabled(Configuration configuration) {
-        boolean ret;
+        boolean ret = false;
 
         if (configuration.containsKey(HAConfiguration.ATLAS_SERVER_HA_ENABLED_KEY)) {
             ret = configuration.getBoolean(ATLAS_SERVER_HA_ENABLED_KEY);
+            LOG.info("isHAEnabled: key '{}' found in config, value={}", ATLAS_SERVER_HA_ENABLED_KEY, ret);
         } else {
-            String[] ids = configuration.getStringArray(HAConfiguration.ATLAS_SERVER_IDS);
-
-            ret = ids != null && ids.length > 1;
+            LOG.info("isHAEnabled: key '{}' NOT found in config, defaulting to false", ATLAS_SERVER_HA_ENABLED_KEY);
         }
 
         return ret;

common/src/main/java/org/apache/atlas/repository/Constants.java

@@ -108,6 +108,19 @@ public final class Constants {
     public static final String PATCH_TYPE_PROPERTY_KEY        = encodePropertyKey(INTERNAL_PROPERTY_KEY_PREFIX + "patch.type");
     public static final String PATCH_ACTION_PROPERTY_KEY      = encodePropertyKey(INTERNAL_PROPERTY_KEY_PREFIX + "patch.action");
     public static final String PATCH_STATE_PROPERTY_KEY       = encodePropertyKey(INTERNAL_PROPERTY_KEY_PREFIX + "patch.state");
+    public static final String PATCH_APPLIED_BY_PROPERTY_KEY  = encodePropertyKey(INTERNAL_PROPERTY_KEY_PREFIX + "patch.appliedBy");
+    public static final String PATCH_APPLIED_AT_PROPERTY_KEY  = encodePropertyKey(INTERNAL_PROPERTY_KEY_PREFIX + "patch.appliedAt");
+    public static final String PATCH_CLAIMED_BY_PROPERTY_KEY  = encodePropertyKey(INTERNAL_PROPERTY_KEY_PREFIX + "patch.claimedBy");
+    public static final String PATCH_CLAIM_STARTED_AT_KEY     = encodePropertyKey(INTERNAL_PROPERTY_KEY_PREFIX + "patch.claimStartedAt");
+    /**
+     * TypeDef bootstrap claim keys.
+     */
+    public static final String TYPEDEF_BOOTSTRAP_FILE_KEY         = encodePropertyKey(INTERNAL_PROPERTY_KEY_PREFIX + "typedef.bootstrap.file");
+    public static final String TYPEDEF_BOOTSTRAP_STATE_KEY        = encodePropertyKey(INTERNAL_PROPERTY_KEY_PREFIX + "typedef.bootstrap.state");
+    public static final String TYPEDEF_BOOTSTRAP_CLAIMED_BY_KEY   = encodePropertyKey(INTERNAL_PROPERTY_KEY_PREFIX + "typedef.bootstrap.claimedBy");
+    public static final String TYPEDEF_BOOTSTRAP_CLAIM_STARTED_AT = encodePropertyKey(INTERNAL_PROPERTY_KEY_PREFIX + "typedef.bootstrap.claimStartedAt");
+    public static final String TYPEDEF_BOOTSTRAP_APPLIED_BY_KEY   = encodePropertyKey(INTERNAL_PROPERTY_KEY_PREFIX + "typedef.bootstrap.appliedBy");
+    public static final String TYPEDEF_BOOTSTRAP_APPLIED_AT_KEY   = encodePropertyKey(INTERNAL_PROPERTY_KEY_PREFIX + "typedef.bootstrap.appliedAt");
     /**
      * The homeId field is used when saving into Atlas a copy of an object that is being imported from another
      * repository. The homeId will be set to a String that identifies the other repository. The specific format
@@ -218,6 +231,68 @@ public final class Constants {
     public static final String TASK_START_TIME        = encodePropertyKey(TASK_PREFIX + "startTime");
     public static final String TASK_END_TIME          = encodePropertyKey(TASK_PREFIX + "endTime");
     public static final String TASK_TYPE_NAME         = INTERNAL_PROPERTY_KEY_PREFIX + "AtlasTaskDef";
+
+    /**
+     * Cluster-wide claim marker, used by {@code GraphClaimable} implementations to guarantee that a
+     * single node performs a deferred action at a time.
+     *
+     * <p>{@link #CLAIM_KEY} is registered as a globally unique property key, so the compare in the
+     * Compare-And-Swap is performed by the store rather than by the claimant.  Two nodes reading a
+     * claimable state and both writing their own marker is not a swap at all - neither write fails
+     * - and on the rdbms backend there is no locking exception to lose the race with.  Uniqueness
+     * of the claim name is what makes exactly one write succeed.
+     *
+     * <p>The claim name identifies what is being serialised; the holder vertex is whatever the
+     * claimant is working on (for tasks, the task vertex itself).
+     */
+    public static final String CLAIM_KEY       = encodePropertyKey(INTERNAL_PROPERTY_KEY_PREFIX + "claim");
+    public static final String CLAIM_OWNER_KEY = encodePropertyKey(INTERNAL_PROPERTY_KEY_PREFIX + "claimOwner");
+    public static final String CLAIM_TIME_KEY  = encodePropertyKey(INTERNAL_PROPERTY_KEY_PREFIX + "claimTime");
+
+    /**
+     * When a leased claim stops being honoured, so peers can take over from a holder that died.
+     *
+     * <p>The instant is stored rather than a duration because it is the <em>holder's</em> lease that
+     * decides when its claim lapses.  A peer must not apply its own idea of how long the work should
+     * take: a six-hour purge would be displaced by anyone checking with a two-minute threshold.
+     */
+    public static final String CLAIM_EXPIRY_KEY = encodePropertyKey(INTERNAL_PROPERTY_KEY_PREFIX + "claimExpiry");
+
+    /**
+     * Marks a vertex that exists only to hold a claim.
+     *
+     * <p>Uniqueness discriminates between vertices, not between writers of one vertex: a second
+     * write of the same claim to the same vertex changes nothing, and a write of a different value
+     * drops the old uniqueness entry before adding its own.  So a claim recorded on a shared
+     * singleton vertex is not exclusive at all.  Claimants that have no natural per-claimant vertex
+     * (leases such as purge or index recovery, where the resource is a single vertex shared by
+     * every node) create one of these instead, and creation is what the store adjudicates.
+     */
+    public static final String CLAIM_VERTEX_TYPE_KEY  = encodePropertyKey(INTERNAL_PROPERTY_KEY_PREFIX + "claim_v_type");
+    public static final String CLAIM_VERTEX_TYPE_NAME = INTERNAL_PROPERTY_KEY_PREFIX + "AtlasClaim";
+
+    /** Claim names serialising each deferred action across the cluster. */
+    public static final String CLAIM_TASK_RUNNER  = "ATLAS_TASK_RUNNER";
+    public static final String CLAIM_PURGE        = "ATLAS_PURGE";
+    public static final String CLAIM_ASYNC_IMPORT = "ATLAS_ASYNC_IMPORT";
+    /**
+     * Guards loading the bootstrap models, which one node does for the whole cluster.
+     *
+     * <p>Sharing the model files out between nodes looked like the faster way to start, but a node
+     * only holds the types it loaded itself, so every node ended up with part of the schema.  Patches
+     * are claimed individually and land on whichever node takes them, and one that had not loaded the
+     * model failed against a type that was already in the store.  Loading is therefore done by one
+     * node, and the others read the finished types back.
+     */
+    public static final String CLAIM_TYPEDEF_BOOTSTRAP = "ATLAS_TYPEDEF_BOOTSTRAP";
+    /** Patches are claimed one at a time each, so unrelated patches can still proceed in parallel. */
+    public static final String CLAIM_PATCH_PREFIX = "ATLAS_PATCH:";
+
+    /**
+     * Guards index work.  Index initialization and index recovery share one claim deliberately, so
+     * that recovery never runs against a half-built index.
+     */
+    public static final String CLAIM_INDEX = "ATLAS_INDEX";
     /**
      * Index Recovery vertex property keys.
      */

common/src/test/java/org/apache/atlas/ha/HAConfigurationTest.java

@@ -63,13 +63,13 @@ public void testIsHAEnabledByLegacyConfiguration() {
     }
 
     @Test
-    public void testIsHAEnabledByIds() {
+    public void testIsHAEnabledByIds_doesNotInferWhenFlagMissing() {
         when(configuration.containsKey(HAConfiguration.ATLAS_SERVER_HA_ENABLED_KEY)).thenReturn(false);
         when(configuration.getStringArray(HAConfiguration.ATLAS_SERVER_IDS)).thenReturn(TEST_ATLAS_SERVER_IDS_HA);
         boolean isHAEnabled = HAConfiguration.isHAEnabled(configuration);
-        assertTrue(isHAEnabled);
+        assertFalse(isHAEnabled);
 
-        // restore
+        // single-id remains disabled as well when explicit flag is absent
         when(configuration.getStringArray(HAConfiguration.ATLAS_SERVER_IDS)).thenReturn(new String[] {"id1"});
         isHAEnabled = HAConfiguration.isHAEnabled(configuration);
         assertFalse(isHAEnabled);

dev-support/atlas-docker/.env.active-active

@@ -0,0 +1,64 @@
+# =============================================================================
+# Atlas Active-Active ADDITIONAL settings
+#
+# All infrastructure versions (HADOOP_VERSION, HBASE_VERSION, KAFKA_VERSION,
+# ATLAS_VERSION, etc.) are already defined in the existing .env file — do NOT
+# duplicate them here.
+#
+# Usage (two options):
+#
+#   Option A — append to the shared .env (simplest):
+#       cat .env.active-active >> .env
+#       docker compose -f docker-compose.atlas-active-active.yml up -d
+#
+#   Option B — pass both files explicitly:
+#       docker compose \
+#         --env-file .env \
+#         --env-file .env.active-active \
+#         -f docker-compose.atlas-active-active.yml up -d
+#
+#   Scale replicas without editing this file:
+#       METADATA_SERVER_REPLICAS=5 docker compose \
+#         -f docker-compose.atlas-active-active.yml up -d
+#     or:
+#       docker compose -f docker-compose.atlas-active-active.yml \
+#         up -d --scale atlas-metadata-server=5
+# =============================================================================
+
+# ---------------------------------------------------------------------------
+# Replica counts — only active-active specific settings below
+# ---------------------------------------------------------------------------
+
+# Atlas graph backend selector for active-active:
+#   hbase    -> HBase graph + HBase audit
+#   postgres -> RDBMS graph + Postgres audit
+ATLAS_BACKEND=hbase
+
+# Number of METADATA_SERVER replicas (REST + search + entity CRUD)
+METADATA_SERVER_REPLICAS=2
+
+# Number of NOTIFICATION_PROCESSOR replicas (hook Kafka consumer)
+NOTIFICATION_PROC_REPLICAS=2
+
+# Host port for the Nginx load balancer (proxies to all metadata-server replicas)
+ATLAS_LB_PORT=21000
+
+# ---------------------------------------------------------------------------
+# Patch toggles for initializer run
+# ---------------------------------------------------------------------------
+# Rebuild mixed indexes (JAVA_PATCH_0000_006 / ReIndexPatch)
+ATLAS_REBUILD_INDEX=true
+# Enable composite index status promotion (JAVA_PATCH_0000_010)
+ATLAS_UPDATE_COMPOSITE_INDEX_STATUS=true
+
+# ---------------------------------------------------------------------------
+# Index recovery service toggles
+# ---------------------------------------------------------------------------
+# Enable background index recovery monitor on metadata nodes
+ATLAS_INDEX_RECOVERY_ENABLE=true
+
+# ---------------------------------------------------------------------------
+# Postgres backend pool tuning (used only when ATLAS_BACKEND=postgres)
+# ---------------------------------------------------------------------------
+ATLAS_RDBMS_MAX_POOL_SIZE=15
+ATLAS_RDBMS_MIN_IDLE=2

dev-support/atlas-docker/.env.bkp

@@ -0,0 +1,28 @@
+BUILD_HOST_SRC=true
+SKIPTESTS=true
+GIT_URL=https://github.com/apache/atlas.git
+BRANCH=master
+PROFILE=dist,external-hbase-solr
+
+# Java version for AtlasBase image.
+# This image gets used as base docker image for all images.
+# Valid values: 8, 11, 17
+ATLAS_BASE_JAVA_VERSION=8
+
+# Java version to use to build Apache Atlas
+# Valid values: 8, 11, 17
+ATLAS_BUILD_JAVA_VERSION=8
+
+# Java version to use to run Atlas server
+# Valid values: 8, 11, 17
+ATLAS_SERVER_JAVA_VERSION=8
+
+ATLAS_VERSION=3.0.0-SNAPSHOT
+UBUNTU_VERSION=20.04
+HADOOP_VERSION=3.4.2
+HBASE_VERSION=2.6.4
+KAFKA_VERSION=2.8.2
+HIVE_VERSION=3.1.3
+HIVE_HADOOP_VERSION=3.1.1
+
+ATLAS_BACKEND=hbase

dev-support/atlas-docker/Dockerfile.atlas

@@ -25,8 +25,11 @@ ENV JAVA_HOME=/usr/lib/jvm/java-${ATLAS_SERVER_JAVA_VERSION}-openjdk-${TARGETARC
 RUN update-java-alternatives --set /usr/lib/jvm/java-1.${ATLAS_SERVER_JAVA_VERSION}.0-openjdk-${TARGETARCH}
 
 COPY ./scripts/atlas.sh                                 ${ATLAS_SCRIPTS}/
+COPY ./scripts/atlas-active-active.sh                   ${ATLAS_SCRIPTS}/
 COPY ./dist/apache-atlas-${ATLAS_VERSION}-server.tar.gz /home/atlas/dist/
 
+RUN chmod +x ${ATLAS_SCRIPTS}/atlas.sh ${ATLAS_SCRIPTS}/atlas-active-active.sh
+
 RUN tar xfz /home/atlas/dist/apache-atlas-${ATLAS_VERSION}-server.tar.gz --directory=/opt/ && \
     ln -s /opt/apache-atlas-${ATLAS_VERSION} ${ATLAS_HOME} && \
     rm -f /home/atlas/dist/apache-atlas-${ATLAS_VERSION}-server.tar.gz && \