fix(search): treat stale parentOf as warning during time-series reindex (open-metadata#27417) (open-metadata#27800)

* fix(search): treat stale parentOf as warning during time-series reindex (open-metadata#27417)

Time-series records (testCaseResolutionStatus, testCaseResult, ...) whose
parentOf entity_relationship row is missing surface as
"does not have expected relationship parentOf to/from entity type ..."
from EntityRepository.ensureSingleRelationship and were failing the entire
reindex batch. Mirror the warning-vs-failure split already used for
EntityInterface sources: extract isEntityNotFoundError into ReindexingUtil,
broaden it to match the relationship-not-found message, and apply the same
partitioning to PaginatedEntityTimeSeriesSource.read/readWithCursor/
readNextKeyset so orphaned rows are counted as warnings via
result.getWarningsCount() instead of failing the job.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* review: tighten matcher, propagate warnings on cursor/keyset paths

Address PR open-metadata#27800 review:
- Drop bare "not found" from isEntityNotFoundError; add "entity not found"
  so we still match EntityNotFoundException's "Entity not found:" form but
  no longer misclassify "Column 'foo' not found in result set" or
  "SSL certificate not found" as warnings.
- Call updateStats(success, failed, warnings) in readWithCursor and
  readNextKeyset so the source's StepStats.warningRecords reflects warnings
  for cursor- and keyset-based reads (was already correct in read()).
- partitionErrors: requireNonNull(warningsOut) and document the contract.
- Tests: cover the new bare-"not found" exclusion, the "Entity not found:"
  inclusion, the readWithCursor stats propagation, and the null-warningsOut
  guard.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* Show warn for failing test case resolution

* Always recreate

* review: rename matcher + record reader failures before early-return

Address two Copilot comments on PR open-metadata#27800:

1. PartitionWorker.processBatch was returning early when readSuccessCount==0
   without invoking failureRecorder.recordReaderEntityFailure for the per-row
   errors in that batch — losing entity-level failure diagnostics for
   "all-error" batches. Extract the failure-recording into a helper and call
   it before the early-return so it runs whether or not the batch contained
   any successful rows.

2. Rename ReindexingUtil.isEntityNotFoundError -> isStaleReferenceError to
   reflect that the matcher now also catches the relationship-not-found
   message ("does not have expected relationship ...") raised by
   EntityRepository.ensureSingleRelationship, not just plain entity-not-found.
   Update PaginatedEntitiesSource and ReindexingUtilStaleRelationshipTest
   call sites. EntityRepository has its own private isEntityNotFoundError
   helper that is unrelated and unchanged.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fix: import java.util.Objects instead of fully qualified name in ReindexingUtil

Agent-Logs-Url: https://github.com/open-metadata/OpenMetadata/sessions/e6d9ce67-1181-4e61-80f4-e7aba664dfe7

Co-authored-by: mohityadav766 <105265192+mohityadav766@users.noreply.github.com>

* Fill End time

* Not * reads

* Fix Fields issue

* Add tags back

* Fix failing test

* Revert "Always recreate"

This reverts commit 23578b0.

* review: cover all EntityNotFoundException formats; quiet null-id WARN

Address two review concerns on PR open-metadata#27800:

1. The stale-reference matcher missed EntityNotFoundException's primary
   factory messages — byId ("Entity with id [...] not found."), byName
   ("Entity with name [...] not found."), byVersion ("Entity with id
   [...] and version [...] not found."), and byParserSchema ("Parser
   schema not found ..."). These would have been classified as real
   failures instead of warnings. Add specific contains() patterns
   ("entity with id", "entity with name", "parser schema not found")
   that match each factory's exact prefix without reintroducing the
   over-broad bare "not found" check. byFilter ("Entity not found for
   query params [...]") was already covered by "entity not found".

2. PartitionWorker.recordReaderFailures emitted WARN per error when
   entityId was null. EntityTimeSeriesRepository builds EntityError
   with only a message (no entity reference), so every time-series
   error hit that branch — log spam under load. Downgrade to DEBUG
   with a comment explaining why the id is absent.

Tests: ReindexingUtilStaleRelationshipTest now exercises every
EntityNotFoundException factory message.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* review: NPE guard, locale-stable matcher, less log spam, accurate Javadoc

Address remaining Copilot comments on PR open-metadata#27800:

1. EntityTimeSeriesRepository.getResultList(...) returns ResultList with
   errors=null on the success path. PaginatedEntityTimeSeriesSource was
   reading result.getErrors().size() / .isEmpty() right after, which would
   NPE on the common no-error path. Normalize errors to an empty list
   inside filterStaleRelationshipErrors so callers can rely on non-null.

2. ReindexingUtil.isStaleReferenceError now lowercases with Locale.ROOT
   instead of the platform default — avoids Turkish-locale style edge
   cases where 'I' lowercases differently and substring matches fail.

3. PaginatedEntityTimeSeriesSource.read() was logging every real reader
   error message at WARN. For large failed batches this floods logs.
   Switch to a single WARN with the error count plus the first 5 message
   details at DEBUG (gated by isDebugEnabled).

4. OmAppJobListener.fillTerminalTimings Javadoc claimed "does nothing if
   endTime is already set" but the body still backfills executionTime in
   that case. Rewrite to accurately describe the per-field idempotency.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* Relaign Changes

* Add Exception handling

* fix(reindex): degrade to required-set, not '*', when filter step throws

ReindexingUtil.getSearchIndexFields was catching every exception in the
filter path and returning ["*"]. That defeats the entire point of
selective fields — sending all fields silently masks any drift between
SearchIndex.COMMON_REINDEX_FIELDS and the entity's schema, instead of
surfacing it at the PaginatedEntitiesSource boundary.

Split the try/catch so:
- getReindexFieldsFor() throwing → ["*"] (we have no required set to
  fall back to; pre-selective behavior).
- getOnlySupportedFields() throwing (typically because the
  EntityRepository isn't registered yet — boot/test scenarios) → return
  the unfiltered required set. PaginatedEntitiesSource validates the
  fields when an actual entity flows through, so any real drift surfaces
  loudly rather than being silently swallowed by a "*" wildcard.

Restores the assertion and intent of
ReindexingUtilTest.unregisteredRepositoryReturnsRequiredUnfiltered, and
fixes the parametrized parity tests that were also seeing "*" because of
this regression. Also stubs repo.getOnlySupportedFields(...) in the test
mocks so it returns a real EntityUtil.Fields built against the declared
allowedFields — matches the production code path.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* Add Only Supported field at other call sites

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>

Author: 3 people

openmetadata-service/src/main/java/org/openmetadata/service/Entity.java

@@ -563,6 +563,11 @@ public static Fields getFields(String entityType, List<String> fields) {
     return entityRepository.getFields(String.join(",", fields));
   }
 
+  public static Fields getOnlySupportedFields(String entityType, List<String> fields) {
+    EntityRepository<?> entityRepository = Entity.getEntityRepository(entityType);
+    return entityRepository.getOnlySupportedFields(String.join(",", fields));
+  }
+
   public static <T> T getEntity(EntityReference ref, String fields, Include include) {
     if (ref == null) {
       return null;

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/EntityReader.java

@@ -3,6 +3,7 @@
 import static org.openmetadata.service.Entity.QUERY_COST_RECORD;
 import static org.openmetadata.service.Entity.TEST_CASE_RESOLUTION_STATUS;
 import static org.openmetadata.service.Entity.TEST_CASE_RESULT;
+import static org.openmetadata.service.workflows.searchIndex.ReindexingUtil.getSearchIndexFields;
 
 import java.util.ArrayList;
 import java.util.List;

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/ReindexingOrchestrator.java

@@ -24,6 +24,7 @@
 import org.openmetadata.service.Entity;
 import org.openmetadata.service.apps.bundles.searchIndex.listeners.LoggingProgressListener;
 import org.openmetadata.service.apps.bundles.searchIndex.listeners.SlackProgressListener;
+import org.openmetadata.service.apps.scheduler.OmAppJobListener;
 import org.openmetadata.service.jdbi3.CollectionDAO;
 import org.openmetadata.service.jdbi3.SystemRepository;
 import org.openmetadata.service.search.SearchRepository;
@@ -109,7 +110,7 @@ public void stop() {
 
     AppRunRecord appRecord = context.getJobRecord();
     appRecord.setStatus(AppRunRecord.Status.STOPPED);
-    appRecord.setEndTime(System.currentTimeMillis());
+    OmAppJobListener.fillTerminalTimings(appRecord);
     context.storeRunRecord(JsonUtils.pojoToJson(appRecord));
     context.pushStatusUpdate(appRecord, true);
     sendUpdates();
@@ -368,6 +369,7 @@ private void finalizeJobExecution() {
     if (stopped) {
       AppRunRecord appRecord = context.getJobRecord();
       appRecord.setStatus(AppRunRecord.Status.STOPPED);
+      OmAppJobListener.fillTerminalTimings(appRecord);
       context.storeRunRecord(JsonUtils.pojoToJson(appRecord));
     }
   }
@@ -383,6 +385,7 @@ private void sendUpdates() {
   private void updateRecordToDbAndNotify() {
     AppRunRecord appRecord = context.getJobRecord();
     appRecord.setStatus(AppRunRecord.Status.fromValue(jobData.getStatus().value()));
+    OmAppJobListener.fillTerminalTimings(appRecord);
 
     if (jobData.getFailure() != null) {
       appRecord.setFailureContext(

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/distributed/DistributedJobStatsAggregator.java

@@ -35,6 +35,7 @@
 import org.openmetadata.service.apps.bundles.searchIndex.BulkSink;
 import org.openmetadata.service.apps.bundles.searchIndex.ReindexingJobContext;
 import org.openmetadata.service.apps.bundles.searchIndex.ReindexingProgressListener;
+import org.openmetadata.service.apps.scheduler.OmAppJobListener;
 import org.openmetadata.service.jdbi3.CollectionDAO;
 import org.openmetadata.service.socket.WebSocketManager;
 
@@ -562,6 +563,7 @@ private AppRunRecord convertToAppRunRecord(
     appRecord.setStartTime(appStartTime != null ? appStartTime : job.getStartedAt());
     appRecord.setEndTime(job.getCompletedAt());
     appRecord.setTimestamp(job.getUpdatedAt());
+    OmAppJobListener.fillTerminalTimings(appRecord);
 
     // Add stats as success context
     SuccessContext successContext = new SuccessContext();

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/distributed/PartitionWorker.java

@@ -506,15 +506,16 @@ private BatchResult processBatch(
     ResultList<?> resultList = readEntitiesKeyset(entityType, keysetCursor, batchSize);
     long readDurationNanos = System.nanoTime() - readStartNanos;
 
-    if (resultList == null || resultList.getData() == null || resultList.getData().isEmpty()) {
-      LOG.debug("{} read={}ms returned empty", entityType, readDurationNanos / 1_000_000L);
-      return new BatchResult(0, 0, 0, null);
-    }
-
-    String nextCursor = resultList.getPaging() != null ? resultList.getPaging().getAfter() : null;
-    int readSuccessCount = listOrEmpty(resultList.getData()).size();
-    int readErrorCount = listOrEmpty(resultList.getErrors()).size();
-    int warningsCount = resultList.getWarningsCount() != null ? resultList.getWarningsCount() : 0;
+    int readSuccessCount = resultList != null ? listOrEmpty(resultList.getData()).size() : 0;
+    int readErrorCount = resultList != null ? listOrEmpty(resultList.getErrors()).size() : 0;
+    int warningsCount =
+        (resultList != null && resultList.getWarningsCount() != null)
+            ? resultList.getWarningsCount()
+            : 0;
+    String nextCursor =
+        (resultList != null && resultList.getPaging() != null)
+            ? resultList.getPaging().getAfter()
+            : null;
 
     if (statsTracker != null) {
       // Reader timing = wall-clock time of the keyset DB read (listAfter + setFieldsInBulk
@@ -523,28 +524,16 @@ private BatchResult processBatch(
           readSuccessCount, readErrorCount, warningsCount, readDurationNanos);
     }
 
-    if (failureRecorder != null && readErrorCount > 0) {
-      for (EntityError entityError : listOrEmpty(resultList.getErrors())) {
-        Object rawEntity = entityError.getEntity();
-        String entityId = null;
-        if (rawEntity instanceof EntityInterface) {
-          UUID id = ((EntityInterface) rawEntity).getId();
-          if (id != null) {
-            entityId = id.toString();
-          }
-        } else if (rawEntity != null) {
-          entityId = rawEntity.toString();
-        }
-        if (entityId == null) {
-          LOG.warn(
-              "Skipping reader failure record for entityType={}: entityId is null, message={}",
-              entityType,
-              entityError.getMessage());
-          continue;
-        }
-        failureRecorder.recordReaderEntityFailure(
-            entityType, entityId, null, entityError.getMessage());
-      }
+    recordReaderFailures(entityType, resultList, readErrorCount);
+
+    if (readSuccessCount == 0) {
+      LOG.debug(
+          "{} read={}ms returned no indexable rows (warnings={}, errors={})",
+          entityType,
+          readDurationNanos / 1_000_000L,
+          warningsCount,
+          readErrorCount);
+      return new BatchResult(0, readErrorCount, warningsCount, nextCursor);
     }
 
     Map<String, Object> contextData = createContextData(entityType, statsTracker);
@@ -572,6 +561,44 @@ private BatchResult processBatch(
     }
   }
 
+  /**
+   * Persist per-entity reader failures so that downstream tooling (e.g. the failures dashboard)
+   * can show which specific records the reader could not hydrate. Runs whether or not the batch
+   * has any successful rows — losing failure diagnostics for "all-error" batches would defeat
+   * the point of the recorder.
+   */
+  private void recordReaderFailures(
+      String entityType, ResultList<?> resultList, int readErrorCount) {
+    if (failureRecorder == null || readErrorCount == 0 || resultList == null) {
+      return;
+    }
+    for (EntityError entityError : listOrEmpty(resultList.getErrors())) {
+      Object rawEntity = entityError.getEntity();
+      String entityId = null;
+      if (rawEntity instanceof EntityInterface) {
+        UUID id = ((EntityInterface) rawEntity).getId();
+        if (id != null) {
+          entityId = id.toString();
+        }
+      } else if (rawEntity != null) {
+        entityId = rawEntity.toString();
+      }
+      if (entityId == null) {
+        // Time-series readers (EntityTimeSeriesRepository) build EntityError without an id —
+        // they only have access to the JSON row, not the entity reference. Per-entity recording
+        // requires an id, so log at DEBUG (not WARN) to avoid spamming logs for every error in
+        // large time-series batches.
+        LOG.debug(
+            "No entityId on reader failure for entityType={} — skipping per-entity record. message={}",
+            entityType,
+            entityError.getMessage());
+        continue;
+      }
+      failureRecorder.recordReaderEntityFailure(
+          entityType, entityId, null, entityError.getMessage());
+    }
+  }
+
   /**
    * Read entities from the database.
    *

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/listeners/QuartzProgressListener.java

@@ -23,6 +23,7 @@
 import org.openmetadata.service.apps.bundles.searchIndex.ReindexingJobContext;
 import org.openmetadata.service.apps.bundles.searchIndex.ReindexingProgressListener;
 import org.openmetadata.service.apps.bundles.searchIndex.distributed.DistributedJobContext;
+import org.openmetadata.service.apps.scheduler.OmAppJobListener;
 import org.openmetadata.service.socket.WebSocketManager;
 import org.quartz.JobExecutionContext;
 
@@ -228,6 +229,7 @@ private void broadcastViaWebSocket(AppRunRecord appRecord) {
   private AppRunRecord getUpdatedAppRunRecord() {
     AppRunRecord appRecord = readExistingRecord();
     appRecord.setStatus(AppRunRecord.Status.fromValue(jobData.getStatus().value()));
+    OmAppJobListener.fillTerminalTimings(appRecord);
 
     if (jobData.getStats() != null) {
       SuccessContext ctx = appRecord.getSuccessContext();

openmetadata-service/src/main/java/org/openmetadata/service/apps/scheduler/AppScheduler.java

@@ -465,7 +465,7 @@ private void updateAndBroadcastStoppedStatus(JobExecutionContext context) {
       if (runRecord != null) {
         // Update status to STOPPED
         runRecord.withStatus(AppRunRecord.Status.STOPPED);
-        runRecord.withEndTime(System.currentTimeMillis());
+        OmAppJobListener.fillTerminalTimings(runRecord);
 
         // Get WebSocket channel name
         String webSocketChannelName =

openmetadata-service/src/main/java/org/openmetadata/service/apps/scheduler/OmAppJobListener.java

@@ -47,6 +47,45 @@ protected OmAppJobListener() {
     this.repository = new AppRepository();
   }
 
+  /**
+   * Populate {@code endTime} and {@code executionTime} on a terminal-state run record. Each field
+   * is filled independently and only if currently null:
+   *
+   * <ul>
+   *   <li>{@code endTime} defaults to {@code System.currentTimeMillis()} if absent.
+   *   <li>{@code executionTime} is computed from {@code endTime - startTime} if absent and both
+   *       endpoints are available — this means callers that pre-populated {@code endTime} (e.g.
+   *       from {@code job.getCompletedAt()}) still get an accurate {@code executionTime}.
+   * </ul>
+   *
+   * <p>The method is a no-op for non-terminal statuses, so it is safe to call from progress
+   * listeners that may persist before {@link #jobWasExecuted} runs. Without this, mid-flight
+   * writes by progress listeners (e.g. {@code QuartzProgressListener} firing {@code onJobFailed})
+   * would persist a terminal status to the DB without timings; if the job dies before {@code
+   * jobWasExecuted} fires, polling consumers would see {@code status=FAILED} with no
+   * {@code endTime} / {@code executionTime}.
+   */
+  public static void fillTerminalTimings(AppRunRecord record) {
+    if (record == null || record.getStatus() == null || !isTerminalStatus(record.getStatus())) {
+      return;
+    }
+    if (record.getEndTime() == null) {
+      record.withEndTime(System.currentTimeMillis());
+    }
+    if (record.getExecutionTime() == null
+        && record.getStartTime() != null
+        && record.getEndTime() != null) {
+      record.setExecutionTime(record.getEndTime() - record.getStartTime());
+    }
+  }
+
+  private static boolean isTerminalStatus(AppRunRecord.Status status) {
+    return switch (status) {
+      case SUCCESS, FAILED, ACTIVE_ERROR, STOPPED, COMPLETED -> true;
+      default -> false;
+    };
+  }
+
   @Override
   public String getName() {
     return JOB_LISTENER_NAME;

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

@@ -6607,6 +6607,13 @@ public final Fields getFields(String fields) {
     return new Fields(allowedFields, fields);
   }
 
+  public final Fields getOnlySupportedFields(String fields) {
+    if ("*".equals(fields)) {
+      return new Fields(allowedFields, String.join(",", allowedFields), true);
+    }
+    return new Fields(allowedFields, fields, true);
+  }
+
   protected final Fields getFields(Set<String> fields) {
     return new Fields(allowedFields, fields);
   }

openmetadata-service/src/main/java/org/openmetadata/service/search/indexes/SearchIndex.java

@@ -67,6 +67,7 @@ public interface SearchIndex {
           "followers",
           "votes",
           "extension",
+          "tags",
           "certification",
           "dataProducts");