MINOR: Update Incident Manager Flow

openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/CollectionDAO.java

@@ -3708,12 +3708,6 @@ default String getTimeSeriesTableName() {
       return "data_quality_data_time_series";
     }
 
-    @SqlQuery(
-        value =
-            "SELECT DISTINCT incidentId FROM data_quality_data_time_series "
-                + "WHERE entityFQNHash = :testCaseFQNHash AND incidentId IS NOT NULL")
-    List<String> getResultsWithIncidents(@BindFQN("testCaseFQNHash") String testCaseFQNHash);
-
     @SqlUpdate(
         value =
             "UPDATE data_quality_data_time_series SET incidentId = NULL "

openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/EntityTimeSeriesRepository.java

@@ -38,11 +38,15 @@ protected EntityTimeSeriesRepository(
   @Transaction
   public T createNewRecord(T recordEntity, String extension, String recordFQN) {
     recordEntity.setId(UUID.randomUUID());
-    if (extension != null) {
-      timeSeriesDao.insert(recordFQN, extension, entityType, JsonUtils.pojoToJson(recordEntity));
-    } else {
-      timeSeriesDao.insert(recordFQN, entityType, JsonUtils.pojoToJson(recordEntity));
-    }
+    timeSeriesDao.insert(recordFQN, extension, entityType, JsonUtils.pojoToJson(recordEntity));
+    postCreate(recordEntity);
+    return recordEntity;
+  }
+
+  @Transaction
+  public T createNewRecord(T recordEntity, String recordFQN) {
+    recordEntity.setId(UUID.randomUUID());
+    timeSeriesDao.insert(recordFQN, entityType, JsonUtils.pojoToJson(recordEntity));
     postCreate(recordEntity);
     return recordEntity;
   }

openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/TestCaseRepository.java

@@ -16,6 +16,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import java.util.UUID;
 import javax.json.JsonPatch;
 import javax.ws.rs.core.Response;
@@ -281,7 +282,8 @@ public RestUtil.PutResponse<TestCaseResult> addTestCaseResult(
             JsonUtils.pojoToJson(testCaseResult),
             incidentStateId != null ? incidentStateId.toString() : null);
 
-    setFieldsInternal(testCase, new EntityUtil.Fields(allowedFields, TEST_SUITE_FIELD));
+    setFieldsInternal(
+        testCase, new EntityUtil.Fields(allowedFields, Set.of(TEST_SUITE_FIELD, INCIDENTS_FIELD)));
     setTestSuiteSummary(
         testCase, testCaseResult.getTimestamp(), testCaseResult.getTestCaseStatus(), false);
     setTestCaseResult(testCase, testCaseResult, false);
@@ -326,7 +328,7 @@ private UUID getOrCreateIncidentOnFailure(TestCase testCase, String updatedBy) {
 
     TestCaseResolutionStatus incident =
         testCaseResolutionStatusRepository.createNewRecord(
-            status, null, testCase.getFullyQualifiedName());
+            status, testCase.getFullyQualifiedName());
 
     return incident.getStateId();
   }
@@ -538,16 +540,12 @@ public ResultList<TestCaseResult> getTestCaseResults(String fqn, Long startTs, L
   private UUID getIncidentId(TestCase test) {
     UUID ongoingIncident = null;
 
-    List<UUID> incidents =
-        daoCollection
-            .dataQualityDataTimeSeriesDao()
-            .getResultsWithIncidents(test.getFullyQualifiedName())
-            .stream()
-            .map(UUID::fromString)
-            .toList();
+    String json =
+        daoCollection.dataQualityDataTimeSeriesDao().getLatestRecord(test.getFullyQualifiedName());
+    TestCaseResult latestTestCaseResult = JsonUtils.readValue(json, TestCaseResult.class);
 
-    if (!nullOrEmpty(incidents)) {
-      ongoingIncident = incidents.get(0);
+    if (!nullOrEmpty(latestTestCaseResult)) {
+      ongoingIncident = latestTestCaseResult.getIncidentId();
     }
 
     return ongoingIncident;
@@ -792,12 +790,6 @@ public TestCase performTask(String userName, ResolveTask resolveTask) {
               JsonUtils.pojoToJson(testCaseResolutionStatus));
       testCaseResolutionStatusRepository.postCreate(testCaseResolutionStatus);
 
-      // When we resolve a task, we clean up the test case results associated
-      // with the resolved stateId
-      dataQualityDataTimeSeriesDao.cleanTestCaseIncident(
-          latestTestCaseResolutionStatus.getTestCaseReference().getFullyQualifiedName(),
-          latestTestCaseResolutionStatus.getStateId().toString());
-
       // Return the TestCase with the StateId to avoid any unnecessary PATCH when resolving the task
       // in the feed repo,
       // since the `threadContext.getAboutEntity()` will give us the task with the `incidentId`

openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/TestCaseResolutionStatusRepository.java

@@ -141,7 +141,7 @@ private void validateStatus(
   @Override
   @Transaction
   public TestCaseResolutionStatus createNewRecord(
-      TestCaseResolutionStatus recordEntity, String extension, String recordFQN) {
+      TestCaseResolutionStatus recordEntity, String recordFQN) {
 
     TestCaseResolutionStatus lastIncident =
         getLatestRecord(recordEntity.getTestCaseReference().getFullyQualifiedName());
@@ -168,19 +168,12 @@ public TestCaseResolutionStatus createNewRecord(
 
     switch (recordEntity.getTestCaseResolutionStatusType()) {
       case New -> {
-        // If there is already an unresolved incident, return it without doing any
-        // further logic.
+        // If there is already an existing New incident we'll return it
         if (Boolean.TRUE.equals(unresolvedIncident(lastIncident))) {
-          return getLatestRecord(lastIncident.getTestCaseReference().getFullyQualifiedName());
+          return lastIncident;
         }
-        // When we create a NEW incident, we need to open a task with the test case owner or
-        // the table owner as the assignee.
-        openNewTask(recordEntity);
       }
-      case Ack -> {
-        /* nothing to do for ACK. The Owner already has the task open */
-      }
-      case Assigned -> assignTask(recordEntity, lastIncident);
+      case Ack, Assigned -> openOrAssignTask(recordEntity);
       case Resolved -> {
         // When the incident is Resolved, we will close the Assigned task.
         resolveTask(recordEntity, lastIncident);
@@ -192,36 +185,38 @@ public TestCaseResolutionStatus createNewRecord(
       default -> throw new IllegalArgumentException(
           String.format("Invalid status %s", recordEntity.getTestCaseResolutionStatusType()));
     }
-    return super.createNewRecord(recordEntity, extension, recordFQN);
-  }
-
-  private void openNewTask(TestCaseResolutionStatus incidentStatus) {
-
-    TestCase testCase =
-        Entity.getEntity(incidentStatus.getTestCaseReference(), "owner", Include.NON_DELETED);
-
-    createTask(incidentStatus, Collections.singletonList(testCase.getOwner()), "New Incident");
+    return super.createNewRecord(recordEntity, recordFQN);
   }
 
-  private void assignTask(
-      TestCaseResolutionStatus newIncidentStatus, TestCaseResolutionStatus lastIncidentStatus) {
-
-    if (lastIncidentStatus == null) {
-      throw new IncidentManagerException(
+  private void openOrAssignTask(TestCaseResolutionStatus incidentStatus) {
+    switch (incidentStatus.getTestCaseResolutionStatusType()) {
+      case Ack -> // If the incident has been acknowledged, the task will be assigned to the user
+      // who acknowledged it
+      createTask(
+          incidentStatus, Collections.singletonList(incidentStatus.getUpdatedBy()), "New Incident");
+      case Assigned -> {
+        // If no existing task is found (New -> Assigned), we'll create a new one,
+        // otherwise (Ack -> Assigned) we'll update the existing
+        Thread existingTask = getIncidentTask(incidentStatus);
+        Assigned assigned =
+            JsonUtils.convertValue(
+                incidentStatus.getTestCaseResolutionStatusDetails(), Assigned.class);
+        if (existingTask == null) {
+          // New -> Assigned flow
+          createTask(
+              incidentStatus, Collections.singletonList(assigned.getAssignee()), "New Incident");
+        } else {
+          // Ack -> Assigned or Assigned -> Assigned flow
+          patchTaskAssignee(
+              existingTask, assigned.getAssignee(), incidentStatus.getUpdatedBy().getName());
+        }
+      }
+        // Should not land in the default case as we only call this method for Ack and Assigned
+      default -> throw new IllegalArgumentException(
           String.format(
-              "Cannot find the last incident status for stateId %s",
-              newIncidentStatus.getStateId()));
+              "Task cannot be opened for status `%s`",
+              incidentStatus.getTestCaseResolutionStatusType()));
     }
-
-    Thread thread = getIncidentTask(lastIncidentStatus);
-
-    Assigned assigned =
-        JsonUtils.convertValue(
-            newIncidentStatus.getTestCaseResolutionStatusDetails(), Assigned.class);
-    User updatedBy =
-        Entity.getEntity(Entity.USER, newIncidentStatus.getUpdatedBy().getId(), "", Include.ALL);
-
-    patchTaskAssignee(thread, assigned.getAssignee(), updatedBy.getName());
   }
 
   private void resolveTask(
@@ -234,8 +229,6 @@ private void resolveTask(
               newIncidentStatus.getStateId()));
     }
 
-    Thread thread = getIncidentTask(lastIncidentStatus);
-
     Resolved resolved =
         JsonUtils.convertValue(
             newIncidentStatus.getTestCaseResolutionStatusDetails(), Resolved.class);
@@ -249,11 +242,21 @@ private void resolveTask(
             .withTestCaseFQN(testCase.getFullyQualifiedName())
             .withTestCaseFailureReason(resolved.getTestCaseFailureReason())
             .withNewValue(resolved.getTestCaseFailureComment());
-    Entity.getFeedRepository()
-        .resolveTask(
-            new FeedRepository.ThreadContext(thread),
-            updatedBy.getFullyQualifiedName(),
-            resolveTask);
+
+    Thread thread = getIncidentTask(lastIncidentStatus);
+
+    if (thread != null) {
+      // If there is an existing task, we'll resolve it and create a new incident
+      // status with the Resolved status flow
+      Entity.getFeedRepository()
+          .resolveTask(
+              new FeedRepository.ThreadContext(thread),
+              updatedBy.getFullyQualifiedName(),
+              resolveTask);
+    } else {
+      // if there is no task, we'll simply create a new incident status (e.g. New -> Resolved)
+      super.createNewRecord(newIncidentStatus, testCase.getFullyQualifiedName());
+    }
   }
 
   private void createTask(

openmetadata-service/src/main/java/org/openmetadata/service/resources/EntityTimeSeriesResource.java

@@ -31,4 +31,9 @@ protected Response create(T entity, String extension, String recordFQN) {
     entity = repository.createNewRecord(entity, extension, recordFQN);
     return Response.ok(entity).build();
   }
+
+  protected Response create(T entity, String recordFQN) {
+    entity = repository.createNewRecord(entity, recordFQN);
+    return Response.ok(entity).build();
+  }
 }

openmetadata-service/src/main/java/org/openmetadata/service/resources/dqtests/TestCaseResolutionStatusResource.java

@@ -238,7 +238,7 @@ public Response create(
             createTestCaseResolutionStatus,
             securityContext.getUserPrincipal().getName());
 
-    return create(testCaseResolutionStatus, null, testCaseEntity.getFullyQualifiedName());
+    return create(testCaseResolutionStatus, testCaseEntity.getFullyQualifiedName());
   }
 
   @PATCH

openmetadata-service/src/test/java/org/openmetadata/service/resources/dqtests/TestCaseResourceTest.java

@@ -1115,7 +1115,7 @@ void get_testCaseResultWithIncidentId(TestInfo test)
 
     // We can get it via API with a list of ongoing incidents
     TestCase result = getTestCase(testCaseEntity.getFullyQualifiedName(), ADMIN_AUTH_HEADERS);
-
+    UUID incidentId = result.getIncidentId();
     assertNotNull(result.getIncidentId());
 
     // Resolving the status triggers resolving the task, which triggers removing the ongoing
@@ -1131,10 +1131,22 @@ void get_testCaseResultWithIncidentId(TestInfo test)
                     .withResolvedBy(USER1_REF));
     createTestCaseFailureStatus(createResolvedStatus);
 
-    // If we read again, the incident list will be empty
     result = getTestCase(testCaseEntity.getFullyQualifiedName(), ADMIN_AUTH_HEADERS);
+    assertNotNull(result.getIncidentId());
+    assertEquals(incidentId, result.getIncidentId());
 
-    assertNull(result.getIncidentId());
+    // Add a new failed result, which will create a NEW incident and start a new stateId
+    putTestCaseResult(
+        testCaseEntity.getFullyQualifiedName(),
+        new TestCaseResult()
+            .withResult("result")
+            .withTestCaseStatus(TestCaseStatus.Failed)
+            .withTimestamp(TestUtils.dateToTimestamp("2024-01-02")),
+        ADMIN_AUTH_HEADERS);
+
+    result = getTestCase(testCaseEntity.getFullyQualifiedName(), ADMIN_AUTH_HEADERS);
+    assertNotNull(result.getIncidentId());
+    assertNotEquals(incidentId, result.getIncidentId());
   }
 
   @Test