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 2022/05/20 09:46:37 UTC
[flink] 07/08: [FLINK-17295][runtime] Remove unnecessary fields from Execution and ArchivedExecution
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 579ac0f55422d7ef3215460c69d006acc48ab8b7
Author: Zhu Zhu <re...@gmail.com>
AuthorDate: Tue May 17 16:08:32 2022 +0800
[FLINK-17295][runtime] Remove unnecessary fields from Execution and ArchivedExecution
---
.../flink/runtime/executiongraph/ArchivedExecution.java | 14 ++------------
.../org/apache/flink/runtime/executiongraph/Execution.java | 12 ++++--------
.../runtime/rest/handler/job/JobExceptionsHandlerTest.java | 6 ++----
.../job/SubtaskCurrentAttemptDetailsHandlerTest.java | 6 ++----
.../SubtaskExecutionAttemptAccumulatorsHandlerTest.java | 7 +++----
.../job/SubtaskExecutionAttemptDetailsHandlerTest.java | 7 +++----
6 files changed, 16 insertions(+), 36 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ArchivedExecution.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ArchivedExecution.java
index 16384a59d63..a270d072c4d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ArchivedExecution.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ArchivedExecution.java
@@ -37,8 +37,6 @@ public class ArchivedExecution implements AccessExecution, Serializable {
private final long[] stateTimestamps;
- private final int attemptNumber;
-
private final ExecutionState state;
@Nullable private final ErrorInfo failureInfo; // once assigned, never changes
@@ -50,8 +48,6 @@ public class ArchivedExecution implements AccessExecution, Serializable {
/* Continuously updated map of user-defined accumulators */
private final StringifiedAccumulatorResult[] userAccumulators;
- private final int parallelSubtaskIndex;
-
private final IOMetrics ioMetrics;
public ArchivedExecution(Execution execution) {
@@ -59,12 +55,10 @@ public class ArchivedExecution implements AccessExecution, Serializable {
execution.getUserAccumulatorsStringified(),
execution.getIOMetrics(),
execution.getAttemptId(),
- execution.getAttemptNumber(),
execution.getState(),
execution.getFailureInfo().orElse(null),
execution.getAssignedResourceLocation(),
execution.getAssignedAllocationID(),
- execution.getVertex().getParallelSubtaskIndex(),
execution.getStateTimestamps());
}
@@ -72,22 +66,18 @@ public class ArchivedExecution implements AccessExecution, Serializable {
StringifiedAccumulatorResult[] userAccumulators,
IOMetrics ioMetrics,
ExecutionAttemptID attemptId,
- int attemptNumber,
ExecutionState state,
@Nullable ErrorInfo failureCause,
TaskManagerLocation assignedResourceLocation,
AllocationID assignedAllocationID,
- int parallelSubtaskIndex,
long[] stateTimestamps) {
this.userAccumulators = userAccumulators;
this.ioMetrics = ioMetrics;
this.failureInfo = failureCause;
this.assignedResourceLocation = assignedResourceLocation;
- this.attemptNumber = attemptNumber;
this.attemptId = attemptId;
this.state = state;
this.stateTimestamps = stateTimestamps;
- this.parallelSubtaskIndex = parallelSubtaskIndex;
this.assignedAllocationID = assignedAllocationID;
}
@@ -102,7 +92,7 @@ public class ArchivedExecution implements AccessExecution, Serializable {
@Override
public int getAttemptNumber() {
- return attemptNumber;
+ return attemptId.getAttemptNumber();
}
@Override
@@ -141,7 +131,7 @@ public class ArchivedExecution implements AccessExecution, Serializable {
@Override
public int getParallelSubtaskIndex() {
- return parallelSubtaskIndex;
+ return attemptId.getSubtaskIndex();
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
index 75510f5503d..ff9acef8e29 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
@@ -133,8 +133,6 @@ public class Execution
*/
private final long[] stateTimestamps;
- private final int attemptNumber;
-
private final Time rpcTimeout;
private final Collection<PartitionInfo> partitionInfos;
@@ -212,8 +210,6 @@ public class Execution
attemptNumber);
this.rpcTimeout = checkNotNull(rpcTimeout);
- this.attemptNumber = attemptNumber;
-
this.stateTimestamps = new long[ExecutionState.values().length];
markTimestamp(CREATED, startTimestamp);
@@ -242,7 +238,7 @@ public class Execution
@Override
public int getAttemptNumber() {
- return attemptNumber;
+ return attemptId.getAttemptNumber();
}
@Override
@@ -550,7 +546,7 @@ public class Execution
LOG.info(
"Deploying {} (attempt #{}) with attempt id {} and vertex id {} to {} with allocation id {}",
vertex.getTaskNameWithSubtaskIndex(),
- attemptNumber,
+ getAttemptNumber(),
vertex.getCurrentExecutionAttempt().getAttemptId(),
vertex.getID(),
getAssignedResourceLocation(),
@@ -1467,7 +1463,7 @@ public class Execution
}
public String getVertexWithAttempt() {
- return vertex.getTaskNameWithSubtaskIndex() + " - execution #" + attemptNumber;
+ return vertex.getTaskNameWithSubtaskIndex() + " - execution #" + getAttemptNumber();
}
// ------------------------------------------------------------------------
@@ -1536,7 +1532,7 @@ public class Execution
return String.format(
"Attempt #%d (%s) @ %s - [%s]",
- attemptNumber,
+ getAttemptNumber(),
vertex.getTaskNameWithSubtaskIndex(),
(slot == null ? "(unassigned)" : slot),
state);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobExceptionsHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobExceptionsHandlerTest.java
index 78943aa2ffa..3642c87ea6d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobExceptionsHandlerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobExceptionsHandlerTest.java
@@ -27,7 +27,6 @@ import org.apache.flink.runtime.executiongraph.ArchivedExecution;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionJobVertex;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionVertex;
import org.apache.flink.runtime.executiongraph.ErrorInfo;
-import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.rest.handler.HandlerRequest;
import org.apache.flink.runtime.rest.handler.HandlerRequestException;
@@ -70,6 +69,7 @@ import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
+import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createExecutionAttemptId;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.CoreMatchers.nullValue;
import static org.hamcrest.collection.IsEmptyCollection.empty;
@@ -345,15 +345,13 @@ public class JobExceptionsHandlerTest extends TestLogger {
new ArchivedExecution(
new StringifiedAccumulatorResult[0],
null,
- new ExecutionAttemptID(),
- attempt,
+ createExecutionAttemptId(jobVertexID, subtaskIndex, attempt),
expectedState,
new ErrorInfo(
new RuntimeException("error"),
System.currentTimeMillis()),
assignedResourceLocation,
allocationID,
- subtaskIndex,
timestamps),
new EvictingBoundedList<>(0))
},
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/SubtaskCurrentAttemptDetailsHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/SubtaskCurrentAttemptDetailsHandlerTest.java
index bd60ce534e2..dcf61a79274 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/SubtaskCurrentAttemptDetailsHandlerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/SubtaskCurrentAttemptDetailsHandlerTest.java
@@ -27,7 +27,6 @@ import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.executiongraph.ArchivedExecution;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionVertex;
-import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.executiongraph.IOMetrics;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.rest.handler.HandlerRequest;
@@ -52,6 +51,7 @@ import org.junit.Test;
import java.util.Collections;
import java.util.HashMap;
+import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createExecutionAttemptId;
import static org.junit.Assert.assertEquals;
/** Tests of {@link SubtaskCurrentAttemptDetailsHandler}. */
@@ -90,13 +90,11 @@ public class SubtaskCurrentAttemptDetailsHandlerTest extends TestLogger {
new ArchivedExecution(
new StringifiedAccumulatorResult[0],
ioMetrics,
- new ExecutionAttemptID(),
- attempt,
+ createExecutionAttemptId(jobVertexID, subtaskIndex, attempt),
expectedState,
null,
assignedResourceLocation,
allocationID,
- subtaskIndex,
timestamps);
final ArchivedExecutionVertex executionVertex =
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/SubtaskExecutionAttemptAccumulatorsHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/SubtaskExecutionAttemptAccumulatorsHandlerTest.java
index 0b6b2ea26c0..e5d98dbf8f4 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/SubtaskExecutionAttemptAccumulatorsHandlerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/SubtaskExecutionAttemptAccumulatorsHandlerTest.java
@@ -26,7 +26,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.executiongraph.ArchivedExecution;
-import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.rest.handler.HandlerRequest;
import org.apache.flink.runtime.rest.handler.RestHandlerConfiguration;
import org.apache.flink.runtime.rest.handler.legacy.DefaultExecutionGraphCache;
@@ -47,6 +47,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createExecutionAttemptId;
import static org.junit.Assert.assertEquals;
/** Tests of {@link SubtaskExecutionAttemptAccumulatorsHandler}. */
@@ -93,13 +94,11 @@ public class SubtaskExecutionAttemptAccumulatorsHandlerTest extends TestLogger {
new ArchivedExecution(
accumulatorResults,
null,
- new ExecutionAttemptID(),
- attemptNum,
+ createExecutionAttemptId(new JobVertexID(), subtaskIndex, attemptNum),
ExecutionState.FINISHED,
null,
null,
null,
- subtaskIndex,
new long[ExecutionState.values().length]);
// Invoke tested method.
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/SubtaskExecutionAttemptDetailsHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/SubtaskExecutionAttemptDetailsHandlerTest.java
index 5e19277f5cd..706c4c7701a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/SubtaskExecutionAttemptDetailsHandlerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/SubtaskExecutionAttemptDetailsHandlerTest.java
@@ -28,7 +28,6 @@ import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.executiongraph.ArchivedExecution;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionJobVertex;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionVertex;
-import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.executiongraph.IOMetrics;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.rest.handler.HandlerRequest;
@@ -54,6 +53,7 @@ import org.junit.Test;
import java.util.Collections;
import java.util.HashMap;
+import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createExecutionAttemptId;
import static org.junit.Assert.assertEquals;
/** Tests of {@link SubtaskExecutionAttemptDetailsHandler}. */
@@ -90,13 +90,12 @@ public class SubtaskExecutionAttemptDetailsHandlerTest extends TestLogger {
new ArchivedExecution(
emptyAccumulators,
ioMetrics,
- new ExecutionAttemptID(),
- attempt,
+ createExecutionAttemptId(
+ jobVertexId, subtaskIndex, attempt),
expectedState,
null,
null,
null,
- subtaskIndex,
new long[ExecutionState.values().length]),
new EvictingBoundedList<>(0))
},