[test] Fix 30+ flaky tests across multiple modules

Test-only changes to fix flaky tests. No production behavior changes.

Root cause categories fixed:
- AsyncGauge shared executor: use dedicated MetricsRepositoryUtils executor
- Missing volatile on cross-thread test fields
- Utils.sleep() before assertions: replaced with
waitForNonDeterministicAssertion
- replicationFactor=2 with few servers: reduced to RF=1 where replication
isn't tested
- Version not waited in all regions: wait in ALL DCs before proceeding
- Static DaVinci singleton leak across data provider iterations: @AfterMethod
cleanup
- Fragile exception message assertions: assert on type not message text
- Accumulate poll results instead of expecting all in single poll
- Try-finally for test resource cleanup (executor/MetricsRepository)
- Topic cleanup interval too aggressive for multi-region tests
- Router not yet discovered store version: wrap reads in
waitForNonDeterministicAssertion
- Mockito inline mock maker for final gRPC classes

Author: sushantmane

AGENTS.md

@@ -0,0 +1,47 @@
+# Repository Guidelines
+
+## Project Structure & Module Organization
+
+Venice is a multi-module Gradle build. Runtime services sit in `services/*/src`, while client libraries live under
+`clients/*/src`. Shared logic and generated schemas are centralized in `internal/` and `specs/`. Integration artefacts
+(Docker images, remote runners, connectors) are kept in `integrations/`, `docker/`, and `scripts/`. Tests are co-located
+with code, and end-to-end scenarios live in `tests/venice-pulsar-test`. Consult `docs/` for design notes before altering
+cross-module contracts.
+
+## Build, Test, and Development Commands
+
+- `./gradlew clean build` – compile every module, run unit/integration tests, and assemble distributables (server,
+  router, controller).
+- `./gradlew test` – execute module unit suites; use `-PskipIntegrationTests=true` when iterating quickly.
+- `./gradlew :services:venice-server:installDist` (or similar) – produce runnable binaries for targeted modules.
+- `./gradlew spotlessApply` – auto-format Java sources per Eclipse config; the pre-commit hook calls this.
+- `./gradlew spotbugsMain` & `./gradlew check` – static analysis plus aggregated quality gate.
+
+## Coding Style & Naming Conventions
+
+Java 8+ code formatted via Spotless using `gradle/spotless/eclipse-java-formatter.xml`. Do not hand-tune indentation;
+run Spotless instead. Adopt `UpperCamelCase` for classes, `lowerCamelCase` for fields/methods, and `SNAKE_CASE` only for
+constants. Favor `Optional` over `null` checks in new code and keep logging through `VeniceLogger`. Shared config
+objects live in `Venice*Config` classes; follow existing naming to keep discovery predictable.
+
+## Testing Guidelines
+
+Unit tests use TestNG plus Mockito; place them in the matching module under `src/test/java`. Name classes `*Test` or
+`*IntegrationTest` so Gradle discovers them. Each module defines a `jacocoCoverageThreshold`; add or extend tests to
+keep coverage above those guards. For pulsar/kafka end-to-end flows, reuse the harness in `tests/venice-pulsar-test` and
+run via `./gradlew integrationTest -PtestProfiles=pulsar`.
+
+## Commit & Pull Request Guidelines
+
+Follow the history pattern: `[scope] Concise summary (#issue)` (for example, `[server] Guard admin topic switch`). Group
+related edits into single commits, run `./gradlew clean build` locally, and let the pre-commit hook finish before
+pushing. Pull requests should include: purpose and rollout notes, validation evidence (test command output or
+screenshots for UI changes), and links to tracking tickets. Flag breaking config changes and update `docs/` or `specs/`
+when altering public schemas or protocol fields.
+
+## Environment & Tooling Tips
+
+Run `./gradlew setupWorkspace` once to install repo git hooks and blame ignore rules. Local override configurations
+belong in `~/.gradle/gradle.properties`; never commit credentials—use the sample placeholders in `consumer.config` and
+`run-remote-cp.sh`. When cutting releases, coordinate with `make_tag.py` and document the tag in the internal release
+tracker.

clients/da-vinci-client/src/main/java/com/linkedin/davinci/client/AvroGenericDaVinciClient.java

@@ -896,6 +896,18 @@ public static DaVinciBackend getBackend() {
     }
   }
 
+  /**
+   * Nulls out the static DaVinci backend singleton for test cleanup. When ThreadTimeoutException
+   * interrupts a test, the singleton may leak with stale state (e.g., cache config) that poisons
+   * subsequent tests. The leaked backend's resources are cleaned up by test framework teardown.
+   */
+  // Visible for testing
+  public static void resetDaVinciBackendForTests() {
+    synchronized (AvroGenericDaVinciClient.class) {
+      daVinciBackend = null;
+    }
+  }
+
   @Override
   public synchronized void start() {
     if (isReady()) {

clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerDaVinciRecordTransformerImplTest.java

@@ -474,7 +474,7 @@ public void testMaxBufferSize() throws NoSuchFieldException, IllegalAccessExcept
     }
 
     TestUtils.waitForNonDeterministicAssertion(10, TimeUnit.SECONDS, true, () -> {
-      verify(bufferIsFullCondition).signal();
+      verify(bufferIsFullCondition, atLeastOnce()).signal();
     });
   }
 

clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/KafkaStoreIngestionServiceTest.java

@@ -74,6 +74,7 @@
 import com.linkedin.venice.utils.TestUtils;
 import com.linkedin.venice.utils.VeniceProperties;
 import com.linkedin.venice.utils.locks.ResourceAutoClosableLockManager;
+import com.linkedin.venice.utils.metrics.MetricsRepositoryUtils;
 import io.tehuti.metrics.MetricConfig;
 import io.tehuti.metrics.MetricsRepository;
 import io.tehuti.metrics.stats.AsyncGauge;
@@ -147,7 +148,7 @@ public void setUp() {
         mockMetadataRepo,
         mockSchemaRepo,
         mockLiveClusterConfigRepo,
-        new MetricsRepository(),
+        MetricsRepositoryUtils.createSingleThreadedMetricsRepository(),
         Optional.empty(),
         Optional.empty(),
         AvroProtocolDefinition.PARTITION_STATE.getSerializer(),
@@ -469,7 +470,7 @@ public void testStoreIngestionTaskShutdownLastPartition() {
         mockMetadataRepo,
         mockSchemaRepo,
         mockLiveClusterConfigRepo,
-        new MetricsRepository(),
+        MetricsRepositoryUtils.createSingleThreadedMetricsRepository(),
         Optional.empty(),
         Optional.empty(),
         AvroProtocolDefinition.PARTITION_STATE.getSerializer(),
@@ -793,7 +794,7 @@ public void testInitParticipantConsumptionTask() {
         mockServerConfig,
         Optional.of(mockClientConfig),
         mockClusterInfoProvider,
-        new MetricsRepository(),
+        MetricsRepositoryUtils.createSingleThreadedMetricsRepository(),
         mockIcProvider);
     assertNotNull(
         pct,
@@ -805,7 +806,7 @@ public void testInitParticipantConsumptionTask() {
         mockServerConfig,
         Optional.of(mockClientConfig),
         mockClusterInfoProvider,
-        new MetricsRepository(),
+        MetricsRepositoryUtils.createSingleThreadedMetricsRepository(),
         mockIcProvider);
     assertNull(
         pct,
@@ -817,7 +818,7 @@ public void testInitParticipantConsumptionTask() {
         mockServerConfig,
         Optional.empty(),
         mockClusterInfoProvider,
-        new MetricsRepository(),
+        MetricsRepositoryUtils.createSingleThreadedMetricsRepository(),
         mockIcProvider);
     assertNull(pct, "Participant consumption task should not be initialized when client config is not present");
   }

clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/ParticipantStoreConsumptionTaskTest.java

@@ -3,6 +3,7 @@
 import static com.linkedin.venice.stats.OpenTelemetryMetricsSetup.UNKNOWN_STORE_NAME;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyDouble;
+import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
@@ -20,10 +21,12 @@
 import com.linkedin.venice.participant.protocol.ParticipantMessageValue;
 import com.linkedin.venice.participant.protocol.enums.ParticipantMessageType;
 import com.linkedin.venice.utils.SleepStallingMockTime;
+import com.linkedin.venice.utils.TestUtils;
 import com.linkedin.venice.utils.Utils;
 import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 import org.testng.annotations.Test;
 
@@ -215,11 +218,15 @@ public void testOuterCatchRecordsFailedConsumptionWithUnknownStore() {
     // Wait for the task to start and reach the sleep stall (heartbeat fires before sleep)
     verify(stats, timeout(WAIT).times(1)).recordHeartbeat();
 
-    // Now advance time to unblock the sleep — the task will call getIngestingTopicsWithVersionStatusNotOnline
-    // which throws, hitting the outer catch that records with UNKNOWN_STORE_NAME
-    time.advanceTime(1);
-
-    verify(stats, timeout(WAIT).times(1)).recordKillPushJobFailedConsumption(UNKNOWN_STORE_NAME);
+    // Advance time to unblock the sleep — the task will call getIngestingTopicsWithVersionStatusNotOnline
+    // which throws, hitting the outer catch that records with UNKNOWN_STORE_NAME.
+    // Use waitForNonDeterministicAssertion because there is a small race window: if advanceTime(1)
+    // is called before the task thread enters sleep(), the time advancement is consumed by the next
+    // sleep call instead. Retrying advances time again to guarantee the task progresses.
+    TestUtils.waitForNonDeterministicAssertion(5, TimeUnit.SECONDS, true, () -> {
+      time.advanceTime(1);
+      verify(stats, atLeastOnce()).recordKillPushJobFailedConsumption(UNKNOWN_STORE_NAME);
+    });
 
     // Close the task before the negative assertions so no further loop iterations can fire.
     task.close();