Reindex robustness: selective fields, cache fail-fast, stop actually stops

[harshach]

fixes https://github.com/open-metadata/openmetadata-collate/issues/3948
fixes https://github.com/open-metadata/openmetadata-collate/issues/3947
fixes https://github.com/open-metadata/openmetadata-collate/issues/3949

Describe your changes:

Three independent fixes that all surfaced from the same incident — a 580k-container reindex that froze for hours, then refused to actually stop when the user clicked Stop. Together they remove the cache layer as a single point of fragility for indexing, restore the selective-field optimisation in the distributed pipeline, and make Stop reflect in the UI within seconds instead of minutes.

Selective fields in the distributed reader path. `PartitionWorker` was hardcoding `List.of("*")`, triggering every fieldFetcher in `setFieldsInBulk` — including `fetchAndSetOwns` on Team/User where every owned entity becomes a `getEntityReferenceById` round-trip. PR #27723 fixed this for `EntityReader` (single-server) but the distributed path never picked it up. Lifted the resolution into `ReindexingUtil.getSearchIndexFields()` so both paths share one source of truth.

Cache layer no longer flaps on a single Redis hiccup. `RedisCacheProvider` used to flip the whole provider unavailable on the first 300 ms timeout and flip back on the next PING success. Combined with a 1 s health-check this made the indexer pay one timeout per cycle indefinitely. Replaced with a sliding-window failure detector (5 failures in 30 s to trip, 3 consecutive successes to recover), modelled on `BulkCircuitBreaker`.

`CacheWarmupApp` parses its own schema. It used to parse user config as `EventPublisherJob` (the SearchIndex schema), which broke the Configuration page once `cacheWarmupAppConfig.json` gained a `type` discriminator. Switched to `CacheWarmupAppConfig` in all four parse sites and decoupled runtime status/stats from the parsed config; removed the `readAppConfigFlags()` workaround. Now reports `ACTIVE_ERROR` (not `COMPLETED`) when an entity type bails, and retries on transient cache unavailability instead of aborting on the first miss.

Stop actually stops. Three pieces: `DistributedJobStatsAggregator` skips the WebSocket status broadcast while `STOPPING` so it doesn't overwrite the `AppRunRecord.STOPPED` that `AppScheduler.updateAndBroadcastStoppedStatus` already pushed, and self-stops after a 30 s grace if the executor never gets to call `stop()`. `DistributedSearchIndexExecutor.stop()` now calls `workerExecutor.shutdownNow()` after flagging workers, so threads parked inside the bulk-sink semaphore, `initializeKeysetCursor`, or `waitForSinkOperations` (5-min deadline) get interrupted instead of grinding for minutes. `OpenSearchBulkSink` replaces `concurrentRequestSemaphore.acquire()` with a 60 s `tryAcquire`, recording permanent failure on timeout — a leaked async future can no longer permanently freeze every subsequent flush at a fixed record count.

Tested via 8 new regression tests covering schema-parse, aggregator STOPPING-skip and self-stop, executor `shutdownNow`. All 109 directly relevant tests pass (PartitionWorker, Aggregator, Executor, BulkSink), plus the broader 677-test reindex/cache suite stays green.

Type of change:

• Bug fix
• Improvement

Checklist:

• I have read the CONTRIBUTING document.
• I have commented on my code, particularly in hard-to-understand areas.
• I have added tests around the new logic.

🤖 Generated with Claude Code

---

Summary by Gitar

• Performance test tooling:
  • Added scripts/ingest_100k_containers.py to generate high-volume container entities for stress testing indexing pipelines.
  • Added scripts/reindex-perf-bootstrap.sh to automate stack bring-up, ingestion-bot token retrieval, and data population.
• Reindexing robustness:
  • Expanded ReindexingUtilTest coverage to include AI/LLM/MCP entities, Data Assets, and service types to ensure parity in field filtering.

_{This will update automatically on new commits.}

[Copilot]

Pull request overview

This PR aims to harden reindexing and cache-warmup behavior after a large reindex incident by reducing unnecessary field loading, making Redis cache availability less brittle, and improving stop responsiveness in the distributed reindex flow.

Changes:

• Centralizes selective reindex field resolution and applies it to the distributed PartitionWorker path.
• Adds sliding-window health detection for RedisCacheProvider, cache-warmup config parsing fixes, and partial-warmup handling in CacheWarmupApp.
• Improves distributed stop behavior via aggregator STOPPING handling, executor shutdownNow(), and bounded semaphore acquisition in OpenSearchBulkSink.

Reviewed changes

Copilot reviewed 11 out of 11 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/searchIndex/distributed/DistributedSearchIndexExecutorTest.java	Adds regression tests for executor stop/shutdown behavior.
openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/searchIndex/distributed/DistributedJobStatsAggregatorTest.java	Adds tests for STOPPING broadcast suppression and aggregator self-stop behavior.
openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/cache/CacheWarmupAppConfigParseTest.java	Adds config parsing regression tests for cache warmup app schema.
openmetadata-service/src/main/java/org/openmetadata/service/workflows/searchIndex/ReindexingUtil.java	Introduces shared helper for resolving reindex field lists.
openmetadata-service/src/main/java/org/openmetadata/service/cache/RedisCacheProvider.java	Replaces fail-fast availability toggling with sliding-window failure/recovery tracking.
openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/distributed/PartitionWorker.java	Switches distributed reader path to selective field resolution.
openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/distributed/DistributedSearchIndexExecutor.java	Forces worker interruption on stop via shutdownNow().
openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/distributed/DistributedJobStatsAggregator.java	Skips STOPPING websocket broadcasts and self-stops after a grace period.
openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/OpenSearchBulkSink.java	Bounds semaphore acquisition time to avoid indefinite bulk flush hangs.
openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/EntityReader.java	Reuses the new shared field-resolution helper.
openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/cache/CacheWarmupApp.java	Parses typed cache warmup config, retries on cache unavailability, and reports partial warmups as errors.

