1996fanrui commented on code in PR #22552:
URL: https://github.com/apache/flink/pull/22552#discussion_r1221512024
##########
flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/VertexThreadInfoTrackerTest.java:
##########
@@ -159,106 +188,242 @@ public void
testCachedStatsUpdatedAfterRefreshInterval() throws Exception {
// register a CountDownLatch with the cache so we can await refresh of
the entry
CountDownLatch cacheRefreshed = new CountDownLatch(1);
- Cache<JobVertexThreadInfoTracker.Key, VertexThreadInfoStats>
vertexStatsCache =
+ Cache<VertexThreadInfoTracker.JobVertexKey, VertexThreadInfoStats>
jobVertexStatsCache =
createCache(CLEAN_UP_INTERVAL, new
LatchRemovalListener<>(cacheRefreshed));
- final JobVertexThreadInfoTracker<VertexThreadInfoStats> tracker =
+ final VertexThreadInfoTracker tracker =
createThreadInfoTracker(
CLEAN_UP_INTERVAL,
shortRefreshInterval,
- vertexStatsCache,
+ jobVertexStatsCache,
+ null,
initialThreadInfoStats,
threadInfoStatsAfterRefresh);
// no stats yet, but the request triggers async collection of stats
- assertThat(tracker.getVertexStats(JOB_ID,
EXECUTION_JOB_VERTEX)).isNotPresent();
+ assertThat(tracker.getJobVertexStats(JOB_ID,
EXECUTION_JOB_VERTEX)).isNotPresent();
// block until the async call completes and the first result is
available
tracker.getResultAvailableFuture().get();
// retrieve the entry, triggering the refresh as side effect
assertExpectedEqualsReceived(
- initialThreadInfoStats, tracker.getVertexStats(JOB_ID,
EXECUTION_JOB_VERTEX));
+ initialThreadInfoStats, tracker.getJobVertexStats(JOB_ID,
EXECUTION_JOB_VERTEX));
// wait until the entry is refreshed
cacheRefreshed.await();
// verify that we get the second result on the next request
Optional<VertexThreadInfoStats> result =
- tracker.getVertexStats(JOB_ID, EXECUTION_JOB_VERTEX);
+ tracker.getJobVertexStats(JOB_ID, EXECUTION_JOB_VERTEX);
assertExpectedEqualsReceived(threadInfoStatsAfterRefresh, result);
}
+ /** Tests that cached execution vertex result is NOT reused after refresh
interval. */
+ @Test
+ public void testExecutionVertexCachedStatsUpdatedAfterRefreshInterval()
throws Exception {
Review Comment:
Good suggestion.
I have added a new unit test:
`testExecutionVertexShouldBeIgnoredWhenJobVertexIsPending`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
