This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new c84cf68428d [FLINK-40925][rest] `JobVertexBackPressureHandler` throws
`NoSuchElementException` when execution attempts are pruned concurrently
c84cf68428d is described below
commit c84cf68428d91c96ef6493ebe18b21e1d7f4998f
Author: Sergey Nuyanzin <[email protected]>
AuthorDate: Thu Oct 8 00:20:35 2026 +0200
[FLINK-40925][rest] `JobVertexBackPressureHandler` throws
`NoSuchElementException` when execution attempts are pruned concurrently
---
.../handler/job/JobVertexBackPressureHandler.java | 6 +-
.../job/JobVertexBackPressureHandlerTest.java | 127 +++++++++++++++++++--
2 files changed, 123 insertions(+), 10 deletions(-)
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 cf292951b39..9ad6707e5dd 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
@@ -44,6 +44,7 @@ import javax.annotation.Nullable;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
@@ -114,8 +115,10 @@ public class JobVertexBackPressureHandler
for (Map.Entry<Integer, SubtaskMetricStore> entry :
subtaskMetricStores.entrySet()) {
int subtaskIndex = entry.getKey();
SubtaskMetricStore subtaskMetricStore = entry.getValue();
+ // Snapshot the live view over the concurrent attempts map so the
size()/containsKey()/
+ // iterator()/get() calls below stay consistent if another thread
mutates it.
Map<Integer, ComponentMetricStore> allAttemptsMetricStores =
- subtaskMetricStore.getAllAttemptsMetricStores();
+ new
HashMap<>(subtaskMetricStore.getAllAttemptsMetricStores());
if (allAttemptsMetricStores.isEmpty() ||
allAttemptsMetricStores.size() == 1) {
result.add(
createSubtaskAttemptBackpressureInfo(
@@ -126,7 +129,6 @@ public class JobVertexBackPressureHandler
? -1
:
representativeAttempts.getOrDefault(subtaskIndex, -1);
if
(!allAttemptsMetricStores.containsKey(representativeAttempt)) {
- // allAttemptsMetricStores is not empty here
representativeAttempt =
allAttemptsMetricStores.keySet().iterator().next();
}
List<SubtaskBackPressureInfo> otherConcurrentAttempts =
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 8850a065cbd..8c8b2589d85 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
@@ -27,6 +27,8 @@ import
org.apache.flink.runtime.metrics.dump.QueryScopeInfo.TaskQueryScopeInfo;
import org.apache.flink.runtime.rest.handler.HandlerRequest;
import org.apache.flink.runtime.rest.handler.legacy.metrics.MetricFetcher;
import org.apache.flink.runtime.rest.handler.legacy.metrics.MetricStore;
+import
org.apache.flink.runtime.rest.handler.legacy.metrics.MetricStore.ComponentMetricStore;
+import
org.apache.flink.runtime.rest.handler.legacy.metrics.MetricStore.SubtaskMetricStore;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
import org.apache.flink.runtime.rest.messages.JobVertexBackPressureHeaders;
@@ -40,20 +42,25 @@ import
org.apache.flink.runtime.webmonitor.TestingRestfulGateway;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.lang.reflect.Field;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import static
org.apache.flink.runtime.rest.messages.JobVertexBackPressureInfo.VertexBackPressureLevel.HIGH;
import static
org.apache.flink.runtime.rest.messages.JobVertexBackPressureInfo.VertexBackPressureLevel.LOW;
import static
org.apache.flink.runtime.rest.messages.JobVertexBackPressureInfo.VertexBackPressureLevel.OK;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
/** Tests for {@link JobVertexBackPressureHandler}. */
class JobVertexBackPressureHandlerTest {
@@ -120,7 +127,7 @@ class JobVertexBackPressureHandlerTest {
new JobVertexBackPressureHandler(
() ->
CompletableFuture.completedFuture(restfulGateway),
Duration.ofSeconds(10),
- Collections.emptyMap(),
+ Map.of(),
JobVertexBackPressureHeaders.getInstance(),
new MetricFetcher() {
private long updateCount = 0;
@@ -211,8 +218,8 @@ class JobVertexBackPressureHandlerTest {
EmptyRequestBody.getInstance(),
new JobVertexMessageParameters(),
pathParameters,
- Collections.emptyMap(),
- Collections.emptyList());
+ Map.of(),
+ List.of());
final CompletableFuture<JobVertexBackPressureInfo>
jobVertexBackPressureInfoCompletableFuture =
@@ -266,8 +273,8 @@ class JobVertexBackPressureHandlerTest {
EmptyRequestBody.getInstance(),
new JobVertexMessageParameters(),
pathParameters,
- Collections.emptyMap(),
- Collections.emptyList());
+ Map.of(),
+ List.of());
final CompletableFuture<JobVertexBackPressureInfo>
jobVertexBackPressureInfoCompletableFuture =
@@ -300,7 +307,7 @@ class JobVertexBackPressureHandlerTest {
new JobVertexBackPressureHandler(
() ->
CompletableFuture.completedFuture(restfulGateway),
Duration.ofSeconds(10),
- Collections.emptyMap(),
+ Map.of(),
JobVertexBackPressureHeaders.getInstance(),
new MetricFetcher() {
private long updateCount = 0;
@@ -331,8 +338,8 @@ class JobVertexBackPressureHandlerTest {
EmptyRequestBody.getInstance(),
new JobVertexMessageParameters(),
pathParameters,
- Collections.emptyMap(),
- Collections.emptyList());
+ Map.of(),
+ List.of());
final CompletableFuture<JobVertexBackPressureInfo>
jobVertexBackPressureInfoCompletableFuture =
@@ -427,4 +434,108 @@ class JobVertexBackPressureHandlerTest {
.collect(Collectors.toList()))
.containsExactly(0, 1);
}
+
+ /**
+ * The handler must not throw when a subtask's attempts are pruned
concurrently (by {@code
+ * SubtaskMetricStore.retainAttempts()}) between its {@code size()}/{@code
containsKey()} checks
+ * and the {@code keySet().iterator().next()} read.
+ */
+ @Test
+ void testGetBackPressureWhenAttemptsRemovedConcurrently() throws Exception
{
+ MetricStore multipleAttemptsMetricStore = new MetricStore();
+ for (MetricDump metricDump : getMultipleAttemptsMetricDumps()) {
+ multipleAttemptsMetricStore.add(metricDump);
+ }
+
+ // Empty subtask 0's attempts in the window between the handler's
check and its iteration.
+ SubtaskMetricStore subtask =
+ multipleAttemptsMetricStore
+ .getTaskMetricStore(
+
TEST_JOB_ID_BACK_PRESSURE_STATS_AVAILABLE.toString(),
+ TEST_JOB_VERTEX_ID.toString())
+ .getAllSubtaskMetricStores()
+ .get(0);
+ Field attemptsField =
SubtaskMetricStore.class.getDeclaredField("attempts");
+ attemptsField.setAccessible(true);
+
+ Map<Integer, ComponentMetricStore> originalAttempts =
+ (Map<Integer, ComponentMetricStore>)
attemptsField.get(subtask);
+ attemptsField.set(subtask, new SelfEmptyingOnKeySet(originalAttempts));
+
+ JobVertexBackPressureHandler handler =
+ new JobVertexBackPressureHandler(
+ () ->
CompletableFuture.completedFuture(restfulGateway),
+ Duration.ofSeconds(10),
+ Map.of(),
+ JobVertexBackPressureHeaders.getInstance(),
+ new MetricFetcher() {
+ @Override
+ public MetricStore getMetricStore() {
+ return multipleAttemptsMetricStore;
+ }
+
+ @Override
+ public void update() {}
+
+ @Override
+ public long getLastUpdateTime() {
+ return 0;
+ }
+ });
+
+ Map<String, String> pathParameters = new HashMap<>();
+ pathParameters.put(
+ JobIDPathParameter.KEY,
TEST_JOB_ID_BACK_PRESSURE_STATS_AVAILABLE.toString());
+ pathParameters.put(JobVertexIdPathParameter.KEY,
TEST_JOB_VERTEX_ID.toString());
+ HandlerRequest<EmptyRequestBody> request =
+ HandlerRequest.resolveParametersAndCreate(
+ EmptyRequestBody.getInstance(),
+ new JobVertexMessageParameters(),
+ pathParameters,
+ Map.of(),
+ List.of());
+
+ assertThatCode(() -> handler.handleRequest(request, restfulGateway))
+ .doesNotThrowAnyException();
+ }
+
+ private static final class SelfEmptyingOnKeySet
+ extends ConcurrentHashMap<Integer, ComponentMetricStore> {
+ private boolean emptied;
+
+ private SelfEmptyingOnKeySet(Map<Integer, ComponentMetricStore>
initial) {
+ super(initial);
+ }
+
+ private void emptyOnce() {
+ if (!emptied) {
+ emptied = true;
+ clear();
+ }
+ }
+
+ @Override
+ public KeySetView<Integer, ComponentMetricStore> keySet() {
+ emptyOnce();
+ return super.keySet();
+ }
+
+ @Override
+ public KeySetView<Integer, ComponentMetricStore>
keySet(ComponentMetricStore mappedValue) {
+ emptyOnce();
+ return super.keySet(mappedValue);
+ }
+
+ @Override
+ public Collection<ComponentMetricStore> values() {
+ emptyOnce();
+ return super.values();
+ }
+
+ @Override
+ public Set<Entry<Integer, ComponentMetricStore>> entrySet() {
+ emptyOnce();
+ return super.entrySet();
+ }
+ }
}