[Copilot]

Pull request overview

Copilot reviewed 12 out of 12 changed files in this pull request and generated 3 comments.

[Copilot]

Pull request overview

Copilot reviewed 22 out of 22 changed files in this pull request and generated 6 comments.

[Copilot]

Pull request overview

Copilot reviewed 26 out of 26 changed files in this pull request and generated 8 comments.

[Copilot]

Pull request overview

Copilot reviewed 29 out of 29 changed files in this pull request and generated 5 comments.

Comments suppressed due to low confidence (1)

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/cache/CacheWarmupApp.java:180

• This path now marks the run FAILED but never populates jobData.failure, so the AppRunRecord is persisted without a failureContext. Operators will see a terminal failure with no indication that the warmup aborted because Redis was unavailable.

      if (cacheProvider == null || !cacheProvider.available()) {
        // Surface this as FAILED — initJobData set status to RUNNING above, and the finally block
        // will broadcast the terminal state. Leaving it RUNNING here would pin the job record in
        // an active state indefinitely.
        LOG.warn("Cache not available, skipping warmup");
        jobData.setStatus(EventPublisherJob.Status.FAILED);

[gitar-bot]

Code Review ✅ Approved 7 resolved / 7 findings

Improves reindexing robustness by implementing selective field filtering, a sliding-window failure detector for Redis, and forceful executor termination for stop operations. All identified issues, including race conditions and configuration parsing errors, have been addressed and verified with comprehensive regression tests.

✅ 7 resolved

✅ Bug: pruneOldFailures early-break skips out-of-order timestamps

📄 openmetadata-service/src/main/java/org/openmetadata/service/cache/RedisCacheProvider.java:142-152
In pruneOldFailures, the iterator breaks as soon as it encounters a timestamp >= cutoff, assuming chronological order. Since recordFailure calls System.currentTimeMillis() and then addLast without synchronization, two concurrent threads can produce out-of-order entries (e.g., [..., 100, 99]). The early break would then leave stale entries behind.

In practice this is bounded (failure rate is low, window is 30s, and health-checks call recordSuccess → pruneOldFailures every second), so it won't grow unboundedly. But the logic is subtly incorrect. Consider removing the break and always iterating the full deque, or switching to a time-ordered structure.

✅ Bug: recordSuccess/recordFailure race can cause premature recovery

📄 openmetadata-service/src/main/java/org/openmetadata/service/cache/RedisCacheProvider.java:104-117 📄 openmetadata-service/src/main/java/org/openmetadata/service/cache/RedisCacheProvider.java:127-140
The recordSuccess / recordFailure pair performs non-atomic check-then-act on shared state (available, consecutiveSuccesses, failureTimestamps). A plausible interleaving:

1. Thread A (success): reads available=false, increments consecutiveSuccesses to 3
2. Thread B (failure): resets consecutiveSuccesses to 0, adds failure timestamp
3. Thread A: sees count=3 >= RECOVERY_THRESHOLD, sets available=true

This flips the provider back to available despite a concurrent failure, partially defeating the sliding-window detector. In practice the health-checker is single-threaded and real-op failures are rare simultaneously with pings, so the window is narrow. Still, synchronizing recordSuccess and recordFailure (or using a small synchronized block around the state transition) would make the contract solid.

✅ Quality: CacheWarmupAppConfigParseTest test asserts nothing meaningful

📄 openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/cache/CacheWarmupAppConfigParseTest.java:70-84
rejectsUnknownFields (line 71-88) catches any throwable and then asserts t == null || t.getMessage() != null, which is trivially true in all cases. The comment says "both behaviours are acceptable" — but a test that always passes regardless of outcome provides no regression guard. Either assert that the unknown field is rejected (if that's the intended schema contract) or remove the test.

✅ Quality: CachedEntityDao.put* methods lack bypass guards (inconsistent)

📄 openmetadata-service/src/main/java/org/openmetadata/service/cache/CachedEntityDao.java:83-97
CachedRelationshipDao and CachedTagUsageDao guard both their get and put methods with EntityCacheBypass.isSkipped(), but CachedEntityDao's putBase, putByName, putReference, and putReferenceByName do not. While these put methods don't appear to be called from reindex read paths today, the inconsistency means a future caller could inadvertently write to Redis during a bypassed reindex. For symmetry and defensive correctness, consider adding the same guard.

✅ Bug: semaphoreAcquireTimeoutSeconds should be volatile

📄 openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/searchIndex/OpenSearchBulkSink.java:909
The new semaphoreAcquireTimeoutSeconds field (plain long) is written by setSemaphoreAcquireTimeoutSecondsForTesting and read by flushInternal, which can execute on different threads (the scheduler thread vs. the thread that sets up the processor). Without volatile, there is no happens-before edge guaranteeing the updated value is visible to the reading thread. In practice the current test writes before invoking flushInternal on the same thread so it works today, but the missing volatile is a latent data-race that would bite if the test or any future caller ever sets it while flushing is active.

...and 2 more resolved from earlier reviews

Options

Display: compact → Showing less information.

Comment with these commands to change:

Compact

gitar display:verbose

_{Was this helpful? React with 👍 / 👎 | Gitar}

[sonarqubecloud]

Quality Gate passed for 'open-metadata-ingestion'

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud