This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch release-2.2
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.2 by this push:
new dae60fd1430 [FLINK-40596] Closing metric might lead to a number of
test failures
dae60fd1430 is described below
commit dae60fd1430d31d6c4451687b15809b6691b521e
Author: Sergey Nuyanzin <[email protected]>
AuthorDate: Wed Sep 9 09:02:32 2026 +0200
[FLINK-40596] Closing metric might lead to a number of test failures
---
.../metrics/groups/AbstractMetricGroup.java | 12 ++--
.../runtime/minicluster/MiniClusterITCase.java | 79 ++++++++++++++++++++++
2 files changed, 87 insertions(+), 4 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java
index ec5509707f9..3f31d59bf25 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java
@@ -85,6 +85,9 @@ public abstract class AbstractMetricGroup<A extends
AbstractMetricGroup<?>> impl
/** All metric subgroups of this group. */
private final Map<String, AbstractMetricGroup<?>> groups = new HashMap<>();
+ /** Key under which this group sits in its parent's {@link #groups} map. */
+ private volatile String nameInParent;
+
/**
* The metrics scope represented by this group. For example ["host-7",
"taskmanager-2",
* "window_word_count", "my-mapper" ].
@@ -337,15 +340,15 @@ public abstract class AbstractMetricGroup<A extends
AbstractMetricGroup<?>> impl
metrics.clear();
}
}
- if (parent != null) {
- parent.removeChildGroup(this);
+ if (parent != null && nameInParent != null) {
+ parent.removeChildGroup(nameInParent, this);
}
}
- void removeChildGroup(AbstractMetricGroup<?> childGroup) {
+ void removeChildGroup(String name, AbstractMetricGroup<?> childGroup) {
synchronized (this) {
if (!closed) {
- groups.values().remove(childGroup);
+ groups.remove(name, childGroup);
}
}
}
@@ -518,6 +521,7 @@ public abstract class AbstractMetricGroup<A extends
AbstractMetricGroup<?>> impl
AbstractMetricGroup<?> prior = groups.put(name, newGroup);
if (prior == null || prior.isClosed()) {
// no prior group or closed group with that name
+ newGroup.nameInParent = name;
return newGroup;
} else {
// had a prior group with that name, add the prior group
back
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/MiniClusterITCase.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/MiniClusterITCase.java
index d91b6de6240..793fe7868ca 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/MiniClusterITCase.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/minicluster/MiniClusterITCase.java
@@ -23,12 +23,15 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.core.testutils.FlinkAssertions;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.client.JobExecutionException;
+import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobGraphTestUtils;
import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.jobmanager.Tasks.AgnosticBinaryReceiver;
import org.apache.flink.runtime.jobmanager.Tasks.AgnosticReceiver;
import org.apache.flink.runtime.jobmanager.Tasks.AgnosticTertiaryReceiver;
@@ -41,6 +44,7 @@ import
org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
import org.apache.flink.runtime.jobmaster.JobResult;
import org.apache.flink.runtime.jobmaster.TestingAbstractInvokables.Receiver;
import org.apache.flink.runtime.jobmaster.TestingAbstractInvokables.Sender;
+import org.apache.flink.runtime.metrics.groups.AbstractMetricGroup;
import org.apache.flink.runtime.testtasks.BlockingNoOpInvokable;
import org.apache.flink.runtime.testtasks.NoOpInvokable;
import org.apache.flink.runtime.testtasks.WaitingNoOpInvokable;
@@ -51,11 +55,15 @@ import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import static
org.apache.flink.runtime.util.JobVertexConnectionUtils.connectNewDataSetAsInput;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Integration test cases for the {@link MiniCluster}. */
@@ -190,6 +198,51 @@ class MiniClusterITCase {
}
}
+ @Test
+ void slowMetricTeardownStarvesTheNextJob() throws Exception {
+ final Configuration config = new Configuration();
+ final Duration timeout = Duration.ofSeconds(4);
+ config.set(JobManagerOptions.SLOT_REQUEST_TIMEOUT, timeout);
+
config.set(JobManagerOptions.SCHEDULER_SUBMISSION_RESOURCE_WAIT_TIMEOUT,
timeout);
+ config.set(ResourceManagerOptions.REQUIREMENTS_CHECK_DELAY,
Duration.ofMillis(20));
+ config.set(
+ ResourceManagerOptions.STANDALONE_CLUSTER_STARTUP_PERIOD_TIME,
+ Duration.ofMillis(1L));
+
+ final MiniClusterConfiguration cfg =
+ new MiniClusterConfiguration.Builder()
+ .withRandomPorts()
+ .setNumTaskManagers(1)
+ .setNumSlotsPerTaskManager(1)
+ .setConfiguration(config)
+ .build();
+
+ // Job A grabs the single slot.
+ final JobVertex slow = new JobVertex("slow-metric-teardown");
+ slow.setParallelism(1);
+ slow.setInvokableClass(SlowMetricTeardownInvokable.class);
+ final JobGraph jobA = JobGraphTestUtils.streamingJobGraph(slow);
+
+ // Job B just needs the slot once A releases it.
+ final JobVertex worker = new JobVertex("worker");
+ worker.setParallelism(1);
+ worker.setInvokableClass(NoOpInvokable.class);
+ final JobGraph jobB = JobGraphTestUtils.streamingJobGraph(worker);
+
+ SlowMetricTeardownInvokable.slotHeld = new CountDownLatch(1);
+
+ try (final MiniCluster miniCluster = new MiniCluster(cfg)) {
+ miniCluster.start();
+
+ miniCluster.submitJob(jobA).get();
+ // Wait until A holds the slot and is entering its teardown.
+ SlowMetricTeardownInvokable.slotHeld.await();
+
+ // Passes only if A releases its slot before the timeout.
+ assertThatCode(() ->
miniCluster.executeJobBlocking(jobB)).doesNotThrowAnyException();
+ }
+ }
+
@Test
void testForwardJob() throws Exception {
final int parallelism = 31;
@@ -777,4 +830,30 @@ class MiniClusterITCase {
throw new OutOfMemoryError("Java heap space");
}
}
+
+ public static class SlowMetricTeardownInvokable extends AbstractInvokable {
+
+ public static final int NUM_GROUPS = 50_000;
+
+ /** Counted down once the slot is held and the (slow) teardown is
about to start. */
+ public static volatile CountDownLatch slotHeld = new CountDownLatch(1);
+
+ public SlowMetricTeardownInvokable(Environment environment) {
+ super(environment);
+ }
+
+ @Override
+ public void invoke() throws Exception {
+ final MetricGroup parent =
getEnvironment().getMetricGroup().addGroup("splits");
+ final List<AbstractMetricGroup<?>> kids = new
ArrayList<>(NUM_GROUPS);
+ for (int i = 0; i < NUM_GROUPS; i++) {
+ kids.add((AbstractMetricGroup<?>) parent.addGroup("s" + i));
+ }
+ slotHeld.countDown();
+ // per-split teardown: each close() -> parent.removeChildGroup.
+ for (AbstractMetricGroup<?> kid : kids) {
+ kid.close();
+ }
+ }
+ }
}
