-
Notifications
You must be signed in to change notification settings - Fork 86
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[changelog][dvc] Disable TC internal retry in ThinClientMetaStoreBase…
…dRepository (#1235) * [changelog][dvc] Disable TC internal retry in ThinClientMetaStoreBasedRepository 1. The existing TC internal retry implemented in RetriableStoreClient is vulnerable to a dead lock issue due to deserialization thread performing and waiting on the retry. Once the thread pool is exhausted all deserialization threads will be waiting forever because there won't be any threads left to handle the transport response. We will fix this internal retry in another commit. 2. Disable the TC internal retry in ThinClientMetaStoreBasedRepository because we recently added external retry with RetryUtils.executeWithMaxAttempt and timeout. Remove hardcoded retry configs from NativeMetadataRepository since it's no longer needed and it was leaking implementation details. 3. Added staleness metric in NativeMetadataRepository that should help us detect any issue causing the local metadata to be stale (including the dead lock issue mentioned above) for both DVC and change log use cases. * Addressed review comments
- Loading branch information
Showing
9 changed files
with
174 additions
and
37 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
42 changes: 42 additions & 0 deletions
42
...-vinci-client/src/main/java/com/linkedin/davinci/stats/NativeMetadataRepositoryStats.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,42 @@ | ||
| package com.linkedin.davinci.stats; | ||
|
|
||
| import com.linkedin.venice.stats.AbstractVeniceStats; | ||
| import com.linkedin.venice.utils.concurrent.VeniceConcurrentHashMap; | ||
| import io.tehuti.metrics.MetricsRepository; | ||
| import io.tehuti.metrics.Sensor; | ||
| import io.tehuti.metrics.stats.AsyncGauge; | ||
| import java.time.Clock; | ||
| import java.util.Map; | ||
|
|
||
|
|
||
| public class NativeMetadataRepositoryStats extends AbstractVeniceStats { | ||
| private final Sensor storeMetadataStalenessSensor; | ||
| private final Map<String, Long> metadataCacheTimestampMapInMs = new VeniceConcurrentHashMap<>(); | ||
| private final Clock clock; | ||
|
|
||
| public NativeMetadataRepositoryStats(MetricsRepository metricsRepository, String name, Clock clock) { | ||
| super(metricsRepository, name); | ||
| this.clock = clock; | ||
| this.storeMetadataStalenessSensor = registerSensor( | ||
| new AsyncGauge( | ||
| (ignored1, ignored2) -> getMetadataStalenessHighWatermarkMs(), | ||
| "store_metadata_staleness_high_watermark_ms")); | ||
| } | ||
|
|
||
| final double getMetadataStalenessHighWatermarkMs() { | ||
| if (this.metadataCacheTimestampMapInMs.isEmpty()) { | ||
| return Double.NaN; | ||
| } else { | ||
| long oldest = metadataCacheTimestampMapInMs.values().stream().min(Long::compareTo).get(); | ||
| return clock.millis() - oldest; | ||
| } | ||
| } | ||
|
|
||
| public void updateCacheTimestamp(String storeName, long cacheTimeStampInMs) { | ||
| metadataCacheTimestampMapInMs.put(storeName, cacheTimeStampInMs); | ||
| } | ||
|
|
||
| public void removeCacheTimestamp(String storeName) { | ||
| metadataCacheTimestampMapInMs.remove(storeName); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
33 changes: 33 additions & 0 deletions
33
...ci-client/src/test/java/com/linkedin/davinci/stats/NativeMetadataRepositoryStatsTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| package com.linkedin.davinci.stats; | ||
|
|
||
| import static org.mockito.Mockito.doReturn; | ||
| import static org.mockito.Mockito.mock; | ||
|
|
||
| import io.tehuti.metrics.MetricsRepository; | ||
| import java.time.Clock; | ||
| import org.testng.Assert; | ||
| import org.testng.annotations.Test; | ||
|
|
||
|
|
||
| public class NativeMetadataRepositoryStatsTest { | ||
| @Test | ||
| public void testStats() { | ||
| Clock mockClock = mock(Clock.class); | ||
| doReturn(1000L).when(mockClock).millis(); | ||
| String store1 = "testStore1"; | ||
| String store2 = "testStore2"; | ||
| NativeMetadataRepositoryStats stats = new NativeMetadataRepositoryStats(new MetricsRepository(), "test", mockClock); | ||
| Assert.assertEquals(stats.getMetadataStalenessHighWatermarkMs(), Double.NaN); | ||
| stats.updateCacheTimestamp(store1, 0); | ||
| Assert.assertEquals(stats.getMetadataStalenessHighWatermarkMs(), 1000d); | ||
| stats.updateCacheTimestamp(store2, 1000); | ||
| Assert.assertEquals(stats.getMetadataStalenessHighWatermarkMs(), 1000d); | ||
| doReturn(1500L).when(mockClock).millis(); | ||
| stats.updateCacheTimestamp(store1, 1100); | ||
| Assert.assertEquals(stats.getMetadataStalenessHighWatermarkMs(), 500d); | ||
| stats.removeCacheTimestamp(store2); | ||
| Assert.assertEquals(stats.getMetadataStalenessHighWatermarkMs(), 400d); | ||
| stats.removeCacheTimestamp(store1); | ||
| Assert.assertEquals(stats.getMetadataStalenessHighWatermarkMs(), Double.NaN); | ||
| } | ||
| } |
Oops, something went wrong.