feat(dar): Granted lifecycle, filters, sort, and self-service create policy

bootstrap/sql/migrations/native/2.0.1/mysql/schemaChanges.sql

@@ -9,3 +9,9 @@ CREATE TABLE IF NOT EXISTS task_migration_mapping (
     PRIMARY KEY (old_thread_id),
     KEY idx_task_migration_mapping_new_task_id (new_task_id)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
+
+-- Data Access Request: indexed approver column for the Approver filter on /v1/tasks and /v1/tasks/dataAccessRequests.
+ALTER TABLE task_entity
+    ADD COLUMN IF NOT EXISTS approvedById varchar(36)
+        GENERATED ALWAYS AS (json_unquote(json_extract(`json`,_utf8mb4'$.approvedById'))) STORED,
+    ADD KEY IF NOT EXISTS idx_approved_by_id (approvedById);

bootstrap/sql/migrations/native/2.0.1/postgres/schemaChanges.sql

@@ -11,3 +11,10 @@ CREATE TABLE IF NOT EXISTS task_migration_mapping (
 
 CREATE INDEX IF NOT EXISTS idx_task_migration_mapping_new_task_id
     ON task_migration_mapping (new_task_id);
+
+-- Data Access Request: indexed approver column for the Approver filter on /v1/tasks and /v1/tasks/dataAccessRequests.
+ALTER TABLE task_entity
+    ADD COLUMN IF NOT EXISTS approvedbyid character varying(36)
+        GENERATED ALWAYS AS ((json ->> 'approvedById'::text)) STORED;
+
+CREATE INDEX IF NOT EXISTS idx_task_approved_by_id ON task_entity (approvedbyid);

openmetadata-integration-tests/pom.xml

@@ -241,6 +241,25 @@
       <version>2.3.0</version>
       <scope>test</scope>
     </dependency>
+    <!-- Required by DriveFileUploadIT (test fixture generation: PDF + XLSX). -->
+    <dependency>
+      <groupId>org.apache.pdfbox</groupId>
+      <artifactId>pdfbox</artifactId>
+      <version>2.0.31</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.poi</groupId>
+      <artifactId>poi</artifactId>
+      <version>5.4.1</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.poi</groupId>
+      <artifactId>poi-ooxml</artifactId>
+      <version>5.4.1</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
   <profiles>
     <!-- MySQL + Elasticsearch (default) -->

openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/DataAccessRequestIT.java

@@ -60,13 +60,15 @@
  *   <li>Seed: DataAccessRequest form schema and DataAccessRequestTaskWorkflow are loaded on boot.
  *   <li>Create: POST /tasks with category=DataAccess, type=DataAccessRequest and an
  *       accessType+reason payload succeeds and lands the task at the "review" stage.
- *   <li>Approve: /resolve transitions the task to status=InProgress, stage="approved",
- *       and surfaces a "revoke" available transition (matches the IncidentResolution pattern).
- *   <li>Revoke: /resolve from the approved stage closes the task with status=Revoked and
- *       resolution.type=Revoked.
+ *   <li>Approve: /resolve transitions the task to status=Approved, stage="approved",
+ *       captures approvedBy/approvedAt, and surfaces "markAsGranted" + "revoke" transitions.
+ *   <li>Grant: /resolve with markAsGranted moves the task to status=Granted (active access).
+ *   <li>Revoke: /resolve from either Approved or Granted closes the task with status=Revoked.
  *   <li>Reject: alternative terminal path lands at status=Rejected.
  *   <li>Validation: missing required fields (accessType/reason) are rejected by the form
  *       schema validator.
+ *   <li>Policy: non-admin users can create DARs via the DataConsumerPolicy Create-task rule.
+ *   <li>Filters: /v1/tasks/dataAccessRequests honors status/accessType/requestedBy/sortOrder.
  * </ul>
  */
 @Execution(ExecutionMode.CONCURRENT)
@@ -133,12 +135,13 @@ void darWorkflowDefinitionIsSeeded() {
     List<String> nodeNames = workflow.getNodes().stream().map(n -> n.getName()).toList();
     assertTrue(nodeNames.contains("TaskReview"));
     assertTrue(nodeNames.contains("ApprovedAccess"));
+    assertTrue(nodeNames.contains("GrantedAccess"));
     assertTrue(nodeNames.contains("RejectedEnd"));
     assertTrue(nodeNames.contains("RevokedEnd"));
   }
 
   @Test
-  void createApproveAndRevokeLifecycle(TestNamespace ns) {
+  void createApproveGrantRevokeLifecycle(TestNamespace ns) {
     String tableFqn = createTargetTable(ns);
 
     Task created =
@@ -167,27 +170,47 @@ void createApproveAndRevokeLifecycle(TestNamespace ns) {
 
     Task reviewed = reviewTaskRef.get();
 
-    // Approve → moves to ApprovedAccess userTask. Status stays non-terminal so the workflow
-    // can continue to a Revoke transition (matches IncidentResolution pattern).
+    // Approve → status=Approved (awaiting grant). approvedBy/approvedAt captured.
+    // Available transitions: markAsGranted (provision) and revoke (back out).
     Task approved =
         SdkClients.adminClient()
             .tasks()
             .resolve(
                 reviewed.getId().toString(),
                 new ResolveTask().withTransitionId("approve").withComment("approved"));
 
-    assertEquals(TaskEntityStatus.InProgress, approved.getStatus());
+    assertEquals(TaskEntityStatus.Approved, approved.getStatus());
     assertEquals("approved", approved.getWorkflowStageId());
+    assertNotNull(approved.getApprovedBy(), "approvedBy must be captured on approve transition");
+    assertNotNull(approved.getApprovedById());
+    assertNotNull(approved.getApprovedAt());
     List<String> approvedTransitions =
         approved.getAvailableTransitions().stream().map(TaskAvailableTransition::getId).toList();
-    assertEquals(List.of("revoke"), approvedTransitions);
+    assertTrue(approvedTransitions.contains("markAsGranted"));
+    assertTrue(approvedTransitions.contains("revoke"));
 
-    // Revoke → terminal Revoked status with resolution.
-    Task revoked =
+    // Mark as granted → status=Granted (active access).
+    Task granted =
         SdkClients.adminClient()
             .tasks()
             .resolve(
                 approved.getId().toString(),
+                new ResolveTask().withTransitionId("markAsGranted").withComment("provisioned"));
+
+    assertEquals(TaskEntityStatus.Granted, granted.getStatus());
+    assertEquals("granted", granted.getWorkflowStageId());
+    // approvedBy must persist through the grant transition.
+    assertEquals(approved.getApprovedById(), granted.getApprovedById());
+    List<String> grantedTransitions =
+        granted.getAvailableTransitions().stream().map(TaskAvailableTransition::getId).toList();
+    assertEquals(List.of("revoke"), grantedTransitions);
+
+    // Revoke from Granted → terminal Revoked status with resolution.
+    Task revoked =
+        SdkClients.adminClient()
+            .tasks()
+            .resolve(
+                granted.getId().toString(),
                 new ResolveTask().withTransitionId("revoke").withComment("revoking"));
 
     assertEquals(TaskEntityStatus.Revoked, revoked.getStatus());
@@ -196,6 +219,32 @@ void createApproveAndRevokeLifecycle(TestNamespace ns) {
     assertTrue(revoked.getAvailableTransitions().isEmpty());
   }
 
+  @Test
+  void approvedCanBeRevokedWithoutGranting(TestNamespace ns) {
+    String tableFqn = createTargetTable(ns);
+    Task created =
+        SdkClients.adminClient().tasks().create(buildDarRequest(ns, tableFqn, "FullAccess"));
+
+    Task approved =
+        SdkClients.adminClient()
+            .tasks()
+            .resolve(
+                created.getId().toString(),
+                new ResolveTask().withTransitionId("approve").withComment("approved"));
+    assertEquals(TaskEntityStatus.Approved, approved.getStatus());
+
+    // Revoke directly from the Approved stage (admin backs out before granting).
+    Task revoked =
+        SdkClients.adminClient()
+            .tasks()
+            .resolve(
+                approved.getId().toString(),
+                new ResolveTask().withTransitionId("revoke").withComment("backing out"));
+
+    assertEquals(TaskEntityStatus.Revoked, revoked.getStatus());
+    assertEquals(TaskResolutionType.Revoked, revoked.getResolution().getType());
+  }
+
   @Test
   void rejectLandsAtTerminalRejectedStatus(TestNamespace ns) {
     String tableFqn = createTargetTable(ns);
@@ -256,4 +305,149 @@ void missingAccessTypeIsRejectedByFormSchema(TestNamespace ns) {
     assertThrows(
         InvalidRequestException.class, () -> SdkClients.adminClient().tasks().create(invalid));
   }
+
+  @Test
+  void nonAdminUserCanCreateDar(TestNamespace ns) {
+    // DataConsumerPolicy grants Create on resource=task to every authenticated user, so a
+    // non-admin user can file a DAR without an explicit role. Verifies the policy fix for the
+    // "Principal: ... operations [Create] not allowed" failure when adam.matthews2-style users
+    // tried to request access.
+    String tableFqn = createTargetTable(ns);
+    Task created =
+        SdkClients.user1Client().tasks().create(buildDarRequest(ns, tableFqn, "FullAccess"));
+
+    assertNotNull(created.getId());
+    assertEquals(TaskCategory.DataAccess, created.getCategory());
+    assertEquals(TaskEntityType.DataAccessRequest, created.getType());
+  }
+
+  @Test
+  void darListEndpointFiltersByAccessTypeAndStatusAndSorts(TestNamespace ns) throws Exception {
+    String tableFqn = createTargetTable(ns);
+
+    Task openFull =
+        SdkClients.adminClient().tasks().create(buildDarRequest(ns, tableFqn, "FullAccess"));
+    Task openColumn =
+        SdkClients.adminClient().tasks().create(buildDarRequest(ns, tableFqn, "ColumnLevel"));
+    Task approvedFull =
+        SdkClients.adminClient().tasks().create(buildDarRequest(ns, tableFqn, "FullAccess"));
+    SdkClients.adminClient()
+        .tasks()
+        .resolve(
+            approvedFull.getId().toString(),
+            new ResolveTask().withTransitionId("approve").withComment("approved"));
+
+    // Filter by dataset → all three DARs come back (newest first by default sort DESC on
+    // createdAt).
+    var byDataset =
+        SdkClients.adminClient()
+            .tasks()
+            .listDataAccessRequests(Map.of("dataset", tableFqn, "limit", "50"));
+    List<String> idsByDataset =
+        byDataset.getData().stream().map(t -> t.getId().toString()).toList();
+    assertTrue(idsByDataset.contains(openFull.getId().toString()));
+    assertTrue(idsByDataset.contains(openColumn.getId().toString()));
+    assertTrue(idsByDataset.contains(approvedFull.getId().toString()));
+
+    // Filter by accessType=ColumnLevel → only the ColumnLevel DAR comes back.
+    var byColumnAccess =
+        SdkClients.adminClient()
+            .tasks()
+            .listDataAccessRequests(
+                Map.of("dataset", tableFqn, "accessType", "ColumnLevel", "limit", "50"));
+    List<String> columnIds =
+        byColumnAccess.getData().stream().map(t -> t.getId().toString()).toList();
+    assertTrue(columnIds.contains(openColumn.getId().toString()));
+    assertFalse(columnIds.contains(openFull.getId().toString()));
+    assertFalse(columnIds.contains(approvedFull.getId().toString()));
+
+    // Filter by status=Approved → only the approved DAR comes back.
+    var byApproved =
+        SdkClients.adminClient()
+            .tasks()
+            .listDataAccessRequests(
+                Map.of("dataset", tableFqn, "status", "Approved", "limit", "50"));
+    List<String> approvedIds =
+        byApproved.getData().stream().map(t -> t.getId().toString()).toList();
+    assertEquals(List.of(approvedFull.getId().toString()), approvedIds);
+
+    // sortOrder=asc → oldest first; reverse of default DESC. Both lists span the same scope.
+    var ascending =
+        SdkClients.adminClient()
+            .tasks()
+            .listDataAccessRequests(Map.of("dataset", tableFqn, "sortOrder", "asc", "limit", "50"));
+    var descending =
+        SdkClients.adminClient()
+            .tasks()
+            .listDataAccessRequests(
+                Map.of("dataset", tableFqn, "sortOrder", "desc", "limit", "50"));
+    List<String> ascIds = ascending.getData().stream().map(t -> t.getId().toString()).toList();
+    List<String> descIds = descending.getData().stream().map(t -> t.getId().toString()).toList();
+    assertEquals(ascIds.size(), descIds.size());
+    // The first id of the ascending list is the last id of the descending list and vice versa.
+    assertEquals(ascIds.get(0), descIds.get(descIds.size() - 1));
+    assertEquals(ascIds.get(ascIds.size() - 1), descIds.get(0));
+  }
+
+  @Test
+  void darListEndpointFiltersByApprover(TestNamespace ns) throws Exception {
+    String tableFqn = createTargetTable(ns);
+    Task created =
+        SdkClients.adminClient().tasks().create(buildDarRequest(ns, tableFqn, "FullAccess"));
+    Task approved =
+        SdkClients.adminClient()
+            .tasks()
+            .resolve(
+                created.getId().toString(),
+                new ResolveTask().withTransitionId("approve").withComment("approved"));
+
+    String approverId = approved.getApprovedById();
+    assertNotNull(approverId, "approvedById must be captured on approve");
+
+    var byApprover =
+        SdkClients.adminClient()
+            .tasks()
+            .listDataAccessRequests(
+                Map.of("dataset", tableFqn, "approverId", approverId, "limit", "50"));
+    List<String> ids = byApprover.getData().stream().map(t -> t.getId().toString()).toList();
+    assertTrue(ids.contains(approved.getId().toString()));
+    // A DAR that was never approved by the same user must not appear.
+    Task openDar =
+        SdkClients.adminClient().tasks().create(buildDarRequest(ns, tableFqn, "FullAccess"));
+    var byApproverAgain =
+        SdkClients.adminClient()
+            .tasks()
+            .listDataAccessRequests(
+                Map.of("dataset", tableFqn, "approverId", approverId, "limit", "50"));
+    List<String> idsAgain =
+        byApproverAgain.getData().stream().map(t -> t.getId().toString()).toList();
+    assertFalse(idsAgain.contains(openDar.getId().toString()));
+  }
+
+  @Test
+  void darListEndpointExcludesNonDarTaskTypes(TestNamespace ns) throws Exception {
+    // Verifies that /v1/tasks/dataAccessRequests pre-scopes to category=DataAccess +
+    // type=DataAccessRequest so non-DAR tasks (e.g. a description-update task) never appear.
+    String tableFqn = createTargetTable(ns);
+
+    Task dar = SdkClients.adminClient().tasks().create(buildDarRequest(ns, tableFqn, "FullAccess"));
+
+    // Create a non-DAR task about the same entity.
+    CreateTask nonDar =
+        new CreateTask()
+            .withName(ns.prefix("non-dar-task"))
+            .withCategory(TaskCategory.MetadataUpdate)
+            .withType(TaskEntityType.DescriptionUpdate)
+            .withAbout(tableEntityLink(tableFqn))
+            .withPayload(Map.of("newDescription", "test"));
+    Task descTask = SdkClients.adminClient().tasks().create(nonDar);
+
+    var listed =
+        SdkClients.adminClient()
+            .tasks()
+            .listDataAccessRequests(Map.of("dataset", tableFqn, "limit", "50"));
+    List<String> ids = listed.getData().stream().map(t -> t.getId().toString()).toList();
+    assertTrue(ids.contains(dar.getId().toString()));
+    assertFalse(ids.contains(descTask.getId().toString()));
+  }
 }

openmetadata-sdk/src/main/java/org/openmetadata/sdk/services/tasks/TaskService.java

@@ -215,6 +215,31 @@ public ListResponse<Task> listVisible(
     return deserializeListResponse(responseStr);
   }
 
+  /**
+   * List Data Access Requests with DAR-specific filters and offset-based pagination.
+   * Pre-applies category=DataAccess and type=DataAccessRequest server-side.
+   *
+   * @param filters Optional filters (dataset, service, status, statusGroup, requestedBy,
+   *     requestedById, approver, approverId, accessType, domain, sortOrder, limit, offset,
+   *     include, fields).
+   */
+  public ListResponse<Task> listDataAccessRequests(Map<String, String> filters)
+      throws OpenMetadataException {
+    String path = basePath + "/dataAccessRequests";
+    RequestOptions.Builder optionsBuilder = RequestOptions.builder();
+    if (filters != null) {
+      filters.forEach(
+          (k, v) -> {
+            if (v != null) {
+              optionsBuilder.queryParam(k, v);
+            }
+          });
+    }
+    String responseStr =
+        httpClient.executeForString(HttpMethod.GET, path, null, optionsBuilder.build());
+    return deserializeListResponse(responseStr);
+  }
+
   // ==================== Comment Methods ====================
 
   /**

openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/CreateTask.java

@@ -627,7 +627,9 @@ static boolean isTerminalTaskStatus(TaskEntityStatus status) {
     return status != null
         && status != TaskEntityStatus.Open
         && status != TaskEntityStatus.InProgress
-        && status != TaskEntityStatus.Pending;
+        && status != TaskEntityStatus.Pending
+        && status != TaskEntityStatus.Approved
+        && status != TaskEntityStatus.Granted;
   }
 
   static boolean shouldSkipDeletedWorkflowManagedDraftTask(