You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/10/25 04:16:12 UTC
[GitHub] [flink] TanYuxin-tyx commented on a diff in pull request #21132: [FLINK-29134][metrics] Do not repeatedly add useless metric updating tasks to avoid wasting resources
TanYuxin-tyx commented on code in PR #21132:
URL: https://github.com/apache/flink/pull/21132#discussion_r1003979797
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/metrics/MetricFetcherTest.java:
##########
@@ -263,7 +220,151 @@ public void testShortUpdateInterval() throws InterruptedException {
fetcher.update();
- assertThat(requestMetricQueryServiceGatewaysCounter.get(), is(2));
+ assertThat(requestMetricQueryServiceGatewaysCounter.get()).isEqualTo(2);
+ }
+
+ @Test
+ public void testIgnoreUpdateRequestWhenFetchingMetrics() throws InterruptedException {
+ final long updateInterval = 1000L;
+ final long waitTimeBeforeReturnMetricResults = updateInterval * 2;
+ final Time timeout = Time.seconds(10L);
+ final AtomicInteger requestMetricQueryServiceGatewaysCounter = new AtomicInteger(0);
+ final JobID jobID = new JobID();
+ final ResourceID tmRID = ResourceID.generate();
+
+ // Create metric fetcher
+ MetricFetcher fetcher =
+ createMetricFetcherWithServiceGateways(
+ jobID,
+ tmRID,
+ timeout,
+ updateInterval,
+ waitTimeBeforeReturnMetricResults,
+ requestMetricQueryServiceGatewaysCounter);
+
+ fetcher.update();
+
+ final long start = System.currentTimeMillis();
+ long difference = 0L;
+
+ while (difference <= updateInterval) {
+ Thread.sleep((int) (updateInterval * 1.5f));
+ difference = System.currentTimeMillis() - start;
+ }
+
+ fetcher.update();
+
+ assertThat(requestMetricQueryServiceGatewaysCounter.get()).isEqualTo(1);
+ }
+
+ @Nonnull
+ private MetricFetcher createMetricFetcherWithServiceGateways(
+ JobID jobID,
+ ResourceID tmRID,
+ Time timeout,
+ long updateInterval,
+ long waitTimeBeforeReturnMetricResults,
+ @Nullable AtomicInteger requestMetricQueryServiceGatewaysCounter) {
+ final ExecutorService executor = java.util.concurrent.Executors.newSingleThreadExecutor();
+ // ========= setup QueryServices
+ // ================================================================================
+
+ final MetricQueryServiceGateway jmQueryService =
+ new TestingMetricQueryServiceGateway.Builder()
+ .setQueryMetricsSupplier(
+ () ->
+ CompletableFuture.completedFuture(
+ new MetricDumpSerialization
+ .MetricSerializationResult(
+ new byte[0],
+ new byte[0],
+ new byte[0],
+ new byte[0],
+ 0,
+ 0,
+ 0,
+ 0)))
+ .build();
+
+ MetricDumpSerialization.MetricSerializationResult requestMetricsAnswer =
+ createRequestDumpAnswer(tmRID, jobID);
+ final MetricQueryServiceGateway tmQueryService =
+ new TestingMetricQueryServiceGateway.Builder()
+ .setQueryMetricsSupplier(
+ () -> {
+ if (waitTimeBeforeReturnMetricResults > 0) {
+ CompletableFuture<
+ MetricDumpSerialization
+ .MetricSerializationResult>
+ metricsAnswerFuture = new CompletableFuture<>();
+ CompletableFuture.completedFuture(null)
+ .thenRunAsync(
+ waitTimeMs(
+ waitTimeBeforeReturnMetricResults),
+ executor)
+ .whenCompleteAsync(
+ (ignore, throwable) -> {
+ if (throwable != null) {
+ fail(throwable.getMessage());
+ }
+ metricsAnswerFuture.complete(
+ requestMetricsAnswer);
+ });
+ return metricsAnswerFuture;
+ } else {
+ return CompletableFuture.completedFuture(
+ requestMetricsAnswer);
+ }
+ })
+ .build();
+
+ // ========= setup JobManager
+ // ==================================================================================
+
+ final TestingRestfulGateway restfulGateway =
+ new TestingRestfulGateway.Builder()
+ .setRequestMultipleJobDetailsSupplier(
+ () ->
+ CompletableFuture.completedFuture(
+ new MultipleJobsDetails(Collections.emptyList())))
+ .setRequestMetricQueryServiceGatewaysSupplier(
+ () -> {
+ if (requestMetricQueryServiceGatewaysCounter != null) {
+ requestMetricQueryServiceGatewaysCounter.incrementAndGet();
+ }
+ return CompletableFuture.completedFuture(
+ Collections.singleton(jmQueryService.getAddress()));
+ })
+ .setRequestTaskManagerMetricQueryServiceGatewaysSupplier(
+ () ->
+ CompletableFuture.completedFuture(
+ Collections.singleton(
+ Tuple2.of(
+ tmRID,
+ tmQueryService.getAddress()))))
+ .build();
+
+ final GatewayRetriever<RestfulGateway> retriever =
+ () -> CompletableFuture.completedFuture(restfulGateway);
+
+ // ========= start MetricFetcher testing
+ // =======================================================================
+ return new MetricFetcherImpl<>(
+ retriever,
+ address -> CompletableFuture.completedFuture(tmQueryService),
+ Executors.directExecutor(),
+ timeout,
+ updateInterval);
+ }
+
+ private static Runnable waitTimeMs(long sleepTimeMs) {
+ return () -> {
+ try {
+ Thread.sleep(sleepTimeMs);
+ } catch (Throwable throwable) {
+ fail(throwable.getMessage());
Review Comment:
I have removed the method because only one situation used it.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org