ATLAS-5317: Resilient bulk purge with pre-validation, worker batches,…

addons/models/0000-Area0/0010-base_model.json

@@ -15,7 +15,19 @@
         { "ordinal": 6, "value": "TYPE_DEF_UPDATE" },
         { "ordinal": 7, "value": "TYPE_DEF_DELETE" },
         { "ordinal": 8, "value": "SERVER_START" },
-        { "ordinal": 9, "value": "SERVER_STATE_ACTIVE" }
+        { "ordinal": 9, "value": "SERVER_STATE_ACTIVE" },
+        { "ordinal": 10, "value": "AUTO_PURGE" }
+      ]
+    },
+    {
+      "name":        "audit_row_kind",
+      "description": "Distinguishes single, summary, and batch rows in a correlated audit run",
+      "typeVersion": "1.0",
+      "serviceType": "atlas_core",
+      "elementDefs": [
+        { "ordinal": 0, "value": "SINGLE" },
+        { "ordinal": 1, "value": "SUMMARY" },
+        { "ordinal": 2, "value": "BATCH" }
       ]
     }
   ],
@@ -165,7 +177,9 @@
         { "name": "clientId",    "typeName": "string",          "cardinality": "SINGLE", "isIndexable": true,  "isOptional": true,  "isUnique": false },
         { "name": "params",      "typeName": "string",          "cardinality": "SINGLE", "isIndexable": true,  "isOptional": true,  "isUnique": false },
         { "name": "result",      "typeName": "string",          "cardinality": "SINGLE", "isIndexable": false, "isOptional": true,  "isUnique": false },
