From 6e6c3b33df64e0ae72c9a0b0e40d6b5dac212dee Mon Sep 17 00:00:00 2001 From: Sergey Nuyanzin Date: Tue, 8 Sep 2026 18:57:34 +0200 Subject: [PATCH] [FLINK-40596] Closing metric might lead to a number of test failures --- .../metrics/groups/AbstractMetricGroup.java | 12 ++- .../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 ec5509707f9915..3f31d59bf25862 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> impl /** All metric subgroups of this group. */ private final Map> 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 void close() { 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 @@ private AbstractMetricGroup addGroup(String name, ChildType childType) { 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 d91b6de6240be5..793fe7868cabd5 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.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.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 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 @@ private static void tryRunningJobWithoutEnoughSlots(Configuration configuration) } } + @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 @@ public void initializeOnMaster(InitializeOnMasterContext context) { 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> 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(); + } + } + } }