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 03:06:16 UTC
[GitHub] [flink] reswqa commented on a diff in pull request #21132: [FLINK-29134][metrics] Do not repeatedly add useless metric updating tasks to avoid wasting resources
reswqa commented on code in PR #21132:
URL: https://github.com/apache/flink/pull/21132#discussion_r1003944639
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/metrics/MetricFetcherTest.java:
##########
@@ -43,120 +43,77 @@
import org.junit.Test;
Review Comment:
we should use `org.junit.jupiter.api.Test` instead of `org.junit.Test`
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/metrics/MetricFetcherTest.java:
##########
@@ -43,120 +43,77 @@
import org.junit.Test;
import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
/** Tests for the MetricFetcher. */
public class MetricFetcherTest extends TestLogger {
@Test
public void testUpdate() {
Review Comment:
```suggestion
void testUpdate() {
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/metrics/MetricFetcherTest.java:
##########
@@ -43,120 +43,77 @@
import org.junit.Test;
import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
/** Tests for the MetricFetcher. */
public class MetricFetcherTest extends TestLogger {
Review Comment:
```suggestion
@ExtendWith(TestLoggerExtension.class)
class MetricFetcherTest {
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/metrics/MetricFetcherTest.java:
##########
@@ -43,120 +43,77 @@
import org.junit.Test;
import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
/** Tests for the MetricFetcher. */
public class MetricFetcherTest extends TestLogger {
@Test
public void testUpdate() {
final Time timeout = Time.seconds(10L);
-
- // ========= setup TaskManager
- // =================================================================================
-
JobID jobID = new JobID();
ResourceID tmRID = ResourceID.generate();
- // ========= 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(
- () -> CompletableFuture.completedFuture(requestMetricsAnswer))
- .build();
-
- // ========= setup JobManager
- // ==================================================================================
-
- final TestingRestfulGateway restfulGateway =
- new TestingRestfulGateway.Builder()
- .setRequestMultipleJobDetailsSupplier(
- () ->
- CompletableFuture.completedFuture(
- new MultipleJobsDetails(Collections.emptyList())))
- .setRequestMetricQueryServiceGatewaysSupplier(
- () ->
- 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
- // =======================================================================
+ // Create metric fetcher
MetricFetcher fetcher =
- new MetricFetcherImpl<>(
- retriever,
- address -> CompletableFuture.completedFuture(tmQueryService),
- Executors.directExecutor(),
+ createMetricFetcherWithServiceGateways(
+ jobID,
+ tmRID,
timeout,
- MetricOptions.METRIC_FETCHER_UPDATE_INTERVAL.defaultValue());
+ MetricOptions.METRIC_FETCHER_UPDATE_INTERVAL.defaultValue(),
+ 0,
+ null);
// verify that update fetches metrics and updates the store
fetcher.update();
MetricStore store = fetcher.getMetricStore();
synchronized (store) {
- assertEquals("7", store.getJobManagerMetricStore().getMetric("abc.hist_min"));
- assertEquals("6", store.getJobManagerMetricStore().getMetric("abc.hist_max"));
- assertEquals("4.0", store.getJobManagerMetricStore().getMetric("abc.hist_mean"));
- assertEquals("0.5", store.getJobManagerMetricStore().getMetric("abc.hist_median"));
- assertEquals("5.0", store.getJobManagerMetricStore().getMetric("abc.hist_stddev"));
- assertEquals("0.75", store.getJobManagerMetricStore().getMetric("abc.hist_p75"));
- assertEquals("0.9", store.getJobManagerMetricStore().getMetric("abc.hist_p90"));
- assertEquals("0.95", store.getJobManagerMetricStore().getMetric("abc.hist_p95"));
- assertEquals("0.98", store.getJobManagerMetricStore().getMetric("abc.hist_p98"));
- assertEquals("0.99", store.getJobManagerMetricStore().getMetric("abc.hist_p99"));
- assertEquals("0.999", store.getJobManagerMetricStore().getMetric("abc.hist_p999"));
-
- assertEquals(
- "x",
- store.getTaskManagerMetricStore(tmRID.toString()).metrics.get("abc.gauge"));
- assertEquals("5.0", store.getJobMetricStore(jobID.toString()).metrics.get("abc.jc"));
- assertEquals(
- "2",
- store.getTaskMetricStore(jobID.toString(), "taskid").metrics.get("2.abc.tc"));
- assertEquals(
- "1",
- store.getTaskMetricStore(jobID.toString(), "taskid")
- .metrics
- .get("2.opname.abc.oc"));
+ assertThat("7").isEqualTo(store.getJobManagerMetricStore().getMetric("abc.hist_min"));
Review Comment:
```suggestion
assertThat(store.getJobManagerMetricStore().getMetric("abc.hist_min")).isEqualTo("7");
```
Other places should also be modified like this
##########
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());
Review Comment:
We should try to avoid using the `fail` method. Maybe we can use other approach to do the similar logic, such as making the future complete with exception.
##########
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 {
Review Comment:
```suggestion
void testIgnoreUpdateRequestWhenFetchingMetrics() throws InterruptedException {
```
##########
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
Review Comment:
If I remember correctly, it is non-null by default in flink, except for special case that we annotations as @Nullable
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/metrics/MetricFetcherTest.java:
##########
@@ -239,7 +196,7 @@ public void testLongUpdateInterval() {
fetcher.update();
fetcher.update();
- assertThat(requestMetricQueryServiceGatewaysCounter.get(), is(1));
+ assertThat(requestMetricQueryServiceGatewaysCounter.get()).isEqualTo(1);
Review Comment:
```suggestion
assertThat(requestMetricQueryServiceGatewaysCounter).hasValue(1);
```
##########
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);
Review Comment:
```suggestion
assertThat(requestMetricQueryServiceGatewaysCounter).hasValue(2);
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/metrics/MetricFetcherTest.java:
##########
@@ -50,9 +50,7 @@
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicInteger;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
Review Comment:
Sure, I will take a look.
##########
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);
Review Comment:
```suggestion
assertThat(requestMetricQueryServiceGatewaysCounter).hasValue(1);
```
##########
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:
Maybe just throw an exception directly is ok.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/metrics/MetricFetcherTest.java:
##########
@@ -43,120 +43,77 @@
import org.junit.Test;
import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
/** Tests for the MetricFetcher. */
public class MetricFetcherTest extends TestLogger {
Review Comment:
Unless otherwise specified, the test class and methods should be package-private.
--
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