You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2021/01/12 19:40:31 UTC
[flink] 03/05: [FLINK-20852][metrics] Deprecate back pressure
sampling config options
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 512e7f34cedefa6313380cb7a7ce4123a48e5d58
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Mon Jan 11 17:26:05 2021 +0100
[FLINK-20852][metrics] Deprecate back pressure sampling config options
---
docs/_includes/generated/web_configuration.html | 24 ------------------
.../flink/configuration/ConfigConstants.java | 24 +++---------------
.../org/apache/flink/configuration/WebOptions.java | 27 +++++++++-----------
.../handler/job/JobVertexBackPressureHandler.java | 29 +++++++++++-----------
.../metrics/AggregatingSubtasksMetricsHandler.java | 2 +-
.../rest/handler/legacy/metrics/MetricStore.java | 5 ++--
.../job/JobVertexBackPressureHandlerTest.java | 16 ++++++------
7 files changed, 43 insertions(+), 84 deletions(-)
diff --git a/docs/_includes/generated/web_configuration.html b/docs/_includes/generated/web_configuration.html
index 4b5a98d..4070331 100644
--- a/docs/_includes/generated/web_configuration.html
+++ b/docs/_includes/generated/web_configuration.html
@@ -15,30 +15,6 @@
<td>Access-Control-Allow-Origin header for all responses from the web-frontend.</td>
</tr>
<tr>
- <td><h5>web.backpressure.cleanup-interval</h5></td>
- <td style="word-wrap: break-word;">600000</td>
- <td>Integer</td>
- <td>Time, in milliseconds, after which cached stats are cleaned up if not accessed.</td>
- </tr>
- <tr>
- <td><h5>web.backpressure.delay-between-samples</h5></td>
- <td style="word-wrap: break-word;">50</td>
- <td>Integer</td>
- <td>Delay between samples to determine back pressure in milliseconds.</td>
- </tr>
- <tr>
- <td><h5>web.backpressure.num-samples</h5></td>
- <td style="word-wrap: break-word;">100</td>
- <td>Integer</td>
- <td>Number of samples to take to determine back pressure.</td>
- </tr>
- <tr>
- <td><h5>web.backpressure.refresh-interval</h5></td>
- <td style="word-wrap: break-word;">60000</td>
- <td>Integer</td>
- <td>Time, in milliseconds, after which available stats are deprecated and need to be refreshed (by resampling).</td>
- </tr>
- <tr>
<td><h5>web.checkpoints.history</h5></td>
<td style="word-wrap: break-word;">10</td>
<td>Integer</td>
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
index 1428a6d..a694e50 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
@@ -767,38 +767,22 @@ public final class ConfigConstants {
public static final String JOB_MANAGER_WEB_CHECKPOINTS_HISTORY_SIZE =
"jobmanager.web.checkpoints.history";
- /**
- * Time after which cached stats are cleaned up if not accessed.
- *
- * @deprecated Use {@link WebOptions#BACKPRESSURE_CLEANUP_INTERVAL} instead.
- */
+ /** @deprecated - no longer used. */
@Deprecated
public static final String JOB_MANAGER_WEB_BACK_PRESSURE_CLEAN_UP_INTERVAL =
"jobmanager.web.backpressure.cleanup-interval";
- /**
- * Time after which available stats are deprecated and need to be refreshed (by resampling).
- *
- * @deprecated Use {@link WebOptions#BACKPRESSURE_REFRESH_INTERVAL} instead.
- */
+ /** @deprecated - no longer used. */
@Deprecated
public static final String JOB_MANAGER_WEB_BACK_PRESSURE_REFRESH_INTERVAL =
"jobmanager.web.backpressure.refresh-interval";
- /**
- * Number of samples to take to determine back pressure.
- *
- * @deprecated Use {@link WebOptions#BACKPRESSURE_NUM_SAMPLES} instead.
- */
+ /** @deprecated - no longer used. */
@Deprecated
public static final String JOB_MANAGER_WEB_BACK_PRESSURE_NUM_SAMPLES =
"jobmanager.web.backpressure.num-samples";
- /**
- * Delay between samples to determine back pressure.
- *
- * @deprecated Use {@link WebOptions#BACKPRESSURE_DELAY} instead.
- */
+ /** @deprecated - no longer used. */
@Deprecated
public static final String JOB_MANAGER_WEB_BACK_PRESSURE_DELAY =
"jobmanager.web.backpressure.delay-between-samples";
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/WebOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/WebOptions.java
index a7ea64f..9e3bec9 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/WebOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/WebOptions.java
@@ -124,40 +124,37 @@ public class WebOptions {
.withDeprecatedKeys("jobmanager.web.checkpoints.history")
.withDescription("Number of checkpoints to remember for recent history.");
- /** Time, in milliseconds, after which cached stats are cleaned up if not accessed. */
+ /** @deprecated - no longer used. */
+ @Deprecated
public static final ConfigOption<Integer> BACKPRESSURE_CLEANUP_INTERVAL =
key("web.backpressure.cleanup-interval")
.defaultValue(10 * 60 * 1000)
.withDeprecatedKeys("jobmanager.web.backpressure.cleanup-interval")
- .withDescription(
- "Time, in milliseconds, after which cached stats are cleaned up if not accessed.");
+ .withDescription("This config option is no longer used");
- /**
- * Time, in milliseconds, after which available stats are deprecated and need to be refreshed
- * (by resampling).
- */
+ /** @deprecated - no longer used. */
+ @Deprecated
public static final ConfigOption<Integer> BACKPRESSURE_REFRESH_INTERVAL =
key("web.backpressure.refresh-interval")
.defaultValue(60 * 1000)
.withDeprecatedKeys("jobmanager.web.backpressure.refresh-interval")
- .withDescription(
- "Time, in milliseconds, after which available stats are deprecated and need to be refreshed"
- + " (by resampling).");
+ .withDescription("This config option is no longer used");
- /** Number of samples to take to determine back pressure. */
+ /** @deprecated - no longer used. */
+ @Deprecated
public static final ConfigOption<Integer> BACKPRESSURE_NUM_SAMPLES =
key("web.backpressure.num-samples")
.defaultValue(100)
.withDeprecatedKeys("jobmanager.web.backpressure.num-samples")
- .withDescription("Number of samples to take to determine back pressure.");
+ .withDescription("This config option is no longer used");
- /** Delay between samples to determine back pressure in milliseconds. */
+ /** @deprecated - no longer used. */
+ @Deprecated
public static final ConfigOption<Integer> BACKPRESSURE_DELAY =
key("web.backpressure.delay-between-samples")
.defaultValue(50)
.withDeprecatedKeys("jobmanager.web.backpressure.delay-between-samples")
- .withDescription(
- "Delay between samples to determine back pressure in milliseconds.");
+ .withDescription("This config option is no longer used");
/** Timeout for asynchronous operations by the web monitor in milliseconds. */
public static final ConfigOption<Long> TIMEOUT =
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandler.java
index ed26bc9..47b9bda 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandler.java
@@ -41,7 +41,7 @@ import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
import javax.annotation.Nonnull;
import java.util.ArrayList;
-import java.util.Collection;
+import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
@@ -90,35 +90,36 @@ public class JobVertexBackPressureHandler
}
private JobVertexBackPressureInfo createJobVertexBackPressureInfo(
- Collection<ComponentMetricStore> allSubtaskMetricStores) {
-
+ Map<Integer, ComponentMetricStore> allSubtaskMetricStores) {
+ List<SubtaskBackPressureInfo> subtaskBackPressureInfos =
+ createSubtaskBackPressureInfo(allSubtaskMetricStores);
return new JobVertexBackPressureInfo(
JobVertexBackPressureInfo.VertexBackPressureStatus.OK,
- getBackPressureLevel(getMaxBackPressureRatio(allSubtaskMetricStores)),
+ getBackPressureLevel(getMaxBackPressureRatio(subtaskBackPressureInfos)),
metricFetcher.getLastUpdateTime(),
- createSubtaskBackPressureInfo(allSubtaskMetricStores));
+ subtaskBackPressureInfos);
}
private List<SubtaskBackPressureInfo> createSubtaskBackPressureInfo(
- Collection<ComponentMetricStore> subtaskMetricStores) {
+ Map<Integer, ComponentMetricStore> subtaskMetricStores) {
List<SubtaskBackPressureInfo> result = new ArrayList<>(subtaskMetricStores.size());
- int subTaskIndex = 0;
- for (ComponentMetricStore subtaskMetricStore : subtaskMetricStores) {
+ for (Map.Entry<Integer, ComponentMetricStore> entry : subtaskMetricStores.entrySet()) {
+ int subtaskIndex = entry.getKey();
+ ComponentMetricStore subtaskMetricStore = entry.getValue();
double backPressureRatio = getBackPressureRatio(subtaskMetricStore);
result.add(
new SubtaskBackPressureInfo(
- subTaskIndex,
+ subtaskIndex,
getBackPressureLevel(backPressureRatio),
backPressureRatio));
- subTaskIndex++;
}
+ result.sort(Comparator.comparingInt(SubtaskBackPressureInfo::getSubtask));
return result;
}
- private double getMaxBackPressureRatio(
- Collection<ComponentMetricStore> allSubtaskMetricStores) {
- return allSubtaskMetricStores.stream()
- .mapToDouble(metricStore -> getBackPressureRatio(metricStore))
+ private double getMaxBackPressureRatio(List<SubtaskBackPressureInfo> subtaskBackPressureInfos) {
+ return subtaskBackPressureInfos.stream()
+ .mapToDouble(backPressureInfo -> backPressureInfo.getRatio())
.max()
.getAsDouble();
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/AggregatingSubtasksMetricsHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/AggregatingSubtasksMetricsHandler.java
index 7cc21c8..eef3387 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/AggregatingSubtasksMetricsHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/AggregatingSubtasksMetricsHandler.java
@@ -83,7 +83,7 @@ public class AggregatingSubtasksMetricsHandler
MetricStore.TaskMetricStore taskMetricStore =
store.getTaskMetricStore(jobID.toString(), taskID.toString());
if (taskMetricStore != null) {
- return taskMetricStore.getAllSubtaskMetricStores();
+ return taskMetricStore.getAllSubtaskMetricStores().values();
} else {
return Collections.emptyList();
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/metrics/MetricStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/metrics/MetricStore.java
index e151cca..93f289f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/metrics/MetricStore.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/metrics/MetricStore.java
@@ -27,7 +27,6 @@ import org.slf4j.LoggerFactory;
import javax.annotation.concurrent.ThreadSafe;
-import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -380,8 +379,8 @@ public class MetricStore {
return subtasks.get(subtaskIndex);
}
- public Collection<ComponentMetricStore> getAllSubtaskMetricStores() {
- return subtasks.values();
+ public Map<Integer, ComponentMetricStore> getAllSubtaskMetricStores() {
+ return subtasks;
}
private static TaskMetricStore unmodifiable(TaskMetricStore source) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java
index 262d923..89dbaaf 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexBackPressureHandlerTest.java
@@ -74,19 +74,21 @@ public class JobVertexBackPressureHandlerTest {
private static Collection<MetricDump> getMetricDumps() {
Collection<MetricDump> dumps = new ArrayList<>();
- TaskQueryScopeInfo task1 =
+ TaskQueryScopeInfo task0 =
new TaskQueryScopeInfo(
TEST_JOB_ID_BACK_PRESSURE_STATS_AVAILABLE.toString(),
TEST_JOB_VERTEX_ID.toString(),
- 1);
- dumps.add(new GaugeDump(task1, MetricNames.TASK_BACK_PRESSURED_TIME, "1000"));
+ 0);
+ dumps.add(new GaugeDump(task0, MetricNames.TASK_BACK_PRESSURED_TIME, "1000"));
- TaskQueryScopeInfo task2 =
+ TaskQueryScopeInfo task1 =
new TaskQueryScopeInfo(
TEST_JOB_ID_BACK_PRESSURE_STATS_AVAILABLE.toString(),
TEST_JOB_VERTEX_ID.toString(),
- 2);
- dumps.add(new GaugeDump(task2, MetricNames.TASK_BACK_PRESSURED_TIME, "500"));
+ 1);
+ dumps.add(new GaugeDump(task1, MetricNames.TASK_BACK_PRESSURED_TIME, "500"));
+
+ // missing task2
TaskQueryScopeInfo task3 =
new TaskQueryScopeInfo(
@@ -172,7 +174,7 @@ public class JobVertexBackPressureHandlerTest {
jobVertexBackPressureInfo.getSubtasks().stream()
.map(JobVertexBackPressureInfo.SubtaskBackPressureInfo::getSubtask)
.collect(Collectors.toList()),
- contains(0, 1, 2));
+ contains(0, 1, 3));
}
@Test