You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by zh...@apache.org on 2023/04/20 03:57:14 UTC
[flink] branch master updated (e107783d91f -> bc859f6ffa9)
This is an automated email from the ASF dual-hosted git repository.
zhuzh pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from e107783d91f [FLINK-31859][build] Update cyclonedx-maven-plugin from 2.7.3 to 2.7.7
new c20910ad991 [hotfix][test] Migrate FailoverStrategyCheckpointCoordinatorTest to Junit5 and AssertJ.
new bc859f6ffa9 [FLINK-31853][runtime] Remove use of ExecutionAttemptMappingProvider
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../runtime/checkpoint/CheckpointCoordinator.java | 13 +---
.../runtime/checkpoint/CheckpointStatsTracker.java | 8 +--
.../ExecutionAttemptMappingProvider.java | 82 ----------------------
.../executiongraph/DefaultExecutionGraph.java | 2 -
.../CheckpointCoordinatorMasterHooksTest.java | 1 -
.../CheckpointCoordinatorTestingUtils.java | 1 -
.../FailoverStrategyCheckpointCoordinatorTest.java | 43 ++++++------
7 files changed, 26 insertions(+), 124 deletions(-)
delete mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ExecutionAttemptMappingProvider.java
[flink] 02/02: [FLINK-31853][runtime] Remove use of ExecutionAttemptMappingProvider
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhuzh pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit bc859f6ffa9724459f7db1a7a4d3826a9ae67151
Author: JunRuiLee <jr...@gmail.com>
AuthorDate: Wed Apr 19 20:35:16 2023 +0800
[FLINK-31853][runtime] Remove use of ExecutionAttemptMappingProvider
This closes #22429.
---
.../runtime/checkpoint/CheckpointCoordinator.java | 13 +---
.../runtime/checkpoint/CheckpointStatsTracker.java | 8 +--
.../ExecutionAttemptMappingProvider.java | 82 ----------------------
.../executiongraph/DefaultExecutionGraph.java | 2 -
.../CheckpointCoordinatorMasterHooksTest.java | 1 -
.../CheckpointCoordinatorTestingUtils.java | 1 -
.../FailoverStrategyCheckpointCoordinatorTest.java | 1 -
7 files changed, 6 insertions(+), 102 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
index c7f84e73bc1..10195483ba0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
@@ -222,8 +222,6 @@ public class CheckpointCoordinator {
private final CheckpointPlanCalculator checkpointPlanCalculator;
- private final ExecutionAttemptMappingProvider attemptMappingProvider;
-
private boolean baseLocationsForCheckpointInitialized = false;
private boolean forceFullSnapshot;
@@ -242,7 +240,6 @@ public class CheckpointCoordinator {
ScheduledExecutor timer,
CheckpointFailureManager failureManager,
CheckpointPlanCalculator checkpointPlanCalculator,
- ExecutionAttemptMappingProvider attemptMappingProvider,
CheckpointStatsTracker statsTracker) {
this(
@@ -257,7 +254,6 @@ public class CheckpointCoordinator {
timer,
failureManager,
checkpointPlanCalculator,
- attemptMappingProvider,
SystemClock.getInstance(),
statsTracker,
VertexFinishedStateChecker::new);
@@ -276,7 +272,6 @@ public class CheckpointCoordinator {
ScheduledExecutor timer,
CheckpointFailureManager failureManager,
CheckpointPlanCalculator checkpointPlanCalculator,
- ExecutionAttemptMappingProvider attemptMappingProvider,
Clock clock,
CheckpointStatsTracker statsTracker,
BiFunction<
@@ -314,7 +309,6 @@ public class CheckpointCoordinator {
this.checkpointsCleaner = checkNotNull(checkpointsCleaner);
this.failureManager = checkNotNull(failureManager);
this.checkpointPlanCalculator = checkNotNull(checkpointPlanCalculator);
- this.attemptMappingProvider = checkNotNull(attemptMappingProvider);
this.clock = checkNotNull(clock);
this.isExactlyOnceMode = chkConfig.isExactlyOnce();
this.unalignedCheckpointsEnabled = chkConfig.isUnalignedCheckpointsEnabled();
@@ -2070,11 +2064,8 @@ public class CheckpointCoordinator {
}
}
- public void reportStats(long id, ExecutionAttemptID attemptId, CheckpointMetrics metrics)
- throws CheckpointException {
- attemptMappingProvider
- .getVertex(attemptId)
- .ifPresent(ev -> statsTracker.reportIncompleteStats(id, ev, metrics));
+ public void reportStats(long id, ExecutionAttemptID attemptId, CheckpointMetrics metrics) {
+ statsTracker.reportIncompleteStats(id, attemptId, metrics);
}
// ------------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTracker.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTracker.java
index 52f00fa2d0c..0542ae4e01b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTracker.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTracker.java
@@ -22,7 +22,7 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.Metric;
import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import javax.annotation.Nullable;
@@ -258,16 +258,16 @@ public class CheckpointStatsTracker {
}
public void reportIncompleteStats(
- long checkpointId, ExecutionVertex vertex, CheckpointMetrics metrics) {
+ long checkpointId, ExecutionAttemptID attemptId, CheckpointMetrics metrics) {
statsReadWriteLock.lock();
try {
AbstractCheckpointStats stats = history.getCheckpointById(checkpointId);
if (stats instanceof PendingCheckpointStats) {
((PendingCheckpointStats) stats)
.reportSubtaskStats(
- vertex.getJobvertexId(),
+ attemptId.getJobVertexId(),
new SubtaskStateStats(
- vertex.getParallelSubtaskIndex(),
+ attemptId.getSubtaskIndex(),
System.currentTimeMillis(),
metrics.getBytesPersistedOfThisCheckpoint(),
metrics.getTotalBytesPersisted(),
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ExecutionAttemptMappingProvider.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ExecutionAttemptMappingProvider.java
deleted file mode 100644
index f74804ba0c9..00000000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ExecutionAttemptMappingProvider.java
+++ /dev/null
@@ -1,82 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.checkpoint;
-
-import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
-import org.apache.flink.runtime.executiongraph.ExecutionVertex;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-
-/**
- * Provides a mapping from {@link ExecutionAttemptID} to {@link ExecutionVertex} for currently
- * running execution attempts.
- */
-public class ExecutionAttemptMappingProvider {
-
- /** A full list of tasks. */
- private final List<ExecutionVertex> tasks;
-
- /** The cached mapping, which would only be updated on miss. */
- private Map<ExecutionAttemptID, ExecutionVertex> cachedTasksById;
-
- public ExecutionAttemptMappingProvider(Iterable<ExecutionVertex> tasksIterable) {
- this.tasks = new ArrayList<>();
- tasksIterable.forEach(this.tasks::add);
-
- this.cachedTasksById = new HashMap<>(tasks.size());
- }
-
- public Optional<ExecutionVertex> getVertex(ExecutionAttemptID id) {
- ExecutionVertex vertex = cachedTasksById.get(id);
- if (vertex != null || cachedTasksById.containsKey(id)) {
- return Optional.ofNullable(vertex);
- }
-
- return updateAndGet(id);
- }
-
- private Optional<ExecutionVertex> updateAndGet(ExecutionAttemptID id) {
- synchronized (tasks) {
- ExecutionVertex vertex = cachedTasksById.get(id);
- if (vertex != null || cachedTasksById.containsKey(id)) {
- return Optional.ofNullable(vertex);
- }
-
- Map<ExecutionAttemptID, ExecutionVertex> mappings = getCurrentAttemptMappings();
- if (!mappings.containsKey(id)) {
- mappings.put(id, null);
- }
- cachedTasksById = mappings;
- return Optional.ofNullable(cachedTasksById.get(id));
- }
- }
-
- private Map<ExecutionAttemptID, ExecutionVertex> getCurrentAttemptMappings() {
- Map<ExecutionAttemptID, ExecutionVertex> attemptMappings = new HashMap<>(tasks.size());
- for (ExecutionVertex task : tasks) {
- attemptMappings.put(task.getCurrentExecutionAttempt().getAttemptId(), task);
- }
-
- return attemptMappings;
- }
-}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
index dd3a1092353..6f7127ae468 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
@@ -43,7 +43,6 @@ import org.apache.flink.runtime.checkpoint.CheckpointStatsTracker;
import org.apache.flink.runtime.checkpoint.CheckpointsCleaner;
import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
import org.apache.flink.runtime.checkpoint.DefaultCheckpointPlanCalculator;
-import org.apache.flink.runtime.checkpoint.ExecutionAttemptMappingProvider;
import org.apache.flink.runtime.checkpoint.MasterTriggerRestoreHook;
import org.apache.flink.runtime.checkpoint.OperatorCoordinatorCheckpointContext;
import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
@@ -521,7 +520,6 @@ public class DefaultExecutionGraph implements ExecutionGraph, InternalExecutionG
failureManager,
createCheckpointPlanCalculator(
chkConfig.isEnableCheckpointsAfterTasksFinish()),
- new ExecutionAttemptMappingProvider(getAllExecutionVertices()),
checkpointStatsTracker);
// register the master hooks on the checkpoint coordinator
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorMasterHooksTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorMasterHooksTest.java
index d76208e0d90..e31d7811561 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorMasterHooksTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorMasterHooksTest.java
@@ -500,7 +500,6 @@ class CheckpointCoordinatorMasterHooksTest {
new ExecutionGraphCheckpointPlanCalculatorContext(graph),
graph.getVerticesTopologically(),
false),
- new ExecutionAttemptMappingProvider(graph.getAllExecutionVertices()),
new CheckpointStatsTracker(1, new DummyMetricGroup()));
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTestingUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTestingUtils.java
index c5c69582236..ad07d342117 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTestingUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTestingUtils.java
@@ -842,7 +842,6 @@ public class CheckpointCoordinatorTestingUtils {
timer,
failureManager,
checkpointPlanCalculator,
- new ExecutionAttemptMappingProvider(executionGraph.getAllExecutionVertices()),
SystemClock.getInstance(),
checkpointStatsTracker,
vertexFinishedStateCheckerFactory);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailoverStrategyCheckpointCoordinatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailoverStrategyCheckpointCoordinatorTest.java
index 3235250c86b..43ca9aff156 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailoverStrategyCheckpointCoordinatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailoverStrategyCheckpointCoordinatorTest.java
@@ -94,7 +94,6 @@ class FailoverStrategyCheckpointCoordinatorTest {
new ExecutionGraphCheckpointPlanCalculatorContext(graph),
graph.getVerticesTopologically(),
false),
- new ExecutionAttemptMappingProvider(graph.getAllExecutionVertices()),
mock(CheckpointStatsTracker.class));
// switch current execution's state to running to allow checkpoint could be triggered.
[flink] 01/02: [hotfix][test] Migrate FailoverStrategyCheckpointCoordinatorTest to Junit5 and AssertJ.
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhuzh pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c20910ad991171501c6cac5934b173672f586941
Author: JunRuiLee <jr...@gmail.com>
AuthorDate: Wed Apr 19 21:05:52 2023 +0800
[hotfix][test] Migrate FailoverStrategyCheckpointCoordinatorTest to Junit5 and AssertJ.
---
.../FailoverStrategyCheckpointCoordinatorTest.java | 42 +++++++++++-----------
1 file changed, 20 insertions(+), 22 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailoverStrategyCheckpointCoordinatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailoverStrategyCheckpointCoordinatorTest.java
index 45c980c193f..3235250c86b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailoverStrategyCheckpointCoordinatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/FailoverStrategyCheckpointCoordinatorTest.java
@@ -25,34 +25,32 @@ import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.testutils.TestingUtils;
-import org.apache.flink.testutils.executor.TestExecutorResource;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.testutils.executor.TestExecutorExtension;
import org.apache.flink.util.concurrent.Executors;
import org.apache.flink.util.concurrent.ManuallyTriggeredScheduledExecutor;
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.util.Collections;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadLocalRandom;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
/** Tests for actions of {@link CheckpointCoordinator} on task failures. */
-public class FailoverStrategyCheckpointCoordinatorTest extends TestLogger {
+class FailoverStrategyCheckpointCoordinatorTest {
- @ClassRule
- public static final TestExecutorResource<ScheduledExecutorService> EXECUTOR_RESOURCE =
- TestingUtils.defaultExecutorResource();
+ @RegisterExtension
+ static final TestExecutorExtension<ScheduledExecutorService> EXECUTOR_RESOURCE =
+ TestingUtils.defaultExecutorExtension();
private ManuallyTriggeredScheduledExecutor manualThreadExecutor;
- @Before
- public void setUp() {
+ @BeforeEach
+ void setUp() {
manualThreadExecutor = new ManuallyTriggeredScheduledExecutor();
}
@@ -61,7 +59,7 @@ public class FailoverStrategyCheckpointCoordinatorTest extends TestLogger {
* on job failover could handle the {@code currentPeriodicTrigger} null case well.
*/
@Test
- public void testAbortPendingCheckpointsWithTriggerValidation() throws Exception {
+ void testAbortPendingCheckpointsWithTriggerValidation() throws Exception {
final int maxConcurrentCheckpoints = ThreadLocalRandom.current().nextInt(10) + 1;
ExecutionGraph graph =
new CheckpointCoordinatorTestingUtils.CheckpointExecutionGraphBuilder()
@@ -108,18 +106,18 @@ public class FailoverStrategyCheckpointCoordinatorTest extends TestLogger {
.transitionState(ExecutionState.RUNNING));
checkpointCoordinator.startCheckpointScheduler();
- assertTrue(checkpointCoordinator.isCurrentPeriodicTriggerAvailable());
+ assertThat(checkpointCoordinator.isCurrentPeriodicTriggerAvailable()).isTrue();
// only trigger the periodic scheduling
// we can't trigger all scheduled task, because there is also a cancellation scheduled
manualThreadExecutor.triggerPeriodicScheduledTasks();
manualThreadExecutor.triggerAll();
- assertEquals(1, checkpointCoordinator.getNumberOfPendingCheckpoints());
+ assertThat(checkpointCoordinator.getNumberOfPendingCheckpoints()).isOne();
for (int i = 1; i < maxConcurrentCheckpoints; i++) {
checkpointCoordinator.triggerCheckpoint(false);
manualThreadExecutor.triggerAll();
- assertEquals(i + 1, checkpointCoordinator.getNumberOfPendingCheckpoints());
- assertTrue(checkpointCoordinator.isCurrentPeriodicTriggerAvailable());
+ assertThat(checkpointCoordinator.getNumberOfPendingCheckpoints()).isEqualTo(i + 1);
+ assertThat(checkpointCoordinator.isCurrentPeriodicTriggerAvailable()).isTrue();
}
// as we only support limited concurrent checkpoints, after checkpoint triggered more than
@@ -127,13 +125,13 @@ public class FailoverStrategyCheckpointCoordinatorTest extends TestLogger {
// the currentPeriodicTrigger would been assigned as null.
checkpointCoordinator.triggerCheckpoint(false);
manualThreadExecutor.triggerAll();
- assertEquals(
- maxConcurrentCheckpoints, checkpointCoordinator.getNumberOfPendingCheckpoints());
+ assertThat(checkpointCoordinator.getNumberOfPendingCheckpoints())
+ .isEqualTo(maxConcurrentCheckpoints);
checkpointCoordinator.abortPendingCheckpoints(
new CheckpointException(CheckpointFailureReason.JOB_FAILOVER_REGION));
// after aborting checkpoints, we ensure currentPeriodicTrigger still available.
- assertTrue(checkpointCoordinator.isCurrentPeriodicTriggerAvailable());
- assertEquals(0, checkpointCoordinator.getNumberOfPendingCheckpoints());
+ assertThat(checkpointCoordinator.isCurrentPeriodicTriggerAvailable()).isTrue();
+ assertThat(checkpointCoordinator.getNumberOfPendingCheckpoints()).isZero();
}
}