-        { "name": "resultCount", "typeName": "long",            "cardinality": "SINGLE", "isIndexable": true,  "isOptional": true,  "isUnique": false }
+        { "name": "resultCount", "typeName": "long",            "cardinality": "SINGLE", "isIndexable": true,  "isOptional": true,  "isUnique": false },
+        { "name": "runId",        "typeName": "string",          "cardinality": "SINGLE", "isIndexable": true,  "isOptional": true,  "isUnique": false },
+        { "name": "auditRowKind", "typeName": "audit_row_kind",  "cardinality": "SINGLE", "isIndexable": true,  "isOptional": true,  "isUnique": false }
       ]
     },
     {

addons/models/0000-Area0/patches/009-base_model_add_audit_correlation_attributes.json

@@ -0,0 +1,17 @@
+{
+  "patches": [
+    {
+      "id":              "TYPEDEF_PATCH_0009_001",
+      "description":     "Add runId and auditRowKind to __AtlasAuditEntry for purge audit correlation",
+      "action":          "ADD_ATTRIBUTE",
+      "typeName":        "__AtlasAuditEntry",
+      "applyToVersion":  "1.0",
+      "updateToVersion": "1.1",
+      "params":          null,
+      "attributeDefs": [
+        { "name": "runId",        "typeName": "string",         "cardinality": "SINGLE", "isIndexable": true, "isOptional": true, "isUnique": false },
+        { "name": "auditRowKind", "typeName": "audit_row_kind", "cardinality": "SINGLE", "isIndexable": true, "isOptional": true, "isUnique": false }
+      ]
+    }
+  ]
+}

client/client-v2/src/test/java/org/apache/atlas/AtlasClientV2Test.java

@@ -43,6 +43,9 @@
 import org.apache.atlas.model.instance.AtlasRelationship;
 import org.apache.atlas.model.instance.ClassificationAssociateRequest;
 import org.apache.atlas.model.instance.EntityMutationResponse;
+import org.apache.atlas.model.instance.EntityMutations.EntityOperation;
+import org.apache.atlas.model.instance.FailedEntity;
+import org.apache.atlas.model.instance.PurgeSummary;
 import org.apache.atlas.model.lineage.AtlasLineageInfo.LineageDirection;
 import org.apache.atlas.model.profile.AtlasUserSavedSearch;
 import org.apache.atlas.model.typedef.AtlasBusinessMetadataDef;
@@ -52,8 +55,10 @@
 import org.apache.atlas.model.typedef.AtlasRelationshipDef;
 import org.apache.atlas.model.typedef.AtlasStructDef;
 import org.apache.atlas.model.typedef.AtlasTypesDef;
+import org.apache.atlas.utils.AtlasJson;
 import org.apache.commons.configuration2.Configuration;
 import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.http.HttpStatus;
 import org.mockito.ArgumentMatchers;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
@@ -79,6 +84,7 @@
 
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 import static org.testng.Assert.assertEquals;
@@ -2512,6 +2518,79 @@ public void testPurgeEntitiesByGuidsRealExecution() throws Exception {
         assertEquals(result, mockResponse);
     }
 
+    @Test
+    public void testPurgeEntitiesByGuidsAssignsToBaseType() throws Exception {
+        TestableAtlasClientV2 client = new TestableAtlasClientV2();
+        EntityMutationResponse mockResponse = new EntityMutationResponse();
+        client.setMockResponse(mockResponse, EntityMutationResponse.class);
+
+        EntityMutationResponse result = client.purgeEntitiesByGuids(Collections.singleton("guid1"));
+
+        assertNotNull(result);
+        assertEquals(result, mockResponse);
+    }
+
+    @Test
+    public void testPurgeEntitiesByGuidsDeserializesPurgeFieldsOnHttp200() throws Exception {
+        EntityMutationResponse result = invokePurgeWithMockedHttpResponse(HttpStatus.SC_OK, buildSamplePurgeResponseJson());
+
+        assertPurgeResponseFields(result);
+    }
+
+    @Test
+    public void testPurgeEntitiesByGuidsDeserializedResponseAssignsToBaseType() throws Exception {
+        EntityMutationResponse result = invokePurgeWithMockedHttpResponse(HttpStatus.SC_OK, buildSamplePurgeResponseJson());
+
+        assertNotNull(result.getEntitiesByOperation(EntityOperation.PURGE));
+        assertEquals(result.getEntitiesByOperation(EntityOperation.PURGE).size(), 1);
+        assertEquals(result.getEntitiesByOperation(EntityOperation.PURGE).get(0).getGuid(), "purged-guid");
+    }
+
+    private EntityMutationResponse invokePurgeWithMockedHttpResponse(int httpStatus, String responseJson) throws Exception {
+        AtlasClientV2       atlasClient = new AtlasClientV2(service, configuration);
+        WebResource.Builder builder     = setupBuilder(AtlasClientV2.API_V2.PURGE_ENTITIES_BY_GUIDS, service);
+        ClientResponse      response    = mock(ClientResponse.class);
+
+        when(response.getStatus()).thenReturn(httpStatus);
+        when(response.getEntity(String.class)).thenReturn(responseJson);
+        when(builder.method(eq(javax.ws.rs.HttpMethod.PUT), eq(ClientResponse.class), any())).thenReturn(response);
+
+        return atlasClient.purgeEntitiesByGuids(Collections.singleton("purged-guid"));
+    }
+
+    private static String buildSamplePurgeResponseJson() {
+        EntityMutationResponse response = new EntityMutationResponse();
+
+        Map<EntityOperation, List<AtlasEntityHeader>> mutatedEntities = new HashMap<>();
+        List<AtlasEntityHeader> purged = new ArrayList<>();
+        purged.add(new AtlasEntityHeader("Table", "purged-guid", null));
+        mutatedEntities.put(EntityOperation.PURGE, purged);
+        response.setMutatedEntities(mutatedEntities);
+
+        response.addFailedEntity(new FailedEntity("failed-guid", "ATLAS-404-00-006", "instance not found"));
+        response.setSummary(new PurgeSummary(2, 1, 0, 1, 0));
+
+        return AtlasJson.toJson(response);
+    }
+
+    private static void assertPurgeResponseFields(EntityMutationResponse result) {
+        assertNotNull(result);
+        assertNotNull(result.getPurgedEntities());
+        assertEquals(result.getPurgedEntities().size(), 1);
+        assertEquals(result.getPurgedEntities().get(0).getGuid(), "purged-guid");
+
+        assertNotNull(result.getFailedEntities());
+        assertEquals(result.getFailedEntities().size(), 1);
+        assertEquals(result.getFailedEntities().get(0).getGuid(), "failed-guid");
+        assertEquals(result.getFailedEntities().get(0).getErrorCode(), "ATLAS-404-00-006");
+        assertEquals(result.getFailedEntities().get(0).getErrorMessage(), "instance not found");
+
+        assertNotNull(result.getPurgeSummary());
+        assertEquals(result.getPurgeSummary().getRequestedCount(), 2);
+        assertEquals(result.getPurgeSummary().getPurgedCount(), 1);
+        assertEquals(result.getPurgeSummary().getFailedCount(), 1);
+    }
+
     @Test
     public void testAddClassificationByEntityRequest() throws Exception {
         TestableAtlasClientV2 client = new TestableAtlasClientV2();

distro/src/conf/atlas-application.properties

@@ -277,6 +277,17 @@ atlas.rest.notification.enableTLS=false
 #atlas.headers.<headerName>=<headerValue>
 
 
+#########  Purge Configuration ########
+
+# Maximum number of GUIDs allowed per REST purge request (PUT /api/atlas/v2/admin/purge).
+#atlas.purge.api.max.request.size=1000
+
+# Number of worker threads for REST and scheduled bulk purge (WorkItemManager consumers).
+#atlas.purge.workers.count=2
+
+# Number of GUIDs committed per worker batch during bulk purge.
+#atlas.purge.worker.batch.size=100
+
 #########  UI Configuration ########
 
 #atlas.ui.default.version=v2

distro/src/conf/atlas-logback.xml

@@ -106,6 +106,19 @@
     </rollingPolicy>
   </appender>
 
+  <appender name="PURGE_FAILURE" class="ch.qos.logback.core.rolling.RollingFileAppender">
+    <file>${atlas.log.dir}/purgefailure.log</file>
+    <append>true</append>
+    <encoder>
+      <pattern>%date [%thread] %level{5} [%file:%line] %msg%n</pattern>
+    </encoder>
+    <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
+      <fileNamePattern>${atlas.log.dir}/purgefailure-%d.log</fileNamePattern>
+      <maxHistory>20</maxHistory>
+      <cleanHistoryOnStart>false</cleanHistoryOnStart>
+    </rollingPolicy>
+  </appender>
+
   <appender name="NOTIFICATION_PROCESSOR" class="ch.qos.logback.core.rolling.RollingFileAppender">
     <file>${atlas.log.dir}/notification_processor.log</file>
     <append>true</append>
@@ -180,6 +193,10 @@
     <appender-ref ref="FAILED"/>
   </logger>
 
+  <logger name="PURGE_FAILURE" additivity="false" level="info">
+    <appender-ref ref="PURGE_FAILURE"/>
+  </logger>
+
   <logger name="TASKS" additivity="false" level="info">
     <appender-ref ref="TASKS"/>
   </logger>

intg/src/main/java/org/apache/atlas/AtlasConfiguration.java

@@ -121,7 +121,8 @@ public enum AtlasConfiguration {
     ATLAS_ASYNC_IMPORT_MIN_DURATION_OVERRIDE_TEST_AUTOMATION("atlas.async.import.min.duration.override.test.automation", false),
     ASYNC_IMPORT_TOPIC_PREFIX("atlas.async.import.topic.prefix", "ATLAS_IMPORT_"),
     ASYNC_IMPORT_REQUEST_ID_PREFIX("atlas.async.import.request_id.prefix", "async_import_"),
-    REPLACE_HUGE_SPARK_PROCESS_ATTRIBUTES_PATCH("atlas.process.spark.attributes.update.patch", false);
+    REPLACE_HUGE_SPARK_PROCESS_ATTRIBUTES_PATCH("atlas.process.spark.attributes.update.patch", false),
+    PURGE_API_MAX_REQUEST_SIZE("atlas.purge.api.max.request.size", 1000);
     private static final Configuration APPLICATION_PROPERTIES;
 
     private final String propertyName;

intg/src/main/java/org/apache/atlas/AtlasErrorCode.java

@@ -184,6 +184,9 @@ public enum AtlasErrorCode {
     BLANK_VALUE_ATTRIBUTE(400, "ATLAS-400-00-105", "Value Attribute can't be empty!"),
     INVALID_RELATIONSHIP_LABEL(400, "ATLAS-400-00-106", "Invalid relationship label {0}. The referenced entity type {1} could not be resolved from the type registry."),
     NON_INDEXABLE_BM_DELETE_NOT_ALLOWED(400, "ATLAS-400-00-107", "Deletion not allowed for non-indexable Business Metadata ''{0}'' without force=true. Non-indexable attributes cannot be validated efficiently for references; use force=true to skip validation and delete (warning: orphaned references may remain)."),
+    INVALID_GUID(400, "ATLAS-400-00-108", "guid {0} is not a valid UUID"),
+    PURGE_REQUEST_SIZE_EXCEEDS_LIMIT(400, "ATLAS-400-00-109", "purge request size {0} exceeds maximum limit {1}"),
+    NOT_IN_DELETED_STATE(400, "ATLAS-400-00-10A", "entity {0} is not in DELETED state"),
 
     UNAUTHORIZED_ACCESS(403, "ATLAS-403-00-001", "{0} is not authorized to perform {1}"),
 

intg/src/main/java/org/apache/atlas/model/audit/AtlasAuditEntry.java

@@ -46,6 +46,8 @@ public class AtlasAuditEntry extends AtlasBaseModelObject implements Serializabl
     private String         clientId;
     private String         result;
     private long           resultCount;
+    private String         runId;
+    private AuditRowKind   auditRowKind;
 
     public AtlasAuditEntry() {
     }
@@ -120,6 +122,22 @@ public void setResultCount(long resultCount) {
         this.resultCount = resultCount;
     }
 
+    public String getRunId() {
+        return runId;
+    }
+
+    public void setRunId(String runId) {
+        this.runId = runId;
+    }
+
+    public AuditRowKind getAuditRowKind() {
+        return auditRowKind;
+    }
+
+    public void setAuditRowKind(AuditRowKind auditRowKind) {
+        this.auditRowKind = auditRowKind;
+    }
+
     @Override
     public StringBuilder toString(StringBuilder sb) {
         sb.append(", userName: ").append(userName);
@@ -130,10 +148,18 @@ public StringBuilder toString(StringBuilder sb) {
         sb.append(", endTime: ").append(endTime);
         sb.append(", result: ").append(result);
         sb.append(", resultCount: ").append(resultCount);
+        sb.append(", runId: ").append(runId);
+        sb.append(", auditRowKind: ").append(auditRowKind);
 
         return sb;
     }
 
+    public enum AuditRowKind {
+        SINGLE,
+        SUMMARY,
+        BATCH
+    }
+
     public enum AuditOperation {
         PURGE("PURGE"),
         AUTO_PURGE("AUTO_PURGE"),
@@ -155,6 +181,7 @@ public enum AuditOperation {
         public EntityAuditEventV2.EntityAuditActionV2 toEntityAuditActionV2() throws AtlasBaseException {
             switch (this.type) {
                 case "PURGE":
+                case "AUTO_PURGE":
                     return EntityAuditEventV2.EntityAuditActionV2.ENTITY_PURGE;
                 default:
                     try {