You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2022/05/02 13:04:53 UTC
[flink] branch master updated: [FLINK-27460][runtime] Remove unused notifyPartitionDataAvailable process
This is an automated email from the ASF dual-hosted git repository.
chesnay 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 05676c4d525 [FLINK-27460][runtime] Remove unused notifyPartitionDataAvailable process
05676c4d525 is described below
commit 05676c4d525f4c9b01c2c048ad263831617c8927
Author: Zhu Zhu <re...@gmail.com>
AuthorDate: Sat Apr 30 16:15:04 2022 +0800
[FLINK-27460][runtime] Remove unused notifyPartitionDataAvailable process
---
.../ResultPartitionDeploymentDescriptor.java | 11 +-
.../executiongraph/DefaultExecutionGraph.java | 13 --
.../flink/runtime/executiongraph/Execution.java | 15 +-
.../runtime/executiongraph/ExecutionGraph.java | 10 -
.../runtime/executiongraph/ExecutionVertex.java | 17 --
.../ResultPartitionConsumableNotifier.java | 28 ---
.../apache/flink/runtime/jobmaster/JobMaster.java | 8 -
.../flink/runtime/jobmaster/JobMasterGateway.java | 19 --
.../flink/runtime/scheduler/DefaultScheduler.java | 59 +----
.../flink/runtime/scheduler/DeploymentHandle.java | 10 -
.../flink/runtime/scheduler/DeploymentOption.java | 36 ---
.../scheduler/ExecutionVertexDeploymentOption.java | 45 ----
.../flink/runtime/scheduler/SchedulerBase.java | 13 --
.../flink/runtime/scheduler/SchedulerNG.java | 2 -
.../runtime/scheduler/SchedulerOperations.java | 6 +-
.../scheduler/adaptive/AdaptiveScheduler.java | 12 +-
.../adaptive/StateWithExecutionGraph.java | 4 -
.../PipelinedRegionSchedulingStrategy.java | 9 +-
.../strategy/SchedulingStrategyUtils.java | 32 +--
.../strategy/VertexwiseSchedulingStrategy.java | 18 +-
.../runtime/taskexecutor/DefaultJobTable.java | 18 --
.../flink/runtime/taskexecutor/JobTable.java | 6 -
.../flink/runtime/taskexecutor/TaskExecutor.java | 12 -
.../rpc/RpcResultPartitionConsumableNotifier.java | 73 ------
...bleNotifyingResultPartitionWriterDecorator.java | 256 ---------------------
.../org/apache/flink/runtime/taskmanager/Task.java | 32 +--
.../ResultPartitionDeploymentDescriptorTest.java | 3 +-
.../runtime/deployment/ShuffleDescriptorTest.java | 3 +-
.../DefaultExecutionGraphDeploymentTest.java | 2 +-
...hDeploymentWithSmallBlobCacheSizeLimitTest.java | 2 +-
.../ExecutionGraphPartitionReleaseTest.java | 7 -
.../ExecutionGraphVariousFailuesTest.java | 86 -------
.../RemoveCachedShuffleDescriptorTest.java | 2 +-
.../partition/AbstractPartitionTrackerTest.java | 3 +-
.../NoOpResultPartitionConsumableNotifier.java | 29 ---
.../io/network/partition/PartitionTestUtils.java | 3 +-
.../partition/ResultPartitionFactoryTest.java | 3 +-
.../io/network/partition/ResultPartitionTest.java | 184 +--------------
.../jobmaster/utils/TestingJobMasterGateway.java | 14 --
.../utils/TestingJobMasterGatewayBuilder.java | 11 -
.../runtime/scheduler/DeploymentHandleTest.java | 9 +-
.../runtime/scheduler/TestingSchedulerNG.java | 5 -
.../adaptive/StateTrackingMockExecutionGraph.java | 6 -
.../benchmark/SchedulerBenchmarkUtils.java | 10 +-
.../deploying/DeployingTasksBenchmarkBase.java | 2 +-
.../RegionToRestartInBatchJobBenchmark.java | 4 +-
.../PartitionReleaseInBatchJobBenchmark.java | 4 +-
.../PipelinedRegionSchedulingStrategyTest.java | 17 +-
.../scheduler/strategy/StrategyTestUtil.java | 25 +-
.../scheduler/strategy/TestSchedulingStrategy.java | 21 +-
.../strategy/TestingSchedulerOperations.java | 10 +-
.../runtime/shuffle/NettyShuffleUtilsTest.java | 3 +-
.../runtime/taskexecutor/DefaultJobTableTest.java | 2 -
.../TaskExecutorPartitionLifecycleTest.java | 2 -
.../taskexecutor/TaskExecutorSubmissionTest.java | 80 +------
.../TaskSubmissionTestEnvironment.java | 9 -
.../runtime/taskmanager/TaskAsyncCallTest.java | 5 -
.../apache/flink/runtime/taskmanager/TaskTest.java | 15 +-
.../flink/runtime/taskmanager/TestTaskBuilder.java | 11 -
.../runtime/util/JvmExitOnFatalErrorTest.java | 2 -
.../tasks/InterruptSensitiveRestoreTest.java | 2 -
.../streaming/runtime/tasks/StreamTaskITCase.java | 3 +-
.../runtime/tasks/StreamTaskSystemExitTest.java | 2 -
.../runtime/tasks/StreamTaskTerminationTest.java | 2 -
.../runtime/tasks/SynchronousCheckpointITCase.java | 5 -
.../tasks/TaskCheckpointingBehaviourTest.java | 2 -
66 files changed, 87 insertions(+), 1287 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptor.java
index a04223f3353..3bf2c802817 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptor.java
@@ -45,19 +45,14 @@ public class ResultPartitionDeploymentDescriptor implements Serializable {
private final int maxParallelism;
- /** Flag whether the result partition should notify master when its data is available. */
- private final boolean notifyPartitionDataAvailable;
-
public ResultPartitionDeploymentDescriptor(
PartitionDescriptor partitionDescriptor,
ShuffleDescriptor shuffleDescriptor,
- int maxParallelism,
- boolean notifyPartitionDataAvailable) {
+ int maxParallelism) {
this.partitionDescriptor = checkNotNull(partitionDescriptor);
this.shuffleDescriptor = checkNotNull(shuffleDescriptor);
KeyGroupRangeAssignment.checkParallelismPreconditions(maxParallelism);
this.maxParallelism = maxParallelism;
- this.notifyPartitionDataAvailable = notifyPartitionDataAvailable;
}
public IntermediateDataSetID getResultId() {
@@ -88,10 +83,6 @@ public class ResultPartitionDeploymentDescriptor implements Serializable {
return shuffleDescriptor;
}
- public boolean notifyPartitionDataAvailable() {
- return notifyPartitionDataAvailable;
- }
-
@Override
public String toString() {
return String.format(
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 c17317a70b8..08ec769a14b 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
@@ -1412,19 +1412,6 @@ public class DefaultExecutionGraph implements ExecutionGraph, InternalExecutionG
return null;
}
- @Override
- public void notifyPartitionDataAvailable(ResultPartitionID partitionId) {
- assertRunningInJobMasterMainThread();
-
- final Execution execution = currentExecutions.get(partitionId.getProducerId());
-
- checkState(
- execution != null,
- "Cannot find execution for execution Id " + partitionId.getPartitionId() + ".");
-
- execution.getVertex().notifyPartitionDataAvailable(partitionId);
- }
-
@Override
public Map<ExecutionAttemptID, Execution> getRegisteredExecutions() {
return Collections.unmodifiableMap(currentExecutions);
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 487d4e3da77..afeeebc54b1 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
@@ -405,14 +405,12 @@ public class Execution
// Actions
// --------------------------------------------------------------------------------------------
- public CompletableFuture<Void> registerProducedPartitions(
- TaskManagerLocation location, boolean notifyPartitionDataAvailable) {
+ public CompletableFuture<Void> registerProducedPartitions(TaskManagerLocation location) {
assertRunningInJobMasterMainThread();
return FutureUtils.thenApplyAsyncIfNotDone(
- registerProducedPartitions(
- vertex, location, attemptId, notifyPartitionDataAvailable),
+ registerProducedPartitions(vertex, location, attemptId),
vertex.getExecutionGraphAccessor().getJobMasterMainThreadExecutor(),
producedPartitionsCache -> {
producedPartitions = producedPartitionsCache;
@@ -437,14 +435,12 @@ public class Execution
});
}
- @VisibleForTesting
- static CompletableFuture<
+ private static CompletableFuture<
Map<IntermediateResultPartitionID, ResultPartitionDeploymentDescriptor>>
registerProducedPartitions(
ExecutionVertex vertex,
TaskManagerLocation location,
- ExecutionAttemptID attemptId,
- boolean notifyPartitionDataAvailable) {
+ ExecutionAttemptID attemptId) {
ProducerDescriptor producerDescriptor = ProducerDescriptor.create(location, attemptId);
@@ -468,8 +464,7 @@ public class Execution
new ResultPartitionDeploymentDescriptor(
partitionDescriptor,
shuffleDescriptor,
- maxParallelism,
- notifyPartitionDataAvailable));
+ maxParallelism));
partitionRegistrations.add(partitionRegistration);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
index d51086767d4..edbdf5acb3f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
@@ -32,7 +32,6 @@ import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
import org.apache.flink.runtime.checkpoint.MasterTriggerRestoreHook;
import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
import org.apache.flink.runtime.executiongraph.failover.flip1.ResultPartitionAvailabilityChecker;
-import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobVertexID;
@@ -192,15 +191,6 @@ public interface ExecutionGraph extends AccessExecutionGraph {
*/
boolean updateState(TaskExecutionStateTransition state);
- /**
- * Mark the data of a result partition to be available. Note that only PIPELINED partitions are
- * accepted because it is for the case that a TM side PIPELINED result partition has data
- * produced and notifies JM.
- *
- * @param partitionId specifying the result partition whose data have become available
- */
- void notifyPartitionDataAvailable(ResultPartitionID partitionId);
-
Map<ExecutionAttemptID, Execution> getRegisteredExecutions();
void registerJobStatusListener(JobStatusListener listener);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
index 4bf20f3faab..fce20faa8d9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
@@ -28,7 +28,6 @@ import org.apache.flink.runtime.JobException;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.execution.ExecutionState;
-import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobmaster.LogicalSlot;
@@ -50,8 +49,6 @@ import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import static org.apache.flink.runtime.execution.ExecutionState.FINISHED;
-import static org.apache.flink.util.Preconditions.checkArgument;
-import static org.apache.flink.util.Preconditions.checkState;
/**
* The ExecutionVertex is a parallel subtask of the execution. It may be executed once, or several
@@ -466,20 +463,6 @@ public class ExecutionVertex
currentExecution.markFailed(t);
}
- void notifyPartitionDataAvailable(ResultPartitionID partitionId) {
- checkArgument(partitionId.getProducerId().equals(currentExecution.getAttemptId()));
-
- final IntermediateResultPartition partition =
- resultPartitions.get(partitionId.getPartitionId());
- checkState(partition != null, "Unknown partition " + partitionId + ".");
- checkState(
- partition.getResultType().isPipelined(),
- "partition data available notification is "
- + "only valid for pipelined partitions.");
-
- partition.markDataProduced();
- }
-
void cachePartitionInfo(PartitionInfo partitionInfo) {
getCurrentExecutionAttempt().cachePartitionInfo(partitionInfo);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionConsumableNotifier.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionConsumableNotifier.java
deleted file mode 100644
index 791891c7efa..00000000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartitionConsumableNotifier.java
+++ /dev/null
@@ -1,28 +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.io.network.partition;
-
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.runtime.taskmanager.TaskActions;
-
-/** Interface for notifications about consumable partitions. */
-public interface ResultPartitionConsumableNotifier {
- void notifyPartitionConsumable(
- JobID jobId, ResultPartitionID partitionId, TaskActions taskActions);
-}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
index 1884d33230c..e91bfa0461e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
@@ -485,14 +485,6 @@ public class JobMaster extends PermanentlyFencedRpcEndpoint<JobMasterId>
}
}
- @Override
- public CompletableFuture<Acknowledge> notifyPartitionDataAvailable(
- final ResultPartitionID partitionID, final Time timeout) {
-
- schedulerNG.notifyPartitionDataAvailable(partitionID);
- return CompletableFuture.completedFuture(Acknowledge.get());
- }
-
@Override
public CompletableFuture<Acknowledge> disconnectTaskManager(
final ResourceID resourceID, final Exception cause) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java
index ff2fcdbaf27..03c853d4946 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java
@@ -29,8 +29,6 @@ import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
-import org.apache.flink.runtime.executiongraph.ExecutionVertex;
-import org.apache.flink.runtime.io.network.partition.ResultPartition;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
import org.apache.flink.runtime.jobgraph.JobVertexID;
@@ -104,23 +102,6 @@ public interface JobMasterGateway
CompletableFuture<ExecutionState> requestPartitionState(
final IntermediateDataSetID intermediateResultId, final ResultPartitionID partitionId);
- /**
- * Notifies the JobManager about available data for a produced partition.
- *
- * <p>There is a call to this method for each {@link ExecutionVertex} instance once per produced
- * {@link ResultPartition} instance, either when first producing data (for pipelined executions)
- * or when all data has been produced (for staged executions).
- *
- * <p>The JobManager then can decide when to schedule the partition consumers of the given
- * session.
- *
- * @param partitionID The partition which has already produced data
- * @param timeout before the rpc call fails
- * @return Future acknowledge of the notification
- */
- CompletableFuture<Acknowledge> notifyPartitionDataAvailable(
- final ResultPartitionID partitionID, @RpcTimeout final Time timeout);
-
/**
* Disconnects the given {@link org.apache.flink.runtime.taskexecutor.TaskExecutor} from the
* {@link JobMaster}.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultScheduler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultScheduler.java
index e9e369c3d19..9a1f88feb66 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultScheduler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultScheduler.java
@@ -36,7 +36,6 @@ import org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHa
import org.apache.flink.runtime.executiongraph.failover.flip1.FailoverStrategy;
import org.apache.flink.runtime.executiongraph.failover.flip1.FailureHandlingResult;
import org.apache.flink.runtime.executiongraph.failover.flip1.RestartBackoffTimeStrategy;
-import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobmanager.scheduler.CoLocationGroup;
import org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException;
@@ -417,28 +416,13 @@ public class DefaultScheduler extends SchedulerBase implements SchedulerOperatio
return executionVertexOperations.cancel(vertex);
}
- @Override
- protected void notifyPartitionDataAvailableInternal(
- final IntermediateResultPartitionID partitionId) {
- schedulingStrategy.onPartitionConsumable(partitionId);
- }
-
// ------------------------------------------------------------------------
// SchedulerOperations
// ------------------------------------------------------------------------
@Override
- public void allocateSlotsAndDeploy(
- final List<ExecutionVertexDeploymentOption> executionVertexDeploymentOptions) {
- validateDeploymentOptions(executionVertexDeploymentOptions);
-
- final Map<ExecutionVertexID, ExecutionVertexDeploymentOption> deploymentOptionsByVertex =
- groupDeploymentOptionsByVertexId(executionVertexDeploymentOptions);
-
- final List<ExecutionVertexID> verticesToDeploy =
- executionVertexDeploymentOptions.stream()
- .map(ExecutionVertexDeploymentOption::getExecutionVertexId)
- .collect(Collectors.toList());
+ public void allocateSlotsAndDeploy(final List<ExecutionVertexID> verticesToDeploy) {
+ validateDeploymentOptions(verticesToDeploy);
final Map<ExecutionVertexID, ExecutionVertexVersion> requiredVersionByVertex =
executionVertexVersioner.recordVertexModifications(verticesToDeploy);
@@ -446,21 +430,16 @@ public class DefaultScheduler extends SchedulerBase implements SchedulerOperatio
transitionToScheduled(verticesToDeploy);
final List<SlotExecutionVertexAssignment> slotExecutionVertexAssignments =
- allocateSlots(executionVertexDeploymentOptions);
+ allocateSlots(verticesToDeploy);
final List<DeploymentHandle> deploymentHandles =
- createDeploymentHandles(
- requiredVersionByVertex,
- deploymentOptionsByVertex,
- slotExecutionVertexAssignments);
+ createDeploymentHandles(requiredVersionByVertex, slotExecutionVertexAssignments);
waitForAllSlotsAndDeploy(deploymentHandles);
}
- private void validateDeploymentOptions(
- final Collection<ExecutionVertexDeploymentOption> deploymentOptions) {
- deploymentOptions.stream()
- .map(ExecutionVertexDeploymentOption::getExecutionVertexId)
+ private void validateDeploymentOptions(final Collection<ExecutionVertexID> verticesToDeploy) {
+ verticesToDeploy.stream()
.map(this::getExecutionVertex)
.forEach(
v ->
@@ -471,28 +450,13 @@ public class DefaultScheduler extends SchedulerBase implements SchedulerOperatio
v.getExecutionState()));
}
- private static Map<ExecutionVertexID, ExecutionVertexDeploymentOption>
- groupDeploymentOptionsByVertexId(
- final Collection<ExecutionVertexDeploymentOption>
- executionVertexDeploymentOptions) {
- return executionVertexDeploymentOptions.stream()
- .collect(
- Collectors.toMap(
- ExecutionVertexDeploymentOption::getExecutionVertexId,
- Function.identity()));
- }
-
private List<SlotExecutionVertexAssignment> allocateSlots(
- final List<ExecutionVertexDeploymentOption> executionVertexDeploymentOptions) {
- return executionSlotAllocator.allocateSlotsFor(
- executionVertexDeploymentOptions.stream()
- .map(ExecutionVertexDeploymentOption::getExecutionVertexId)
- .collect(Collectors.toList()));
+ final List<ExecutionVertexID> verticesToDeploy) {
+ return executionSlotAllocator.allocateSlotsFor(verticesToDeploy);
}
private static List<DeploymentHandle> createDeploymentHandles(
final Map<ExecutionVertexID, ExecutionVertexVersion> requiredVersionByVertex,
- final Map<ExecutionVertexID, ExecutionVertexDeploymentOption> deploymentOptionsByVertex,
final List<SlotExecutionVertexAssignment> slotExecutionVertexAssignments) {
return slotExecutionVertexAssignments.stream()
@@ -502,7 +466,6 @@ public class DefaultScheduler extends SchedulerBase implements SchedulerOperatio
slotExecutionVertexAssignment.getExecutionVertexId();
return new DeploymentHandle(
requiredVersionByVertex.get(executionVertexId),
- deploymentOptionsByVertex.get(executionVertexId),
slotExecutionVertexAssignment);
})
.collect(Collectors.toList());
@@ -626,15 +589,11 @@ public class DefaultScheduler extends SchedulerBase implements SchedulerOperatio
// the produced partition registration process can be skipped as well
if (logicalSlot != null) {
final ExecutionVertex executionVertex = getExecutionVertex(executionVertexId);
- final boolean notifyPartitionDataAvailable =
- deploymentHandle.getDeploymentOption().notifyPartitionDataAvailable();
final CompletableFuture<Void> partitionRegistrationFuture =
executionVertex
.getCurrentExecutionAttempt()
- .registerProducedPartitions(
- logicalSlot.getTaskManagerLocation(),
- notifyPartitionDataAvailable);
+ .registerProducedPartitions(logicalSlot.getTaskManagerLocation());
return FutureUtils.orTimeout(
partitionRegistrationFuture,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DeploymentHandle.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DeploymentHandle.java
index 051db11d0cf..4724c37bd6d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DeploymentHandle.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DeploymentHandle.java
@@ -34,7 +34,6 @@ import java.util.concurrent.CompletableFuture;
*
* <ul>
* <li>{@link ExecutionVertexVersion}
- * <li>{@link ExecutionVertexDeploymentOption}
* <li>{@link SlotExecutionVertexAssignment}
* </ul>
*/
@@ -42,18 +41,13 @@ class DeploymentHandle {
private final ExecutionVertexVersion requiredVertexVersion;
- private final ExecutionVertexDeploymentOption executionVertexDeploymentOption;
-
private final SlotExecutionVertexAssignment slotExecutionVertexAssignment;
public DeploymentHandle(
final ExecutionVertexVersion requiredVertexVersion,
- final ExecutionVertexDeploymentOption executionVertexDeploymentOption,
final SlotExecutionVertexAssignment slotExecutionVertexAssignment) {
this.requiredVertexVersion = Preconditions.checkNotNull(requiredVertexVersion);
- this.executionVertexDeploymentOption =
- Preconditions.checkNotNull(executionVertexDeploymentOption);
this.slotExecutionVertexAssignment =
Preconditions.checkNotNull(slotExecutionVertexAssignment);
}
@@ -66,10 +60,6 @@ class DeploymentHandle {
return requiredVertexVersion;
}
- public DeploymentOption getDeploymentOption() {
- return executionVertexDeploymentOption.getDeploymentOption();
- }
-
public SlotExecutionVertexAssignment getSlotExecutionVertexAssignment() {
return slotExecutionVertexAssignment;
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DeploymentOption.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DeploymentOption.java
deleted file mode 100644
index 8f08150b0dc..00000000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DeploymentOption.java
+++ /dev/null
@@ -1,36 +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.scheduler;
-
-/**
- * Deployment option which indicates whether the task should notify master when its data is
- * available.
- */
-public class DeploymentOption {
-
- private final boolean notifyPartitionDataAvailable;
-
- public DeploymentOption(boolean notifyPartitionDataAvailable) {
- this.notifyPartitionDataAvailable = notifyPartitionDataAvailable;
- }
-
- public boolean notifyPartitionDataAvailable() {
- return notifyPartitionDataAvailable;
- }
-}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/ExecutionVertexDeploymentOption.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/ExecutionVertexDeploymentOption.java
deleted file mode 100644
index 5c5a85838cc..00000000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/ExecutionVertexDeploymentOption.java
+++ /dev/null
@@ -1,45 +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.scheduler;
-
-import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
-
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
-/** Component that stores the task need to be scheduled and the option for deployment. */
-public class ExecutionVertexDeploymentOption {
-
- private final ExecutionVertexID executionVertexId;
-
- private final DeploymentOption deploymentOption;
-
- public ExecutionVertexDeploymentOption(
- ExecutionVertexID executionVertexId, DeploymentOption deploymentOption) {
- this.executionVertexId = checkNotNull(executionVertexId);
- this.deploymentOption = checkNotNull(deploymentOption);
- }
-
- public ExecutionVertexID getExecutionVertexId() {
- return executionVertexId;
- }
-
- public DeploymentOption getDeploymentOption() {
- return deploymentOption;
- }
-}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java
index 0ca22335c07..8360c1fbe9e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java
@@ -60,7 +60,6 @@ import org.apache.flink.runtime.executiongraph.metrics.DownTimeGauge;
import org.apache.flink.runtime.executiongraph.metrics.UpTimeGauge;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
-import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobVertexID;
@@ -792,18 +791,6 @@ public abstract class SchedulerBase implements SchedulerNG, CheckpointScheduling
return executionGraphHandler.requestPartitionState(intermediateResultId, resultPartitionId);
}
- @Override
- public final void notifyPartitionDataAvailable(final ResultPartitionID partitionId) {
- mainThreadExecutor.assertRunningInMainThread();
-
- executionGraph.notifyPartitionDataAvailable(partitionId);
-
- notifyPartitionDataAvailableInternal(partitionId.getPartitionId());
- }
-
- protected void notifyPartitionDataAvailableInternal(
- IntermediateResultPartitionID resultPartitionId) {}
-
@VisibleForTesting
Iterable<RootExceptionHistoryEntry> getExceptionHistory() {
return exceptionHistory.toArrayList();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerNG.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerNG.java
index faa1b593585..68877e287cd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerNG.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerNG.java
@@ -87,8 +87,6 @@ public interface SchedulerNG extends GlobalFailureHandler, AutoCloseableAsync {
IntermediateDataSetID intermediateResultId, ResultPartitionID resultPartitionId)
throws PartitionProducerDisposedException;
- void notifyPartitionDataAvailable(ResultPartitionID partitionID);
-
ExecutionGraphInfo requestJob();
JobStatus requestJobStatus();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerOperations.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerOperations.java
index 27baf242129..eb6ac99fec9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerOperations.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerOperations.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.scheduler;
+import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
import org.apache.flink.runtime.scheduler.strategy.SchedulingStrategy;
import java.util.List;
@@ -31,8 +32,7 @@ public interface SchedulerOperations {
* with smaller indices will be deployed earlier. Only vertices in CREATED state will be
* accepted. Errors will happen if scheduling Non-CREATED vertices.
*
- * @param executionVertexDeploymentOptions The deployment options of tasks to be deployed
+ * @param verticesToDeploy The execution vertices to deploy
*/
- void allocateSlotsAndDeploy(
- List<ExecutionVertexDeploymentOption> executionVertexDeploymentOptions);
+ void allocateSlotsAndDeploy(List<ExecutionVertexID> verticesToDeploy);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
index 562c17f05c3..7b64106c1f5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
@@ -525,15 +525,6 @@ public class AdaptiveScheduler
.orElseThrow(() -> new PartitionProducerDisposedException(resultPartitionId));
}
- @Override
- public void notifyPartitionDataAvailable(ResultPartitionID partitionID) {
- state.tryRun(
- StateWithExecutionGraph.class,
- stateWithExecutionGraph ->
- stateWithExecutionGraph.notifyPartitionDataAvailable(partitionID),
- "notifyPartitionDataAvailable");
- }
-
@Override
public ExecutionGraphInfo requestJob() {
return new ExecutionGraphInfo(state.getJob(), exceptionHistory.toArrayList());
@@ -998,8 +989,7 @@ public class AdaptiveScheduler
final CompletableFuture<Void> registrationFuture =
executionVertex
.getCurrentExecutionAttempt()
- .registerProducedPartitions(
- assignedSlot.getTaskManagerLocation(), false);
+ .registerProducedPartitions(assignedSlot.getTaskManagerLocation());
Preconditions.checkState(
registrationFuture.isDone(),
"Partition registration must be completed immediately for reactive mode");
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithExecutionGraph.java
index 16c4c75e202..8ba39311714 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithExecutionGraph.java
@@ -167,10 +167,6 @@ abstract class StateWithExecutionGraph implements State {
return logger;
}
- void notifyPartitionDataAvailable(ResultPartitionID partitionID) {
- executionGraph.notifyPartitionDataAvailable(partitionID);
- }
-
SerializedInputSplit requestNextInputSplit(
JobVertexID vertexID, ExecutionAttemptID executionAttempt) throws IOException {
return executionGraphHandler.requestNextInputSplit(vertexID, executionAttempt);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/PipelinedRegionSchedulingStrategy.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/PipelinedRegionSchedulingStrategy.java
index bf7fe27c21e..dda505a4653 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/PipelinedRegionSchedulingStrategy.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/PipelinedRegionSchedulingStrategy.java
@@ -21,8 +21,6 @@ package org.apache.flink.runtime.scheduler.strategy;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
-import org.apache.flink.runtime.scheduler.DeploymentOption;
-import org.apache.flink.runtime.scheduler.ExecutionVertexDeploymentOption;
import org.apache.flink.runtime.scheduler.SchedulerOperations;
import org.apache.flink.util.IterableUtils;
@@ -48,8 +46,6 @@ public class PipelinedRegionSchedulingStrategy implements SchedulingStrategy {
private final SchedulingTopology schedulingTopology;
- private final DeploymentOption deploymentOption = new DeploymentOption(false);
-
/** External consumer regions of each ConsumedPartitionGroup. */
private final Map<ConsumedPartitionGroup, Set<SchedulingPipelinedRegion>>
partitionGroupConsumerRegions = new IdentityHashMap<>();
@@ -228,10 +224,7 @@ public class PipelinedRegionSchedulingStrategy implements SchedulingStrategy {
areRegionVerticesAllInCreatedState(region),
"BUG: trying to schedule a region which is not in CREATED state");
- final List<ExecutionVertexDeploymentOption> vertexDeploymentOptions =
- SchedulingStrategyUtils.createExecutionVertexDeploymentOptions(
- regionVerticesSorted.get(region), id -> deploymentOption);
- schedulerOperations.allocateSlotsAndDeploy(vertexDeploymentOptions);
+ schedulerOperations.allocateSlotsAndDeploy(regionVerticesSorted.get(region));
}
private boolean areRegionInputsAllConsumable(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/SchedulingStrategyUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/SchedulingStrategyUtils.java
index efa2cb42811..27aff17a56c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/SchedulingStrategyUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/SchedulingStrategyUtils.java
@@ -18,53 +18,25 @@
package org.apache.flink.runtime.scheduler.strategy;
-import org.apache.flink.runtime.scheduler.DeploymentOption;
-import org.apache.flink.runtime.scheduler.ExecutionVertexDeploymentOption;
import org.apache.flink.util.IterableUtils;
-import java.util.ArrayList;
-import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Set;
-import java.util.function.Function;
import java.util.stream.Collectors;
/** Utils for {@link SchedulingStrategy}. */
class SchedulingStrategyUtils {
- static List<ExecutionVertexDeploymentOption>
- createExecutionVertexDeploymentOptionsInTopologicalOrder(
- final SchedulingTopology topology,
- final Set<ExecutionVertexID> verticesToDeploy,
- final Function<ExecutionVertexID, DeploymentOption> deploymentOptionRetriever) {
+ static List<ExecutionVertexID> sortExecutionVerticesInTopologicalOrder(
+ final SchedulingTopology topology, final Set<ExecutionVertexID> verticesToDeploy) {
return IterableUtils.toStream(topology.getVertices())
.map(SchedulingExecutionVertex::getId)
.filter(verticesToDeploy::contains)
- .map(
- executionVertexID ->
- new ExecutionVertexDeploymentOption(
- executionVertexID,
- deploymentOptionRetriever.apply(executionVertexID)))
.collect(Collectors.toList());
}
- static List<ExecutionVertexDeploymentOption> createExecutionVertexDeploymentOptions(
- final Collection<ExecutionVertexID> verticesToDeploy,
- final Function<ExecutionVertexID, DeploymentOption> deploymentOptionRetriever) {
-
- final List<ExecutionVertexDeploymentOption> deploymentOptions =
- new ArrayList<>(verticesToDeploy.size());
- for (ExecutionVertexID executionVertexId : verticesToDeploy) {
- final ExecutionVertexDeploymentOption deploymentOption =
- new ExecutionVertexDeploymentOption(
- executionVertexId, deploymentOptionRetriever.apply(executionVertexId));
- deploymentOptions.add(deploymentOption);
- }
- return deploymentOptions;
- }
-
static List<SchedulingPipelinedRegion> sortPipelinedRegionsInTopologicalOrder(
final SchedulingTopology topology, final Set<SchedulingPipelinedRegion> regions) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/VertexwiseSchedulingStrategy.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/VertexwiseSchedulingStrategy.java
index b877b5b53ef..4f7d01e4e8f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/VertexwiseSchedulingStrategy.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/VertexwiseSchedulingStrategy.java
@@ -20,8 +20,6 @@ package org.apache.flink.runtime.scheduler.strategy;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
-import org.apache.flink.runtime.scheduler.DeploymentOption;
-import org.apache.flink.runtime.scheduler.ExecutionVertexDeploymentOption;
import org.apache.flink.runtime.scheduler.SchedulerOperations;
import org.apache.flink.runtime.scheduler.SchedulingTopologyListener;
import org.apache.flink.util.IterableUtils;
@@ -51,8 +49,6 @@ public class VertexwiseSchedulingStrategy
private final SchedulingTopology schedulingTopology;
- private final DeploymentOption deploymentOption = new DeploymentOption(false);
-
private final Set<ExecutionVertexID> newVertices = new HashSet<>();
public VertexwiseSchedulingStrategy(
@@ -141,14 +137,12 @@ public class VertexwiseSchedulingStrategy
if (verticesToDeploy.isEmpty()) {
return;
}
- final List<ExecutionVertexDeploymentOption> vertexDeploymentOptions =
- SchedulingStrategyUtils.createExecutionVertexDeploymentOptionsInTopologicalOrder(
- schedulingTopology, verticesToDeploy, id -> deploymentOption);
-
- vertexDeploymentOptions.forEach(
- option ->
- schedulerOperations.allocateSlotsAndDeploy(
- Collections.singletonList(option)));
+ final List<ExecutionVertexID> sortedVerticesToDeploy =
+ SchedulingStrategyUtils.sortExecutionVerticesInTopologicalOrder(
+ schedulingTopology, verticesToDeploy);
+
+ sortedVerticesToDeploy.forEach(
+ id -> schedulerOperations.allocateSlotsAndDeploy(Collections.singletonList(id)));
}
private boolean areVertexInputsAllConsumable(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/DefaultJobTable.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/DefaultJobTable.java
index dbb04e4ac30..a979738a403 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/DefaultJobTable.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/DefaultJobTable.java
@@ -21,7 +21,6 @@ package org.apache.flink.runtime.taskexecutor;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.execution.librarycache.LibraryCacheManager;
-import org.apache.flink.runtime.io.network.partition.ResultPartitionConsumableNotifier;
import org.apache.flink.runtime.jobmaster.JobMasterGateway;
import org.apache.flink.runtime.jobmaster.JobMasterId;
import org.apache.flink.runtime.taskmanager.CheckpointResponder;
@@ -167,11 +166,6 @@ public final class DefaultJobTable implements JobTable {
return jobServices.getClassLoaderHandle();
}
- @Override
- public ResultPartitionConsumableNotifier getResultPartitionConsumableNotifier() {
- return verifyContainsEstablishedConnection().getResultPartitionConsumableNotifier();
- }
-
@Override
public PartitionProducerStateChecker getPartitionStateChecker() {
return verifyContainsEstablishedConnection().getPartitionStateChecker();
@@ -204,7 +198,6 @@ public final class DefaultJobTable implements JobTable {
TaskManagerActions taskManagerActions,
CheckpointResponder checkpointResponder,
GlobalAggregateManager aggregateManager,
- ResultPartitionConsumableNotifier resultPartitionConsumableNotifier,
PartitionProducerStateChecker partitionStateChecker) {
verifyJobIsNotClosed();
Preconditions.checkState(connection == null);
@@ -216,7 +209,6 @@ public final class DefaultJobTable implements JobTable {
taskManagerActions,
checkpointResponder,
aggregateManager,
- resultPartitionConsumableNotifier,
partitionStateChecker);
resourceIdJobIdIndex.put(resourceId, jobId);
@@ -266,9 +258,6 @@ public final class DefaultJobTable implements JobTable {
// GlobalAggregateManager interface to job manager
private final GlobalAggregateManager globalAggregateManager;
- // Result partition consumable notifier for the specific job manager
- private final ResultPartitionConsumableNotifier resultPartitionConsumableNotifier;
-
// Partition state checker for the specific job manager
private final PartitionProducerStateChecker partitionStateChecker;
@@ -278,15 +267,12 @@ public final class DefaultJobTable implements JobTable {
TaskManagerActions taskManagerActions,
CheckpointResponder checkpointResponder,
GlobalAggregateManager globalAggregateManager,
- ResultPartitionConsumableNotifier resultPartitionConsumableNotifier,
PartitionProducerStateChecker partitionStateChecker) {
this.resourceID = Preconditions.checkNotNull(resourceID);
this.jobMasterGateway = Preconditions.checkNotNull(jobMasterGateway);
this.taskManagerActions = Preconditions.checkNotNull(taskManagerActions);
this.checkpointResponder = Preconditions.checkNotNull(checkpointResponder);
this.globalAggregateManager = Preconditions.checkNotNull(globalAggregateManager);
- this.resultPartitionConsumableNotifier =
- Preconditions.checkNotNull(resultPartitionConsumableNotifier);
this.partitionStateChecker = Preconditions.checkNotNull(partitionStateChecker);
}
@@ -314,10 +300,6 @@ public final class DefaultJobTable implements JobTable {
return globalAggregateManager;
}
- public ResultPartitionConsumableNotifier getResultPartitionConsumableNotifier() {
- return resultPartitionConsumableNotifier;
- }
-
public PartitionProducerStateChecker getPartitionStateChecker() {
return partitionStateChecker;
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/JobTable.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/JobTable.java
index 6ad64ac5416..8640d8b99da 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/JobTable.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/JobTable.java
@@ -21,7 +21,6 @@ package org.apache.flink.runtime.taskexecutor;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.execution.librarycache.LibraryCacheManager;
-import org.apache.flink.runtime.io.network.partition.ResultPartitionConsumableNotifier;
import org.apache.flink.runtime.jobmaster.JobMasterGateway;
import org.apache.flink.runtime.jobmaster.JobMasterId;
import org.apache.flink.runtime.taskmanager.CheckpointResponder;
@@ -149,8 +148,6 @@ public interface JobTable extends AutoCloseable {
* @param taskManagerActions taskManagerActions associated with this connection
* @param checkpointResponder checkpointResponder associated with this connection
* @param aggregateManager aggregateManager associated with this connection
- * @param resultPartitionConsumableNotifier resultPartitionConsumableNotifier associated
- * with this connection
* @param partitionStateChecker partitionStateChecker associated with this connection
* @return the established {@link Connection}
* @throws IllegalStateException if the job is already connected
@@ -161,7 +158,6 @@ public interface JobTable extends AutoCloseable {
TaskManagerActions taskManagerActions,
CheckpointResponder checkpointResponder,
GlobalAggregateManager aggregateManager,
- ResultPartitionConsumableNotifier resultPartitionConsumableNotifier,
PartitionProducerStateChecker partitionStateChecker);
/** Closes this job and removes it from the owning {@link JobTable}. */
@@ -200,8 +196,6 @@ public interface JobTable extends AutoCloseable {
LibraryCacheManager.ClassLoaderHandle getClassLoaderHandle();
- ResultPartitionConsumableNotifier getResultPartitionConsumableNotifier();
-
PartitionProducerStateChecker getPartitionStateChecker();
JobID getJobId();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
index efb16041b2b..0ae17859d47 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
@@ -54,7 +54,6 @@ import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.instance.HardwareDescription;
import org.apache.flink.runtime.instance.InstanceID;
-import org.apache.flink.runtime.io.network.partition.ResultPartitionConsumableNotifier;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.io.network.partition.TaskExecutorPartitionInfo;
import org.apache.flink.runtime.io.network.partition.TaskExecutorPartitionTracker;
@@ -113,7 +112,6 @@ import org.apache.flink.runtime.taskexecutor.rpc.RpcGlobalAggregateManager;
import org.apache.flink.runtime.taskexecutor.rpc.RpcInputSplitProvider;
import org.apache.flink.runtime.taskexecutor.rpc.RpcKvStateRegistryListener;
import org.apache.flink.runtime.taskexecutor.rpc.RpcPartitionStateChecker;
-import org.apache.flink.runtime.taskexecutor.rpc.RpcResultPartitionConsumableNotifier;
import org.apache.flink.runtime.taskexecutor.rpc.RpcTaskOperatorEventGateway;
import org.apache.flink.runtime.taskexecutor.slot.SlotActions;
import org.apache.flink.runtime.taskexecutor.slot.SlotAllocationSnapshot;
@@ -684,8 +682,6 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
LibraryCacheManager.ClassLoaderHandle classLoaderHandle =
jobManagerConnection.getClassLoaderHandle();
- ResultPartitionConsumableNotifier resultPartitionConsumableNotifier =
- jobManagerConnection.getResultPartitionConsumableNotifier();
PartitionProducerStateChecker partitionStateChecker =
jobManagerConnection.getPartitionStateChecker();
@@ -751,7 +747,6 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
fileCache,
taskManagerConfiguration,
taskMetricGroup,
- resultPartitionConsumableNotifier,
partitionStateChecker,
getRpcService().getScheduledExecutor());
@@ -1729,12 +1724,6 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
CheckpointResponder checkpointResponder = new RpcCheckpointResponder(jobMasterGateway);
GlobalAggregateManager aggregateManager = new RpcGlobalAggregateManager(jobMasterGateway);
- ResultPartitionConsumableNotifier resultPartitionConsumableNotifier =
- new RpcResultPartitionConsumableNotifier(
- jobMasterGateway,
- getRpcService().getScheduledExecutor(),
- taskManagerConfiguration.getRpcTimeout());
-
PartitionProducerStateChecker partitionStateChecker =
new RpcPartitionStateChecker(jobMasterGateway);
@@ -1746,7 +1735,6 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
taskManagerActions,
checkpointResponder,
aggregateManager,
- resultPartitionConsumableNotifier,
partitionStateChecker);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/rpc/RpcResultPartitionConsumableNotifier.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/rpc/RpcResultPartitionConsumableNotifier.java
deleted file mode 100644
index 6150cc2c6ef..00000000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/rpc/RpcResultPartitionConsumableNotifier.java
+++ /dev/null
@@ -1,73 +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.taskexecutor.rpc;
-
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.common.time.Time;
-import org.apache.flink.runtime.io.network.partition.ResultPartitionConsumableNotifier;
-import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
-import org.apache.flink.runtime.jobmaster.JobMasterGateway;
-import org.apache.flink.runtime.messages.Acknowledge;
-import org.apache.flink.runtime.taskmanager.TaskActions;
-import org.apache.flink.util.Preconditions;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executor;
-
-public class RpcResultPartitionConsumableNotifier implements ResultPartitionConsumableNotifier {
-
- private static final Logger LOG =
- LoggerFactory.getLogger(RpcResultPartitionConsumableNotifier.class);
-
- private final JobMasterGateway jobMasterGateway;
- private final Executor executor;
- private final Time timeout;
-
- public RpcResultPartitionConsumableNotifier(
- JobMasterGateway jobMasterGateway, Executor executor, Time timeout) {
- this.jobMasterGateway = Preconditions.checkNotNull(jobMasterGateway);
- this.executor = Preconditions.checkNotNull(executor);
- this.timeout = Preconditions.checkNotNull(timeout);
- }
-
- @Override
- public void notifyPartitionConsumable(
- JobID jobId, ResultPartitionID partitionId, final TaskActions taskActions) {
- CompletableFuture<Acknowledge> acknowledgeFuture =
- jobMasterGateway.notifyPartitionDataAvailable(partitionId, timeout);
-
- acknowledgeFuture.whenCompleteAsync(
- (Acknowledge ack, Throwable throwable) -> {
- if (throwable != null) {
- LOG.error(
- "Could not notify partition data available to JobManager.",
- throwable);
-
- taskActions.failExternally(
- new RuntimeException(
- "Could not notify partition data available to JobManager.",
- throwable));
- }
- },
- executor);
- }
-}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/ConsumableNotifyingResultPartitionWriterDecorator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/ConsumableNotifyingResultPartitionWriterDecorator.java
deleted file mode 100644
index 25f585b7b41..00000000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/ConsumableNotifyingResultPartitionWriterDecorator.java
+++ /dev/null
@@ -1,256 +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.taskmanager;
-
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.runtime.deployment.ResultPartitionDeploymentDescriptor;
-import org.apache.flink.runtime.event.AbstractEvent;
-import org.apache.flink.runtime.io.network.api.StopMode;
-import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
-import org.apache.flink.runtime.io.network.partition.BufferAvailabilityListener;
-import org.apache.flink.runtime.io.network.partition.CheckpointedResultPartition;
-import org.apache.flink.runtime.io.network.partition.CheckpointedResultSubpartition;
-import org.apache.flink.runtime.io.network.partition.ResultPartitionConsumableNotifier;
-import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
-import org.apache.flink.runtime.io.network.partition.ResultSubpartitionView;
-import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.Collection;
-import java.util.concurrent.CompletableFuture;
-
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
-/**
- * A utility to wrap a {@link ResultPartitionWriter} for that sends a "consumable" notification on
- * the first record.
- *
- * <p>Note: This is only needed for lazy batch scheduling and thus an artifact that can be removed
- * once we drop lazy-from-sources scheduling.
- */
-public class ConsumableNotifyingResultPartitionWriterDecorator {
-
- /**
- * Optionally decorate the ResultPartitionWriter to call {@link
- * ResultPartitionConsumableNotifier#notifyPartitionConsumable(JobID, ResultPartitionID,
- * TaskActions)} on the first record, iff {@link
- * ResultPartitionDeploymentDescriptor#notifyPartitionDataAvailable()} is true.
- */
- public static ResultPartitionWriter[] decorate(
- Collection<ResultPartitionDeploymentDescriptor> descs,
- ResultPartitionWriter[] partitionWriters,
- TaskActions taskActions,
- JobID jobId,
- ResultPartitionConsumableNotifier notifier) {
-
- ResultPartitionWriter[] consumableNotifyingPartitionWriters =
- new ResultPartitionWriter[partitionWriters.length];
- int counter = 0;
- for (ResultPartitionDeploymentDescriptor desc : descs) {
- if (desc.notifyPartitionDataAvailable() && desc.getPartitionType().isPipelined()) {
- consumableNotifyingPartitionWriters[counter] =
- new ConsumableNotifyingResultPartitionWriter(
- taskActions, jobId, partitionWriters[counter], notifier);
- } else {
- consumableNotifyingPartitionWriters[counter] = partitionWriters[counter];
- }
- counter++;
- }
- return consumableNotifyingPartitionWriters;
- }
-
- /** This is a utility class not meant to be instantiated. */
- private ConsumableNotifyingResultPartitionWriterDecorator() {}
-
- // ------------------------------------------------------------------------
- // wrapper class to send notification
- // ------------------------------------------------------------------------
-
- private static final class ConsumableNotifyingResultPartitionWriter
- implements ResultPartitionWriter, CheckpointedResultPartition {
-
- private final TaskActions taskActions;
-
- private final JobID jobId;
-
- private final ResultPartitionWriter partitionWriter;
-
- private final ResultPartitionConsumableNotifier partitionConsumableNotifier;
-
- private boolean hasNotifiedPipelinedConsumers;
-
- public ConsumableNotifyingResultPartitionWriter(
- TaskActions taskActions,
- JobID jobId,
- ResultPartitionWriter partitionWriter,
- ResultPartitionConsumableNotifier partitionConsumableNotifier) {
- this.taskActions = checkNotNull(taskActions);
- this.jobId = checkNotNull(jobId);
- this.partitionWriter = checkNotNull(partitionWriter);
- this.partitionConsumableNotifier = checkNotNull(partitionConsumableNotifier);
- }
-
- @Override
- public ResultPartitionID getPartitionId() {
- return partitionWriter.getPartitionId();
- }
-
- @Override
- public int getNumberOfSubpartitions() {
- return partitionWriter.getNumberOfSubpartitions();
- }
-
- @Override
- public int getNumTargetKeyGroups() {
- return partitionWriter.getNumTargetKeyGroups();
- }
-
- @Override
- public void setup() throws IOException {
- partitionWriter.setup();
- }
-
- @Override
- public void emitRecord(ByteBuffer record, int targetSubpartition) throws IOException {
- partitionWriter.emitRecord(record, targetSubpartition);
-
- notifyPipelinedConsumers();
- }
-
- @Override
- public void broadcastRecord(ByteBuffer record) throws IOException {
- partitionWriter.broadcastRecord(record);
-
- notifyPipelinedConsumers();
- }
-
- @Override
- public void broadcastEvent(AbstractEvent event, boolean isPriorityEvent)
- throws IOException {
- partitionWriter.broadcastEvent(event, isPriorityEvent);
-
- notifyPipelinedConsumers();
- }
-
- @Override
- public void notifyEndOfData(StopMode mode) throws IOException {
- partitionWriter.notifyEndOfData(mode);
- }
-
- @Override
- public CompletableFuture<Void> getAllDataProcessedFuture() {
- return partitionWriter.getAllDataProcessedFuture();
- }
-
- @Override
- public void setMetricGroup(TaskIOMetricGroup metrics) {
- partitionWriter.setMetricGroup(metrics);
- }
-
- @Override
- public ResultSubpartitionView createSubpartitionView(
- int index, BufferAvailabilityListener availabilityListener) throws IOException {
- return partitionWriter.createSubpartitionView(index, availabilityListener);
- }
-
- @Override
- public void flushAll() {
- partitionWriter.flushAll();
- }
-
- @Override
- public void flush(int subpartitionIndex) {
- partitionWriter.flush(subpartitionIndex);
- }
-
- @Override
- public void finish() throws IOException {
- partitionWriter.finish();
-
- notifyPipelinedConsumers();
- }
-
- @Override
- public boolean isFinished() {
- return partitionWriter.isFinished();
- }
-
- @Override
- public void release(Throwable cause) {
- partitionWriter.release(cause);
- }
-
- @Override
- public boolean isReleased() {
- return partitionWriter.isReleased();
- }
-
- @Override
- public void fail(Throwable throwable) {
- partitionWriter.fail(throwable);
- }
-
- @Override
- public CompletableFuture<?> getAvailableFuture() {
- return partitionWriter.getAvailableFuture();
- }
-
- @Override
- public void close() throws Exception {
- partitionWriter.close();
- }
-
- @Override
- public void finishReadRecoveredState(boolean notifyAndBlockOnCompletion)
- throws IOException {
- getCheckpointablePartition().finishReadRecoveredState(notifyAndBlockOnCompletion);
- }
-
- /**
- * Notifies pipelined consumers of this result partition once.
- *
- * <p>For PIPELINED {@link
- * org.apache.flink.runtime.io.network.partition.ResultPartitionType}s, this will trigger
- * the deployment of consuming tasks after the first buffer has been added.
- */
- private void notifyPipelinedConsumers() {
- if (!hasNotifiedPipelinedConsumers && !partitionWriter.isReleased()) {
- partitionConsumableNotifier.notifyPartitionConsumable(
- jobId, partitionWriter.getPartitionId(), taskActions);
-
- hasNotifiedPipelinedConsumers = true;
- }
- }
-
- @Override
- public CheckpointedResultSubpartition getCheckpointedSubpartition(int subpartitionIndex) {
- return getCheckpointablePartition().getCheckpointedSubpartition(subpartitionIndex);
- }
-
- private CheckpointedResultPartition getCheckpointablePartition() {
- if (partitionWriter instanceof CheckpointedResultPartition) {
- return (CheckpointedResultPartition) partitionWriter;
- } else {
- throw new IllegalStateException(
- "This partition is not checkpointable: " + partitionWriter);
- }
- }
- }
-}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
index 01b5da244f8..2a2ef06dd1f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
@@ -54,7 +54,6 @@ import org.apache.flink.runtime.io.network.NettyShuffleEnvironment;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
import org.apache.flink.runtime.io.network.partition.PartitionProducerStateProvider;
-import org.apache.flink.runtime.io.network.partition.ResultPartitionConsumableNotifier;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.io.network.partition.consumer.IndexedInputGate;
import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
@@ -211,7 +210,7 @@ public class Task
*/
private final SerializedValue<ExecutionConfig> serializedExecutionConfig;
- private final ResultPartitionWriter[] consumableNotifyingPartitionWriters;
+ private final ResultPartitionWriter[] partitionWriters;
private final IndexedInputGate[] inputGates;
@@ -321,7 +320,6 @@ public class Task
FileCache fileCache,
TaskManagerRuntimeInfo taskManagerConfig,
@Nonnull TaskMetricGroup metricGroup,
- ResultPartitionConsumableNotifier resultPartitionConsumableNotifier,
PartitionProducerStateChecker partitionProducerStateChecker,
Executor executor) {
@@ -399,13 +397,7 @@ public class Task
taskShuffleContext, resultPartitionDeploymentDescriptors)
.toArray(new ResultPartitionWriter[] {});
- this.consumableNotifyingPartitionWriters =
- ConsumableNotifyingResultPartitionWriterDecorator.decorate(
- resultPartitionDeploymentDescriptors,
- resultPartitionWriters,
- this,
- jobId,
- resultPartitionConsumableNotifier);
+ this.partitionWriters = resultPartitionWriters;
// consumed intermediate result partitions
final IndexedInputGate[] gates =
@@ -504,13 +496,13 @@ public class Task
public boolean isBackPressured() {
if (invokable == null
- || consumableNotifyingPartitionWriters.length == 0
+ || partitionWriters.length == 0
|| (executionState != ExecutionState.INITIALIZING
&& executionState != ExecutionState.RUNNING)) {
return false;
}
- for (int i = 0; i < consumableNotifyingPartitionWriters.length; ++i) {
- if (!consumableNotifyingPartitionWriters[i].isAvailable()) {
+ for (int i = 0; i < partitionWriters.length; ++i) {
+ if (!partitionWriters[i].isAvailable()) {
return true;
}
}
@@ -649,9 +641,9 @@ public class Task
LOG.debug("Registering task at network: {}.", this);
- setupPartitionsAndGates(consumableNotifyingPartitionWriters, inputGates);
+ setupPartitionsAndGates(partitionWriters, inputGates);
- for (ResultPartitionWriter partitionWriter : consumableNotifyingPartitionWriters) {
+ for (ResultPartitionWriter partitionWriter : partitionWriters) {
taskEventDispatcher.registerPartition(partitionWriter.getPartitionId());
}
@@ -703,7 +695,7 @@ public class Task
kvStateRegistry,
inputSplitProvider,
distributedCacheEntries,
- consumableNotifyingPartitionWriters,
+ partitionWriters,
inputGates,
taskEventDispatcher,
checkpointResponder,
@@ -751,7 +743,7 @@ public class Task
// ----------------------------------------------------------------
// finish the produced partitions. if this fails, we consider the execution failed.
- for (ResultPartitionWriter partitionWriter : consumableNotifyingPartitionWriters) {
+ for (ResultPartitionWriter partitionWriter : partitionWriters) {
if (partitionWriter != null) {
partitionWriter.finish();
}
@@ -976,7 +968,7 @@ public class Task
taskNameWithSubtask,
getExecutionState());
- for (ResultPartitionWriter partitionWriter : consumableNotifyingPartitionWriters) {
+ for (ResultPartitionWriter partitionWriter : partitionWriters) {
taskEventDispatcher.unregisterPartition(partitionWriter.getPartitionId());
}
@@ -995,13 +987,13 @@ public class Task
}
private void failAllResultPartitions() {
- for (ResultPartitionWriter partitionWriter : consumableNotifyingPartitionWriters) {
+ for (ResultPartitionWriter partitionWriter : partitionWriters) {
partitionWriter.fail(getFailureCause());
}
}
private void closeAllResultPartitions() {
- for (ResultPartitionWriter partitionWriter : consumableNotifyingPartitionWriters) {
+ for (ResultPartitionWriter partitionWriter : partitionWriters) {
try {
partitionWriter.close();
} catch (Throwable t) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptorTest.java
index 39e91a45a1f..d2f44943f41 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptorTest.java
@@ -108,7 +108,7 @@ public class ResultPartitionDeploymentDescriptorTest extends TestLogger {
ShuffleDescriptor shuffleDescriptor) throws IOException {
ResultPartitionDeploymentDescriptor orig =
new ResultPartitionDeploymentDescriptor(
- partitionDescriptor, shuffleDescriptor, numberOfSubpartitions, true);
+ partitionDescriptor, shuffleDescriptor, numberOfSubpartitions);
ResultPartitionDeploymentDescriptor copy = CommonTestUtils.createCopySerializable(orig);
verifyResultPartitionDeploymentDescriptorCopy(copy);
return copy;
@@ -121,6 +121,5 @@ public class ResultPartitionDeploymentDescriptorTest extends TestLogger {
assertThat(copy.getPartitionId(), is(partitionId));
assertThat(copy.getPartitionType(), is(partitionType));
assertThat(copy.getNumberOfSubpartitions(), is(numberOfSubpartitions));
- assertThat(copy.notifyPartitionDataAvailable(), is(true));
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ShuffleDescriptorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ShuffleDescriptorTest.java
index 89cc242dd3c..bb0e96c9250 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ShuffleDescriptorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ShuffleDescriptorTest.java
@@ -214,7 +214,6 @@ public class ShuffleDescriptorTest extends TestLogger {
.registerPartitionWithProducer(
jobID, partitionDescriptor, producerDescriptor)
.get();
- return new ResultPartitionDeploymentDescriptor(
- partitionDescriptor, shuffleDescriptor, 1, true);
+ return new ResultPartitionDeploymentDescriptor(partitionDescriptor, shuffleDescriptor, 1);
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphDeploymentTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphDeploymentTest.java
index 381b8d4d023..f5c222e91c5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphDeploymentTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphDeploymentTest.java
@@ -197,7 +197,7 @@ public class DefaultExecutionGraphDeploymentTest extends TestLogger {
vertex.getCurrentExecutionAttempt().transitionState(ExecutionState.SCHEDULED);
vertex.getCurrentExecutionAttempt()
- .registerProducedPartitions(slot.getTaskManagerLocation(), true)
+ .registerProducedPartitions(slot.getTaskManagerLocation())
.get();
vertex.deployToSlot(slot);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphDeploymentWithSmallBlobCacheSizeLimitTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphDeploymentWithSmallBlobCacheSizeLimitTest.java
index e2aa7a5809d..2ea30d6df44 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphDeploymentWithSmallBlobCacheSizeLimitTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphDeploymentWithSmallBlobCacheSizeLimitTest.java
@@ -132,7 +132,7 @@ public class DefaultExecutionGraphDeploymentWithSmallBlobCacheSizeLimitTest
.createTestingLogicalSlot();
final Execution execution = ev.getCurrentExecutionAttempt();
execution.transitionState(ExecutionState.SCHEDULED);
- execution.registerProducedPartitions(slot.getTaskManagerLocation(), true).get();
+ execution.registerProducedPartitions(slot.getTaskManagerLocation()).get();
ev.deployToSlot(slot);
assertEquals(ExecutionState.DEPLOYING, ev.getExecutionState());
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphPartitionReleaseTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphPartitionReleaseTest.java
index 79e5495380f..0ca4f106f95 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphPartitionReleaseTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphPartitionReleaseTest.java
@@ -25,7 +25,6 @@ import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
import org.apache.flink.runtime.io.network.partition.TestingJobMasterPartitionTracker;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
-import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobGraphTestUtils;
import org.apache.flink.runtime.jobgraph.JobVertex;
@@ -199,12 +198,6 @@ public class ExecutionGraphPartitionReleaseTest extends TestLogger {
getCurrentExecution(operator1Vertex, executionGraph);
// finish o1 and schedule the consumers (o2,o3); this should not result in any
// release calls since not all operators of the pipelined region are finished
- for (final IntermediateResultPartitionID partitionId :
- operator1Execution.getVertex().getProducedPartitions().keySet()) {
- scheduler.notifyPartitionDataAvailable(
- new ResultPartitionID(
- partitionId, operator1Execution.getAttemptId()));
- }
scheduler.updateTaskExecutionState(
new TaskExecutionState(
operator1Execution.getAttemptId(), ExecutionState.FINISHED));
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphVariousFailuesTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphVariousFailuesTest.java
deleted file mode 100644
index eaf80d67bb6..00000000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphVariousFailuesTest.java
+++ /dev/null
@@ -1,86 +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.executiongraph;
-
-import org.apache.flink.api.common.JobStatus;
-import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter;
-import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
-import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
-import org.apache.flink.runtime.jobgraph.JobGraphTestUtils;
-import org.apache.flink.runtime.scheduler.SchedulerBase;
-import org.apache.flink.runtime.scheduler.SchedulerTestingUtils;
-import org.apache.flink.testutils.TestingUtils;
-import org.apache.flink.testutils.executor.TestExecutorResource;
-import org.apache.flink.util.TestLogger;
-
-import org.junit.ClassRule;
-import org.junit.Test;
-
-import java.util.concurrent.ScheduledExecutorService;
-
-import static org.hamcrest.Matchers.containsString;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.fail;
-
-public class ExecutionGraphVariousFailuesTest extends TestLogger {
-
- @ClassRule
- public static final TestExecutorResource<ScheduledExecutorService> EXECUTOR_RESOURCE =
- TestingUtils.defaultExecutorResource();
-
- /**
- * Tests that a failing notifyPartitionDataAvailable call with a non-existing execution attempt
- * id, will not fail the execution graph.
- */
- @Test
- public void testFailingNotifyPartitionDataAvailable() throws Exception {
- final SchedulerBase scheduler =
- new SchedulerTestingUtils.DefaultSchedulerBuilder(
- JobGraphTestUtils.emptyJobGraph(),
- ComponentMainThreadExecutorServiceAdapter.forMainThread(),
- EXECUTOR_RESOURCE.getExecutor())
- .build();
- scheduler.startScheduling();
-
- final ExecutionGraph eg = scheduler.getExecutionGraph();
-
- assertEquals(JobStatus.RUNNING, eg.getState());
- ExecutionGraphTestUtils.switchAllVerticesToRunning(eg);
-
- IntermediateResultPartitionID intermediateResultPartitionId =
- new IntermediateResultPartitionID();
- ExecutionAttemptID producerId = new ExecutionAttemptID();
- ResultPartitionID resultPartitionId =
- new ResultPartitionID(intermediateResultPartitionId, producerId);
-
- // The execution attempt id does not exist and thus the notifyPartitionDataAvailable call
- // should fail
-
- try {
- scheduler.notifyPartitionDataAvailable(resultPartitionId);
- fail("Error expected.");
- } catch (IllegalStateException e) {
- // we've expected this exception to occur
- assertThat(e.getMessage(), containsString("Cannot find execution for execution Id"));
- }
-
- assertEquals(JobStatus.RUNNING, eg.getState());
- }
-}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/RemoveCachedShuffleDescriptorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/RemoveCachedShuffleDescriptorTest.java
index f81c7bc7f1d..2ba942dcecc 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/RemoveCachedShuffleDescriptorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/RemoveCachedShuffleDescriptorTest.java
@@ -413,7 +413,7 @@ public class RemoveCachedShuffleDescriptorTest extends TestLogger {
LogicalSlot slot = slotBuilder.createTestingLogicalSlot();
Execution execution = vertex.getCurrentExecutionAttempt();
- execution.registerProducedPartitions(slot.getTaskManagerLocation(), true).get();
+ execution.registerProducedPartitions(slot.getTaskManagerLocation()).get();
execution.transitionState(ExecutionState.SCHEDULED);
vertex.tryAssignResource(slot);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/AbstractPartitionTrackerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/AbstractPartitionTrackerTest.java
index 0324d482e8f..2f44048b25a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/AbstractPartitionTrackerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/AbstractPartitionTrackerTest.java
@@ -93,8 +93,7 @@ public class AbstractPartitionTrackerTest extends TestLogger {
: Optional.empty();
}
},
- 1,
- true);
+ 1);
}
private static class TestPartitionTracker
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/NoOpResultPartitionConsumableNotifier.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/NoOpResultPartitionConsumableNotifier.java
deleted file mode 100644
index c4d9b816840..00000000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/NoOpResultPartitionConsumableNotifier.java
+++ /dev/null
@@ -1,29 +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.io.network.partition;
-
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.runtime.taskmanager.TaskActions;
-
-/** Test implementation of {@link ResultPartitionConsumableNotifier}. */
-public class NoOpResultPartitionConsumableNotifier implements ResultPartitionConsumableNotifier {
- @Override
- public void notifyPartitionConsumable(
- JobID jobId, ResultPartitionID partitionId, TaskActions taskActions) {}
-}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartitionTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartitionTestUtils.java
index 9f9d09ecfda..ffea47dfb41 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartitionTestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartitionTestUtils.java
@@ -133,8 +133,7 @@ public enum PartitionTestUtils {
.setPartitionId(shuffleDescriptor.getResultPartitionID().getPartitionId())
.setPartitionType(partitionType)
.build();
- return new ResultPartitionDeploymentDescriptor(
- partitionDescriptor, shuffleDescriptor, 1, true);
+ return new ResultPartitionDeploymentDescriptor(partitionDescriptor, shuffleDescriptor, 1);
}
public static PartitionedFile createPartitionedFile(
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java
index 7c51f91e09f..0c54e3878a0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionFactoryTest.java
@@ -133,8 +133,7 @@ public class ResultPartitionFactoryTest extends TestLogger {
.setPartitionType(partitionType)
.build(),
NettyShuffleDescriptorBuilder.newBuilder().buildLocal(),
- 1,
- true);
+ 1);
// guard our test assumptions
assertEquals(1, descriptor.getNumberOfSubpartitions());
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionTest.java
index 3299ed245ef..4fbaf198b52 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionTest.java
@@ -18,7 +18,6 @@
package org.apache.flink.runtime.io.network.partition;
-import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.checkpoint.channel.ResultSubpartitionInfo;
import org.apache.flink.runtime.io.disk.FileChannelManager;
import org.apache.flink.runtime.io.disk.FileChannelManagerImpl;
@@ -28,15 +27,10 @@ import org.apache.flink.runtime.io.network.api.EndOfData;
import org.apache.flink.runtime.io.network.api.EndOfPartitionEvent;
import org.apache.flink.runtime.io.network.api.StopMode;
import org.apache.flink.runtime.io.network.api.serialization.EventSerializer;
-import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.buffer.BufferPool;
import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool;
-import org.apache.flink.runtime.taskmanager.ConsumableNotifyingResultPartitionWriterDecorator;
-import org.apache.flink.runtime.taskmanager.NoOpTaskActions;
-import org.apache.flink.runtime.taskmanager.TaskActions;
import org.apache.flink.runtime.util.EnvironmentInformation;
-import org.apache.flink.util.concurrent.FutureConsumerWithException;
import org.hamcrest.Matchers;
import org.junit.AfterClass;
@@ -46,7 +40,6 @@ import org.junit.Test;
import java.io.IOException;
import java.nio.ByteBuffer;
-import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
@@ -103,58 +96,6 @@ public class ResultPartitionTest {
}
}
- /** Tests notifyPartitionDataAvailable behaviour depending on the relevant flags. */
- @Test
- public void testNotifyPartitionDataAvailable() throws Exception {
- FutureConsumerWithException[] notificationCalls =
- new FutureConsumerWithException[] {
- writer -> ((ResultPartitionWriter) writer).finish(),
- writer ->
- ((ResultPartitionWriter) writer)
- .emitRecord(ByteBuffer.allocate(bufferSize), 0),
- writer ->
- ((ResultPartitionWriter) writer)
- .broadcastEvent(EndOfPartitionEvent.INSTANCE, false),
- writer ->
- ((ResultPartitionWriter) writer)
- .broadcastRecord(ByteBuffer.allocate(bufferSize))
- };
-
- for (FutureConsumerWithException notificationCall : notificationCalls) {
- testNotifyPartitionDataAvailable(notificationCall);
- }
- }
-
- private void testNotifyPartitionDataAvailable(
- FutureConsumerWithException<ResultPartitionWriter, Exception> notificationCall)
- throws Exception {
- JobID jobId = new JobID();
- TaskActions taskActions = new NoOpTaskActions();
-
- {
- // Pipelined, send message => notify
- TestResultPartitionConsumableNotifier notifier =
- new TestResultPartitionConsumableNotifier();
- ResultPartitionWriter consumableNotifyingPartitionWriter =
- createConsumableNotifyingResultPartitionWriter(
- ResultPartitionType.PIPELINED, taskActions, jobId, notifier);
- notificationCall.accept(consumableNotifyingPartitionWriter);
- notifier.check(
- jobId, consumableNotifyingPartitionWriter.getPartitionId(), taskActions, 1);
- }
-
- {
- // Blocking, send message => don't notify
- TestResultPartitionConsumableNotifier notifier =
- new TestResultPartitionConsumableNotifier();
- ResultPartitionWriter partition =
- createConsumableNotifyingResultPartitionWriter(
- ResultPartitionType.BLOCKING, taskActions, jobId, notifier);
- notificationCall.accept(partition);
- notifier.check(null, null, null, 0);
- }
- }
-
@Test
public void testAddOnFinishedPipelinedPartition() throws Exception {
testAddOnFinishedPartition(ResultPartitionType.PIPELINED);
@@ -202,24 +143,13 @@ public class ResultPartitionTest {
*/
private void testAddOnFinishedPartition(final ResultPartitionType partitionType)
throws Exception {
- TestResultPartitionConsumableNotifier notifier =
- new TestResultPartitionConsumableNotifier();
BufferWritingResultPartition bufferWritingResultPartition =
createResultPartition(partitionType);
- ResultPartitionWriter partitionWriter =
- ConsumableNotifyingResultPartitionWriterDecorator.decorate(
- Collections.singleton(
- PartitionTestUtils.createPartitionDeploymentDescriptor(
- partitionType)),
- new ResultPartitionWriter[] {bufferWritingResultPartition},
- new NoOpTaskActions(),
- new JobID(),
- notifier)[0];
+
try {
- partitionWriter.finish();
- notifier.reset();
+ bufferWritingResultPartition.finish();
// partitionWriter.emitRecord() should fail
- partitionWriter.emitRecord(ByteBuffer.allocate(bufferSize), 0);
+ bufferWritingResultPartition.emitRecord(ByteBuffer.allocate(bufferSize), 0);
} catch (IllegalStateException e) {
// expected => ignored
} finally {
@@ -228,8 +158,6 @@ public class ResultPartitionTest {
assertEquals(
0,
bufferWritingResultPartition.getBufferPool().bestEffortGetNumOfUsedBuffers());
- // should not have notified either
- notifier.check(null, null, null, 0);
}
}
@@ -249,23 +177,13 @@ public class ResultPartitionTest {
* @param partitionType the result partition type to set up
*/
private void testAddOnReleasedPartition(ResultPartitionType partitionType) throws Exception {
- TestResultPartitionConsumableNotifier notifier =
- new TestResultPartitionConsumableNotifier();
BufferWritingResultPartition bufferWritingResultPartition =
createResultPartition(partitionType);
- ResultPartitionWriter partitionWriter =
- ConsumableNotifyingResultPartitionWriterDecorator.decorate(
- Collections.singleton(
- PartitionTestUtils.createPartitionDeploymentDescriptor(
- partitionType)),
- new ResultPartitionWriter[] {bufferWritingResultPartition},
- new NoOpTaskActions(),
- new JobID(),
- notifier)[0];
+
try {
- partitionWriter.release(null);
+ bufferWritingResultPartition.release(null);
// partitionWriter.emitRecord() should silently drop the given record
- partitionWriter.emitRecord(ByteBuffer.allocate(bufferSize), 0);
+ bufferWritingResultPartition.emitRecord(ByteBuffer.allocate(bufferSize), 0);
} finally {
assertEquals(1, bufferWritingResultPartition.numBuffersOut.getCount());
assertEquals(bufferSize, bufferWritingResultPartition.numBytesOut.getCount());
@@ -273,8 +191,6 @@ public class ResultPartitionTest {
assertEquals(
0,
bufferWritingResultPartition.getBufferPool().bestEffortGetNumOfUsedBuffers());
- // should not have notified either
- notifier.check(null, null, null, 0);
}
}
@@ -313,34 +229,18 @@ public class ResultPartitionTest {
* @param partitionType the result partition type to set up
*/
private void testAddOnPartition(final ResultPartitionType partitionType) throws Exception {
- TestResultPartitionConsumableNotifier notifier =
- new TestResultPartitionConsumableNotifier();
- JobID jobId = new JobID();
- TaskActions taskActions = new NoOpTaskActions();
BufferWritingResultPartition bufferWritingResultPartition =
createResultPartition(partitionType);
- ResultPartitionWriter partitionWriter =
- ConsumableNotifyingResultPartitionWriterDecorator.decorate(
- Collections.singleton(
- PartitionTestUtils.createPartitionDeploymentDescriptor(
- partitionType)),
- new ResultPartitionWriter[] {bufferWritingResultPartition},
- taskActions,
- jobId,
- notifier)[0];
+
try {
// partitionWriter.emitRecord() will allocate a new buffer and copies the record to it
- partitionWriter.emitRecord(ByteBuffer.allocate(bufferSize), 0);
+ bufferWritingResultPartition.emitRecord(ByteBuffer.allocate(bufferSize), 0);
} finally {
assertEquals(1, bufferWritingResultPartition.numBuffersOut.getCount());
assertEquals(bufferSize, bufferWritingResultPartition.numBytesOut.getCount());
assertEquals(
1,
bufferWritingResultPartition.getBufferPool().bestEffortGetNumOfUsedBuffers());
- // should have been notified for pipelined partitions
- if (partitionType.isPipelined()) {
- notifier.check(jobId, partitionWriter.getPartitionId(), taskActions, 1);
- }
}
}
@@ -455,22 +355,6 @@ public class ResultPartitionTest {
}
}
- private ResultPartitionWriter createConsumableNotifyingResultPartitionWriter(
- ResultPartitionType partitionType,
- TaskActions taskActions,
- JobID jobId,
- ResultPartitionConsumableNotifier notifier)
- throws IOException {
- ResultPartition partition = createResultPartition(partitionType);
- return ConsumableNotifyingResultPartitionWriterDecorator.decorate(
- Collections.singleton(
- PartitionTestUtils.createPartitionDeploymentDescriptor(partitionType)),
- new ResultPartitionWriter[] {partition},
- taskActions,
- jobId,
- notifier)[0];
- }
-
private BufferWritingResultPartition createResultPartition(ResultPartitionType partitionType)
throws IOException {
NettyShuffleEnvironment network =
@@ -834,28 +718,15 @@ public class ResultPartitionTest {
}
private void testNumBytesProducedCounter(boolean isBroadcast) throws IOException {
- TestResultPartitionConsumableNotifier notifier =
- new TestResultPartitionConsumableNotifier();
- JobID jobId = new JobID();
- TaskActions taskActions = new NoOpTaskActions();
BufferWritingResultPartition bufferWritingResultPartition =
createResultPartition(ResultPartitionType.BLOCKING);
- ResultPartitionWriter partitionWriter =
- ConsumableNotifyingResultPartitionWriterDecorator.decorate(
- Collections.singleton(
- PartitionTestUtils.createPartitionDeploymentDescriptor(
- ResultPartitionType.BLOCKING)),
- new ResultPartitionWriter[] {bufferWritingResultPartition},
- taskActions,
- jobId,
- notifier)[0];
if (isBroadcast) {
- partitionWriter.broadcastRecord(ByteBuffer.allocate(bufferSize));
+ bufferWritingResultPartition.broadcastRecord(ByteBuffer.allocate(bufferSize));
assertEquals(bufferSize, bufferWritingResultPartition.numBytesProduced.getCount());
assertEquals(2 * bufferSize, bufferWritingResultPartition.numBytesOut.getCount());
} else {
- partitionWriter.emitRecord(ByteBuffer.allocate(bufferSize), 0);
+ bufferWritingResultPartition.emitRecord(ByteBuffer.allocate(bufferSize), 0);
assertEquals(bufferSize, bufferWritingResultPartition.numBytesProduced.getCount());
assertEquals(bufferSize, bufferWritingResultPartition.numBytesOut.getCount());
}
@@ -936,39 +807,4 @@ public class ResultPartitionTest {
assertEquals(7, subpartition1.pollBuffer().buffer().getSize());
assertEquals(0, bufferWritingResultPartition.getSizeOfQueuedBuffersUnsafe());
}
-
- private static class TestResultPartitionConsumableNotifier
- implements ResultPartitionConsumableNotifier {
- private JobID jobID;
- private ResultPartitionID partitionID;
- private TaskActions taskActions;
- private int numNotification;
-
- @Override
- public void notifyPartitionConsumable(
- JobID jobID, ResultPartitionID partitionID, TaskActions taskActions) {
- ++numNotification;
- this.jobID = jobID;
- this.partitionID = partitionID;
- this.taskActions = taskActions;
- }
-
- private void check(
- JobID jobID,
- ResultPartitionID partitionID,
- TaskActions taskActions,
- int numNotification) {
- assertEquals(jobID, this.jobID);
- assertEquals(partitionID, this.partitionID);
- assertEquals(taskActions, this.taskActions);
- assertEquals(numNotification, this.numNotification);
- }
-
- private void reset() {
- jobID = null;
- partitionID = null;
- taskActions = null;
- numNotification = 0;
- }
- }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java
index bee9b191920..33e1fdb0261 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java
@@ -96,10 +96,6 @@ public class TestingJobMasterGateway implements JobMasterGateway {
IntermediateDataSetID, ResultPartitionID, CompletableFuture<ExecutionState>>
requestPartitionStateFunction;
- @Nonnull
- private final Function<ResultPartitionID, CompletableFuture<Acknowledge>>
- notifyPartitionDataAvailableFunction;
-
@Nonnull
private final Function<ResourceID, CompletableFuture<Acknowledge>>
disconnectTaskManagerFunction;
@@ -207,9 +203,6 @@ public class TestingJobMasterGateway implements JobMasterGateway {
ResultPartitionID,
CompletableFuture<ExecutionState>>
requestPartitionStateFunction,
- @Nonnull
- Function<ResultPartitionID, CompletableFuture<Acknowledge>>
- notifyPartitionDataAvailableFunction,
@Nonnull
Function<ResourceID, CompletableFuture<Acknowledge>>
disconnectTaskManagerFunction,
@@ -297,7 +290,6 @@ public class TestingJobMasterGateway implements JobMasterGateway {
this.updateTaskExecutionStateFunction = updateTaskExecutionStateFunction;
this.requestNextInputSplitFunction = requestNextInputSplitFunction;
this.requestPartitionStateFunction = requestPartitionStateFunction;
- this.notifyPartitionDataAvailableFunction = notifyPartitionDataAvailableFunction;
this.disconnectTaskManagerFunction = disconnectTaskManagerFunction;
this.disconnectResourceManagerConsumer = disconnectResourceManagerConsumer;
this.offerSlotsFunction = offerSlotsFunction;
@@ -346,12 +338,6 @@ public class TestingJobMasterGateway implements JobMasterGateway {
return requestPartitionStateFunction.apply(intermediateResultId, partitionId);
}
- @Override
- public CompletableFuture<Acknowledge> notifyPartitionDataAvailable(
- ResultPartitionID partitionID, Time timeout) {
- return notifyPartitionDataAvailableFunction.apply(partitionID);
- }
-
@Override
public CompletableFuture<Acknowledge> disconnectTaskManager(
ResourceID resourceID, Exception cause) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGatewayBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGatewayBuilder.java
index cc7588b6505..276dd89d24e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGatewayBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGatewayBuilder.java
@@ -90,9 +90,6 @@ public class TestingJobMasterGatewayBuilder {
requestPartitionStateFunction =
(ignoredA, ignoredB) ->
CompletableFuture.completedFuture(ExecutionState.RUNNING);
- private Function<ResultPartitionID, CompletableFuture<Acknowledge>>
- notifyPartitionDataAvailableFunction =
- ignored -> CompletableFuture.completedFuture(Acknowledge.get());
private Function<ResourceID, CompletableFuture<Acknowledge>> disconnectTaskManagerFunction =
ignored -> CompletableFuture.completedFuture(Acknowledge.get());
private Consumer<ResourceManagerId> disconnectResourceManagerConsumer = ignored -> {};
@@ -212,13 +209,6 @@ public class TestingJobMasterGatewayBuilder {
return this;
}
- public TestingJobMasterGatewayBuilder setNotifyPartitionDataAvailableFunction(
- Function<ResultPartitionID, CompletableFuture<Acknowledge>>
- notifyPartitionDataAvailableFunction) {
- this.notifyPartitionDataAvailableFunction = notifyPartitionDataAvailableFunction;
- return this;
- }
-
public TestingJobMasterGatewayBuilder setDisconnectTaskManagerFunction(
Function<ResourceID, CompletableFuture<Acknowledge>> disconnectTaskManagerFunction) {
this.disconnectTaskManagerFunction = disconnectTaskManagerFunction;
@@ -400,7 +390,6 @@ public class TestingJobMasterGatewayBuilder {
updateTaskExecutionStateFunction,
requestNextInputSplitFunction,
requestPartitionStateFunction,
- notifyPartitionDataAvailableFunction,
disconnectTaskManagerFunction,
disconnectResourceManagerConsumer,
offerSlotsFunction,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DeploymentHandleTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DeploymentHandleTest.java
index 460a9d5135f..3ad93bbfea3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DeploymentHandleTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DeploymentHandleTest.java
@@ -47,10 +47,6 @@ public class DeploymentHandleTest {
private static final ExecutionVertexVersion TEST_EXECUTION_VERTEX_VERSION =
new ExecutionVertexVersion(TEST_EXECUTION_VERTEX_ID, 0);
- private static final ExecutionVertexDeploymentOption TEST_DEPLOYMENT_OPTION =
- new ExecutionVertexDeploymentOption(
- TEST_EXECUTION_VERTEX_ID, new DeploymentOption(true));
-
private CompletableFuture<LogicalSlot> logicalSlotFuture;
private DeploymentHandle deploymentHandle;
@@ -61,10 +57,7 @@ public class DeploymentHandleTest {
final SlotExecutionVertexAssignment slotExecutionVertexAssignment =
new SlotExecutionVertexAssignment(TEST_EXECUTION_VERTEX_ID, logicalSlotFuture);
deploymentHandle =
- new DeploymentHandle(
- TEST_EXECUTION_VERTEX_VERSION,
- TEST_DEPLOYMENT_OPTION,
- slotExecutionVertexAssignment);
+ new DeploymentHandle(TEST_EXECUTION_VERTEX_VERSION, slotExecutionVertexAssignment);
}
@Test
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/TestingSchedulerNG.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/TestingSchedulerNG.java
index 18680526860..43386a2e705 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/TestingSchedulerNG.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/TestingSchedulerNG.java
@@ -126,11 +126,6 @@ public class TestingSchedulerNG implements SchedulerNG {
return null;
}
- @Override
- public void notifyPartitionDataAvailable(ResultPartitionID partitionID) {
- failOperation();
- }
-
@Override
public ExecutionGraphInfo requestJob() {
failOperation();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/StateTrackingMockExecutionGraph.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/StateTrackingMockExecutionGraph.java
index 222bf986a8a..b354dddc2b5 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/StateTrackingMockExecutionGraph.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/StateTrackingMockExecutionGraph.java
@@ -46,7 +46,6 @@ import org.apache.flink.runtime.executiongraph.IntermediateResult;
import org.apache.flink.runtime.executiongraph.JobStatusListener;
import org.apache.flink.runtime.executiongraph.TaskExecutionStateTransition;
import org.apache.flink.runtime.executiongraph.failover.flip1.ResultPartitionAvailabilityChecker;
-import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobVertexID;
@@ -334,11 +333,6 @@ class StateTrackingMockExecutionGraph implements ExecutionGraph {
throw new UnsupportedOperationException();
}
- @Override
- public void notifyPartitionDataAvailable(ResultPartitionID partitionId) {
- throw new UnsupportedOperationException();
- }
-
@Override
public void updateAccumulators(AccumulatorSnapshot accumulatorSnapshot) {
throw new UnsupportedOperationException();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/SchedulerBenchmarkUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/SchedulerBenchmarkUtils.java
index 1cffa39e075..ed5ef5f7eea 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/SchedulerBenchmarkUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/SchedulerBenchmarkUtils.java
@@ -118,18 +118,14 @@ public class SchedulerBenchmarkUtils {
public static void deployTasks(
ExecutionGraph executionGraph,
JobVertexID jobVertexID,
- TestingLogicalSlotBuilder slotBuilder,
- boolean sendScheduleOrUpdateConsumersMessage)
+ TestingLogicalSlotBuilder slotBuilder)
throws JobException, ExecutionException, InterruptedException {
for (ExecutionVertex vertex : executionGraph.getJobVertex(jobVertexID).getTaskVertices()) {
LogicalSlot slot = slotBuilder.createTestingLogicalSlot();
Execution execution = vertex.getCurrentExecutionAttempt();
execution.transitionState(ExecutionState.SCHEDULED);
- execution
- .registerProducedPartitions(
- slot.getTaskManagerLocation(), sendScheduleOrUpdateConsumersMessage)
- .get();
+ execution.registerProducedPartitions(slot.getTaskManagerLocation()).get();
assignResourceAndDeploy(vertex, slot);
}
}
@@ -142,7 +138,7 @@ public class SchedulerBenchmarkUtils {
LogicalSlot slot = slotBuilder.createTestingLogicalSlot();
Execution execution = vertex.getCurrentExecutionAttempt();
execution.transitionState(ExecutionState.SCHEDULED);
- execution.registerProducedPartitions(slot.getTaskManagerLocation(), true).get();
+ execution.registerProducedPartitions(slot.getTaskManagerLocation()).get();
assignResourceAndDeploy(vertex, slot);
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/deploying/DeployingTasksBenchmarkBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/deploying/DeployingTasksBenchmarkBase.java
index b5aed705d68..e53d4351dbb 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/deploying/DeployingTasksBenchmarkBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/deploying/DeployingTasksBenchmarkBase.java
@@ -54,7 +54,7 @@ public class DeployingTasksBenchmarkBase extends SchedulerBenchmarkBase {
for (ExecutionVertex ev : ejv.getTaskVertices()) {
final LogicalSlot slot = slotBuilder.createTestingLogicalSlot();
final Execution execution = ev.getCurrentExecutionAttempt();
- execution.registerProducedPartitions(slot.getTaskManagerLocation(), true).get();
+ execution.registerProducedPartitions(slot.getTaskManagerLocation()).get();
if (!execution.tryAssignResource(slot)) {
throw new RuntimeException("Error when assigning slot to execution.");
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/failover/RegionToRestartInBatchJobBenchmark.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/failover/RegionToRestartInBatchJobBenchmark.java
index 879e7a3dac7..41cafe4610f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/failover/RegionToRestartInBatchJobBenchmark.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/failover/RegionToRestartInBatchJobBenchmark.java
@@ -45,11 +45,11 @@ public class RegionToRestartInBatchJobBenchmark extends FailoverBenchmarkBase {
final TestingLogicalSlotBuilder slotBuilder = new TestingLogicalSlotBuilder();
- deployTasks(executionGraph, source.getID(), slotBuilder, true);
+ deployTasks(executionGraph, source.getID(), slotBuilder);
transitionTaskStatus(executionGraph, source.getID(), ExecutionState.FINISHED);
- deployTasks(executionGraph, sink.getID(), slotBuilder, true);
+ deployTasks(executionGraph, sink.getID(), slotBuilder);
}
public Set<ExecutionVertexID> calculateRegionToRestart() {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/partitionrelease/PartitionReleaseInBatchJobBenchmark.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/partitionrelease/PartitionReleaseInBatchJobBenchmark.java
index b23b32f848e..cee96e649c4 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/partitionrelease/PartitionReleaseInBatchJobBenchmark.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/partitionrelease/PartitionReleaseInBatchJobBenchmark.java
@@ -56,11 +56,11 @@ public class PartitionReleaseInBatchJobBenchmark extends SchedulerBenchmarkBase
final TestingLogicalSlotBuilder slotBuilder = new TestingLogicalSlotBuilder();
- deployTasks(executionGraph, source.getID(), slotBuilder, true);
+ deployTasks(executionGraph, source.getID(), slotBuilder);
transitionTaskStatus(executionGraph, source.getID(), ExecutionState.FINISHED);
- deployTasks(executionGraph, sink.getID(), slotBuilder, true);
+ deployTasks(executionGraph, sink.getID(), slotBuilder);
}
public void partitionRelease() {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/strategy/PipelinedRegionSchedulingStrategyTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/strategy/PipelinedRegionSchedulingStrategyTest.java
index 670293e9b73..c57f14779a0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/strategy/PipelinedRegionSchedulingStrategyTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/strategy/PipelinedRegionSchedulingStrategyTest.java
@@ -28,7 +28,6 @@ import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobGraphBuilder;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
-import org.apache.flink.runtime.scheduler.ExecutionVertexDeploymentOption;
import org.apache.flink.testutils.TestingUtils;
import org.apache.flink.testutils.executor.TestExecutorResource;
import org.apache.flink.util.TestLogger;
@@ -304,12 +303,12 @@ public class PipelinedRegionSchedulingStrategyTest extends TestLogger {
PipelinedRegionSchedulingStrategy schedulingStrategy = startScheduling(schedulingTopology);
assertEquals(1, testingSchedulerOperation.getScheduledVertices().size());
- final List<ExecutionVertexDeploymentOption> deploymentOptions1 =
+ final List<ExecutionVertexID> scheduledVertices1 =
testingSchedulerOperation.getScheduledVertices().get(0);
- assertEquals(5, deploymentOptions1.size());
+ assertEquals(5, scheduledVertices1.size());
- for (ExecutionVertexDeploymentOption deploymentOption : deploymentOptions1) {
- assertTrue(region1.contains(deploymentOption.getExecutionVertexId()));
+ for (ExecutionVertexID vertexId : scheduledVertices1) {
+ assertTrue(region1.contains(vertexId));
}
// Test whether the region 2 is scheduled correctly when region 1 is finished
@@ -318,12 +317,12 @@ public class PipelinedRegionSchedulingStrategyTest extends TestLogger {
schedulingStrategy.onExecutionStateChange(v22.getID(), ExecutionState.FINISHED);
assertEquals(2, testingSchedulerOperation.getScheduledVertices().size());
- final List<ExecutionVertexDeploymentOption> deploymentOptions2 =
+ final List<ExecutionVertexID> scheduledVertices2 =
testingSchedulerOperation.getScheduledVertices().get(1);
- assertEquals(4, deploymentOptions2.size());
+ assertEquals(4, scheduledVertices2.size());
- for (ExecutionVertexDeploymentOption deploymentOption : deploymentOptions2) {
- assertTrue(region2.contains(deploymentOption.getExecutionVertexId()));
+ for (ExecutionVertexID vertexId : scheduledVertices2) {
+ assertTrue(region2.contains(vertexId));
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/strategy/StrategyTestUtil.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/strategy/StrategyTestUtil.java
index 15e031894ae..38ab94f30e3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/strategy/StrategyTestUtil.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/strategy/StrategyTestUtil.java
@@ -18,8 +18,6 @@
package org.apache.flink.runtime.scheduler.strategy;
-import org.apache.flink.runtime.scheduler.ExecutionVertexDeploymentOption;
-
import java.util.List;
import java.util.stream.Collectors;
@@ -30,26 +28,17 @@ import static org.junit.Assert.assertEquals;
/** Strategy test utilities. */
public class StrategyTestUtil {
- static List<ExecutionVertexID> getExecutionVertexIdsFromDeployOptions(
- final List<ExecutionVertexDeploymentOption> deploymentOptions) {
-
- return deploymentOptions.stream()
- .map(ExecutionVertexDeploymentOption::getExecutionVertexId)
- .collect(Collectors.toList());
- }
-
static void assertLatestScheduledVerticesAreEqualTo(
final List<List<TestingSchedulingExecutionVertex>> expected,
TestingSchedulerOperations testingSchedulerOperation) {
- final List<List<ExecutionVertexDeploymentOption>> deploymentOptions =
+ final List<List<ExecutionVertexID>> allScheduledVertices =
testingSchedulerOperation.getScheduledVertices();
final int expectedScheduledBulks = expected.size();
- assertThat(expectedScheduledBulks, lessThanOrEqualTo(deploymentOptions.size()));
+ assertThat(expectedScheduledBulks, lessThanOrEqualTo(allScheduledVertices.size()));
for (int i = 0; i < expectedScheduledBulks; i++) {
assertEquals(
idsFromVertices(expected.get(expectedScheduledBulks - i - 1)),
- idsFromDeploymentOptions(
- deploymentOptions.get(deploymentOptions.size() - i - 1)));
+ allScheduledVertices.get(allScheduledVertices.size() - i - 1));
}
}
@@ -59,12 +48,4 @@ public class StrategyTestUtil {
.map(TestingSchedulingExecutionVertex::getId)
.collect(Collectors.toList());
}
-
- static List<ExecutionVertexID> idsFromDeploymentOptions(
- final List<ExecutionVertexDeploymentOption> deploymentOptions) {
-
- return deploymentOptions.stream()
- .map(ExecutionVertexDeploymentOption::getExecutionVertexId)
- .collect(Collectors.toList());
- }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/strategy/TestSchedulingStrategy.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/strategy/TestSchedulingStrategy.java
index 62a68da073f..5326f7211d6 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/strategy/TestSchedulingStrategy.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/strategy/TestSchedulingStrategy.java
@@ -20,11 +20,8 @@ package org.apache.flink.runtime.scheduler.strategy;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
-import org.apache.flink.runtime.scheduler.DeploymentOption;
-import org.apache.flink.runtime.scheduler.ExecutionVertexDeploymentOption;
import org.apache.flink.runtime.scheduler.SchedulerOperations;
-import java.util.ArrayList;
import java.util.List;
import java.util.Set;
@@ -37,8 +34,6 @@ public class TestSchedulingStrategy implements SchedulingStrategy {
private final SchedulingTopology schedulingTopology;
- private final DeploymentOption deploymentOption = new DeploymentOption(false);
-
private Set<ExecutionVertexID> receivedVerticesToRestart;
public TestSchedulingStrategy(
@@ -77,21 +72,7 @@ public class TestSchedulingStrategy implements SchedulingStrategy {
}
private void allocateSlotsAndDeploy(final List<ExecutionVertexID> verticesToSchedule) {
- final List<ExecutionVertexDeploymentOption> executionVertexDeploymentOptions =
- createExecutionVertexDeploymentOptions(verticesToSchedule);
- schedulerOperations.allocateSlotsAndDeploy(executionVertexDeploymentOptions);
- }
-
- private List<ExecutionVertexDeploymentOption> createExecutionVertexDeploymentOptions(
- final List<ExecutionVertexID> vertices) {
-
- final List<ExecutionVertexDeploymentOption> executionVertexDeploymentOptions =
- new ArrayList<>(vertices.size());
- for (ExecutionVertexID executionVertexID : vertices) {
- executionVertexDeploymentOptions.add(
- new ExecutionVertexDeploymentOption(executionVertexID, deploymentOption));
- }
- return executionVertexDeploymentOptions;
+ schedulerOperations.allocateSlotsAndDeploy(verticesToSchedule);
}
/** The factory for creating {@link TestSchedulingStrategy}. */
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/strategy/TestingSchedulerOperations.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/strategy/TestingSchedulerOperations.java
index 55edbeb2301..8146310f76e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/strategy/TestingSchedulerOperations.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/strategy/TestingSchedulerOperations.java
@@ -18,7 +18,6 @@
package org.apache.flink.runtime.scheduler.strategy;
-import org.apache.flink.runtime.scheduler.ExecutionVertexDeploymentOption;
import org.apache.flink.runtime.scheduler.SchedulerOperations;
import java.util.ArrayList;
@@ -28,15 +27,14 @@ import java.util.List;
/** A Simple scheduler operation for testing purposes. */
public class TestingSchedulerOperations implements SchedulerOperations {
- private final List<List<ExecutionVertexDeploymentOption>> scheduledVertices = new ArrayList<>();
+ private final List<List<ExecutionVertexID>> scheduledVertices = new ArrayList<>();
@Override
- public void allocateSlotsAndDeploy(
- List<ExecutionVertexDeploymentOption> executionVertexDeploymentOptions) {
- scheduledVertices.add(executionVertexDeploymentOptions);
+ public void allocateSlotsAndDeploy(List<ExecutionVertexID> verticesToDeploy) {
+ scheduledVertices.add(verticesToDeploy);
}
- List<List<ExecutionVertexDeploymentOption>> getScheduledVertices() {
+ List<List<ExecutionVertexID>> getScheduledVertices() {
return Collections.unmodifiableList(scheduledVertices);
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/NettyShuffleUtilsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/NettyShuffleUtilsTest.java
index 398366d0d05..71979f72750 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/NettyShuffleUtilsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/shuffle/NettyShuffleUtilsTest.java
@@ -175,8 +175,7 @@ public class NettyShuffleUtilsTest extends TestLogger {
numSubpartitions,
0);
ResultPartitionDeploymentDescriptor resultPartitionDeploymentDescriptor =
- new ResultPartitionDeploymentDescriptor(
- partitionDescriptor, shuffleDescriptor, 1, true);
+ new ResultPartitionDeploymentDescriptor(partitionDescriptor, shuffleDescriptor, 1);
ExecutionAttemptID consumerID = new ExecutionAttemptID();
Collection<ResultPartition> resultPartitions =
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/DefaultJobTableTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/DefaultJobTableTest.java
index 4babc209015..5a9a856647a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/DefaultJobTableTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/DefaultJobTableTest.java
@@ -21,7 +21,6 @@ package org.apache.flink.runtime.taskexecutor;
import org.apache.flink.api.common.JobID;
import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
-import org.apache.flink.runtime.io.network.partition.NoOpResultPartitionConsumableNotifier;
import org.apache.flink.runtime.jobmaster.utils.TestingJobMasterGatewayBuilder;
import org.apache.flink.runtime.taskmanager.NoOpCheckpointResponder;
import org.apache.flink.runtime.taskmanager.NoOpTaskManagerActions;
@@ -121,7 +120,6 @@ public class DefaultJobTableTest extends TestLogger {
new NoOpTaskManagerActions(),
NoOpCheckpointResponder.INSTANCE,
new TestGlobalAggregateManager(),
- new NoOpResultPartitionConsumableNotifier(),
new NoOpPartitionProducerStateChecker());
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorPartitionLifecycleTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorPartitionLifecycleTest.java
index 89a223ac2b0..dca9ff90d89 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorPartitionLifecycleTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorPartitionLifecycleTest.java
@@ -215,10 +215,8 @@ public class TaskExecutorPartitionLifecycleTest extends TestLogger {
TaskSubmissionTestEnvironment.registerJobMasterConnection(
jobTable,
jobId,
- rpc,
jobMasterGateway,
new NoOpTaskManagerActions(),
- timeout,
taskExecutor.getMainThreadExecutableForTesting());
final TaskExecutorGateway taskExecutorGateway =
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorSubmissionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorSubmissionTest.java
index 36443dca212..6ace8804ff0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorSubmissionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorSubmissionTest.java
@@ -22,9 +22,7 @@ import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
-import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.blob.PermanentBlobKey;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
@@ -60,7 +58,6 @@ import org.apache.flink.runtime.testtasks.BlockingNoOpInvokable;
import org.apache.flink.runtime.util.NettyShuffleDescriptorBuilder;
import org.apache.flink.testutils.TestingUtils;
import org.apache.flink.testutils.executor.TestExecutorResource;
-import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.NetUtils;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.SerializedValue;
@@ -86,7 +83,6 @@ import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
@@ -281,9 +277,6 @@ public class TaskExecutorSubmissionTest extends TestLogger {
TestingJobMasterGateway testingJobMasterGateway =
new TestingJobMasterGatewayBuilder()
.setFencingTokenSupplier(() -> jobMasterId)
- .setNotifyPartitionDataAvailableFunction(
- resultPartitionID ->
- CompletableFuture.completedFuture(Acknowledge.get()))
.build();
try (TaskSubmissionTestEnvironment env =
@@ -543,76 +536,6 @@ public class TaskExecutorSubmissionTest extends TestLogger {
}
}
- /**
- * Test that a failing notifyPartitionDataAvailable call leads to the failing of the respective
- * task.
- *
- * <p>IMPORTANT: We have to make sure that the invokable's cancel method is called, because only
- * then the future is completed. We do this by not eagerly deploying consumer tasks and
- * requiring the invokable to fill one memory segment. The completed memory segment will trigger
- * the scheduling of the downstream operator since it is in pipeline mode. After we've filled
- * the memory segment, we'll block the invokable and wait for the task failure due to the failed
- * notifyPartitionDataAvailable call.
- */
- @Test
- public void testFailingNotifyPartitionDataAvailable() throws Exception {
- final Configuration configuration = new Configuration();
-
- // set the memory segment to the smallest size possible, because we have to fill one
- // memory buffer to trigger notifyPartitionDataAvailable to the downstream
- // operators
- configuration.set(TaskManagerOptions.MEMORY_SEGMENT_SIZE, MemorySize.parse("4096"));
-
- NettyShuffleDescriptor sdd =
- createRemoteWithIdAndLocation(
- new IntermediateResultPartitionID(), ResourceID.generate());
- TaskDeploymentDescriptor tdd =
- createSender(sdd, TestingAbstractInvokables.TestInvokableRecordCancel.class);
- ExecutionAttemptID eid = tdd.getExecutionAttemptId();
-
- final CompletableFuture<Void> taskRunningFuture = new CompletableFuture<>();
-
- final Exception exception = new Exception("Failed notifyPartitionDataAvailable");
-
- final JobMasterId jobMasterId = JobMasterId.generate();
- TestingJobMasterGateway testingJobMasterGateway =
- new TestingJobMasterGatewayBuilder()
- .setFencingTokenSupplier(() -> jobMasterId)
- .setNotifyPartitionDataAvailableFunction(
- resultPartitionID -> FutureUtils.completedExceptionally(exception))
- .build();
-
- try (TaskSubmissionTestEnvironment env =
- new TaskSubmissionTestEnvironment.Builder(jobId)
- .setSlotSize(1)
- .setConfiguration(configuration)
- .addTaskManagerActionListener(
- eid, ExecutionState.RUNNING, taskRunningFuture)
- .setJobMasterId(jobMasterId)
- .setJobMasterGateway(testingJobMasterGateway)
- .useRealNonMockShuffleEnvironment()
- .build(EXECUTOR_RESOURCE.getExecutor())) {
- TaskExecutorGateway tmGateway = env.getTaskExecutorGateway();
- TaskSlotTable<Task> taskSlotTable = env.getTaskSlotTable();
-
- TestingAbstractInvokables.TestInvokableRecordCancel.resetGotCanceledFuture();
-
- taskSlotTable.allocateSlot(0, jobId, tdd.getAllocationId(), Time.seconds(60));
- tmGateway.submitTask(tdd, jobMasterId, timeout).get();
- taskRunningFuture.get();
-
- CompletableFuture<Boolean> cancelFuture =
- TestingAbstractInvokables.TestInvokableRecordCancel.gotCanceled();
-
- assertTrue(cancelFuture.get());
- assertTrue(
- ExceptionUtils.findThrowableWithMessage(
- taskSlotTable.getTask(eid).getFailureCause(),
- exception.getMessage())
- .isPresent());
- }
- }
-
private TaskDeploymentDescriptor createSender(NettyShuffleDescriptor shuffleDescriptor)
throws IOException {
return createSender(shuffleDescriptor, TestingAbstractInvokables.Sender.class);
@@ -627,8 +550,7 @@ public class TaskExecutorSubmissionTest extends TestLogger {
.setPartitionId(shuffleDescriptor.getResultPartitionID().getPartitionId())
.build();
ResultPartitionDeploymentDescriptor resultPartitionDeploymentDescriptor =
- new ResultPartitionDeploymentDescriptor(
- partitionDescriptor, shuffleDescriptor, 1, true);
+ new ResultPartitionDeploymentDescriptor(partitionDescriptor, shuffleDescriptor, 1);
return createTestTaskDeploymentDescriptor(
"Sender",
shuffleDescriptor.getResultPartitionID().getProducerId(),
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskSubmissionTestEnvironment.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskSubmissionTestEnvironment.java
index 5ab9a22edda..ca0f7e43ba2 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskSubmissionTestEnvironment.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskSubmissionTestEnvironment.java
@@ -46,7 +46,6 @@ import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.runtime.rpc.TestingRpcService;
import org.apache.flink.runtime.shuffle.ShuffleEnvironment;
import org.apache.flink.runtime.state.TaskExecutorLocalStateStoresManager;
-import org.apache.flink.runtime.taskexecutor.rpc.RpcResultPartitionConsumableNotifier;
import org.apache.flink.runtime.taskexecutor.slot.DefaultTimerService;
import org.apache.flink.runtime.taskexecutor.slot.TaskSlotTable;
import org.apache.flink.runtime.taskexecutor.slot.TaskSlotUtils;
@@ -194,20 +193,16 @@ class TaskSubmissionTestEnvironment implements AutoCloseable {
registerJobMasterConnection(
jobTable,
jobId,
- testingRpcService,
jobMasterGateway,
taskManagerActions,
- timeout,
taskExecutor.getMainThreadExecutableForTesting());
}
static void registerJobMasterConnection(
JobTable jobTable,
JobID jobId,
- RpcService testingRpcService,
JobMasterGateway jobMasterGateway,
TaskManagerActions taskManagerActions,
- Time timeout,
MainThreadExecutable mainThreadExecutable) {
mainThreadExecutable.runAsync(
() -> {
@@ -220,10 +215,6 @@ class TaskSubmissionTestEnvironment implements AutoCloseable {
taskManagerActions,
new TestCheckpointResponder(),
new TestGlobalAggregateManager(),
- new RpcResultPartitionConsumableNotifier(
- jobMasterGateway,
- testingRpcService.getScheduledExecutor(),
- timeout),
TestingPartitionProducerStateChecker.newBuilder()
.setPartitionProducerStateFunction(
(jobID, intermediateDataSetID, resultPartitionID) ->
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java
index 784bcfce6a2..c3782f7c260 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java
@@ -41,8 +41,6 @@ import org.apache.flink.runtime.filecache.FileCache;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
-import org.apache.flink.runtime.io.network.partition.NoOpResultPartitionConsumableNotifier;
-import org.apache.flink.runtime.io.network.partition.ResultPartitionConsumableNotifier;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider;
@@ -169,8 +167,6 @@ public class TaskAsyncCallTest extends TestLogger {
.build())
.build();
- ResultPartitionConsumableNotifier consumableNotifier =
- new NoOpResultPartitionConsumableNotifier();
PartitionProducerStateChecker partitionProducerStateChecker =
mock(PartitionProducerStateChecker.class);
Executor executor = mock(Executor.class);
@@ -221,7 +217,6 @@ public class TaskAsyncCallTest extends TestLogger {
mock(FileCache.class),
new TestingTaskManagerRuntimeInfo(),
taskMetricGroup,
- consumableNotifier,
partitionProducerStateChecker,
executor);
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
index 543c875d787..bb309448d11 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
@@ -33,8 +33,6 @@ import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.execution.librarycache.TestingClassLoaderLease;
import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
-import org.apache.flink.runtime.io.network.partition.NoOpResultPartitionConsumableNotifier;
-import org.apache.flink.runtime.io.network.partition.ResultPartitionConsumableNotifier;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
import org.apache.flink.runtime.io.network.partition.consumer.RemoteChannelStateChecker;
@@ -295,8 +293,7 @@ public class TaskTest extends TestLogger {
final ShuffleDescriptor shuffleDescriptor =
NettyShuffleDescriptorBuilder.newBuilder().buildLocal();
final ResultPartitionDeploymentDescriptor dummyPartition =
- new ResultPartitionDeploymentDescriptor(
- partitionDescriptor, shuffleDescriptor, 1, false);
+ new ResultPartitionDeploymentDescriptor(partitionDescriptor, shuffleDescriptor, 1);
testExecutionFailsInNetworkRegistration(
Collections.singletonList(dummyPartition), Collections.emptyList());
}
@@ -321,8 +318,6 @@ public class TaskTest extends TestLogger {
throws Exception {
final String errorMessage = "Network buffer pool has already been destroyed.";
- final ResultPartitionConsumableNotifier consumableNotifier =
- new NoOpResultPartitionConsumableNotifier();
final PartitionProducerStateChecker partitionProducerStateChecker =
mock(PartitionProducerStateChecker.class);
@@ -330,7 +325,6 @@ public class TaskTest extends TestLogger {
final Task task =
new TestTaskBuilder(shuffleEnvironment)
.setTaskManagerActions(taskManagerActions)
- .setConsumableNotifier(consumableNotifier)
.setPartitionProducerStateChecker(partitionProducerStateChecker)
.setResultPartitions(resultPartitions)
.setInputGates(inputGates)
@@ -762,9 +756,6 @@ public class TaskTest extends TestLogger {
final PartitionProducerStateChecker partitionChecker =
mock(PartitionProducerStateChecker.class);
- final ResultPartitionConsumableNotifier consumableNotifier =
- new NoOpResultPartitionConsumableNotifier();
-
AtomicInteger callCount = new AtomicInteger(0);
RemoteChannelStateChecker remoteChannelStateChecker =
@@ -779,7 +770,6 @@ public class TaskTest extends TestLogger {
final Task task =
createTaskBuilder()
.setInvokable(InvokableBlockingInInvoke.class)
- .setConsumableNotifier(consumableNotifier)
.setPartitionProducerStateChecker(partitionChecker)
.build(Executors.directExecutor());
TestTaskBuilder.setTaskState(task, ExecutionState.RUNNING);
@@ -810,7 +800,6 @@ public class TaskTest extends TestLogger {
final Task task =
createTaskBuilder()
.setInvokable(InvokableBlockingInInvoke.class)
- .setConsumableNotifier(consumableNotifier)
.setPartitionProducerStateChecker(partitionChecker)
.build(Executors.directExecutor());
TestTaskBuilder.setTaskState(task, ExecutionState.RUNNING);
@@ -845,7 +834,6 @@ public class TaskTest extends TestLogger {
final Task task =
createTaskBuilder()
.setInvokable(InvokableBlockingInInvoke.class)
- .setConsumableNotifier(consumableNotifier)
.setPartitionProducerStateChecker(partitionChecker)
.build(Executors.directExecutor());
@@ -889,7 +877,6 @@ public class TaskTest extends TestLogger {
final Task task =
createTaskBuilder()
.setInvokable(InvokableBlockingInInvoke.class)
- .setConsumableNotifier(consumableNotifier)
.setPartitionProducerStateChecker(partitionChecker)
.build(Executors.directExecutor());
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TestTaskBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TestTaskBuilder.java
index 698c4159b53..c855c086bba 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TestTaskBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TestTaskBuilder.java
@@ -36,8 +36,6 @@ import org.apache.flink.runtime.externalresource.ExternalResourceInfoProvider;
import org.apache.flink.runtime.filecache.FileCache;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
-import org.apache.flink.runtime.io.network.partition.NoOpResultPartitionConsumableNotifier;
-import org.apache.flink.runtime.io.network.partition.ResultPartitionConsumableNotifier;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.jobgraph.tasks.TaskInvokable;
@@ -71,8 +69,6 @@ public final class TestTaskBuilder {
private TaskManagerActions taskManagerActions = new NoOpTaskManagerActions();
private LibraryCacheManager.ClassLoaderHandle classLoaderHandle =
TestingClassLoaderLease.newBuilder().build();
- private ResultPartitionConsumableNotifier consumableNotifier =
- new NoOpResultPartitionConsumableNotifier();
private PartitionProducerStateChecker partitionProducerStateChecker =
new NoOpPartitionProducerStateChecker();
private final ShuffleEnvironment<?, ?> shuffleEnvironment;
@@ -110,12 +106,6 @@ public final class TestTaskBuilder {
return this;
}
- public TestTaskBuilder setConsumableNotifier(
- ResultPartitionConsumableNotifier consumableNotifier) {
- this.consumableNotifier = consumableNotifier;
- return this;
- }
-
public TestTaskBuilder setPartitionProducerStateChecker(
PartitionProducerStateChecker partitionProducerStateChecker) {
this.partitionProducerStateChecker = partitionProducerStateChecker;
@@ -233,7 +223,6 @@ public final class TestTaskBuilder {
mock(FileCache.class),
new TestingTaskManagerRuntimeInfo(taskManagerConfig),
taskMetricGroup,
- consumableNotifier,
partitionProducerStateChecker,
executor);
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/util/JvmExitOnFatalErrorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/util/JvmExitOnFatalErrorTest.java
index 634f4a65fd1..9b0b77ebffb 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/util/JvmExitOnFatalErrorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/util/JvmExitOnFatalErrorTest.java
@@ -40,7 +40,6 @@ import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
-import org.apache.flink.runtime.io.network.partition.NoOpResultPartitionConsumableNotifier;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider;
@@ -251,7 +250,6 @@ public class JvmExitOnFatalErrorTest extends TestLogger {
VoidPermanentBlobService.INSTANCE),
tmInfo,
UnregisteredMetricGroups.createUnregisteredTaskMetricGroup(),
- new NoOpResultPartitionConsumableNotifier(),
new NoOpPartitionProducerStateChecker(),
executor);
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/InterruptSensitiveRestoreTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/InterruptSensitiveRestoreTest.java
index d65a1755fdb..5fc0709045d 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/InterruptSensitiveRestoreTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/InterruptSensitiveRestoreTest.java
@@ -43,7 +43,6 @@ import org.apache.flink.runtime.filecache.FileCache;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
-import org.apache.flink.runtime.io.network.partition.NoOpResultPartitionConsumableNotifier;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider;
@@ -292,7 +291,6 @@ public class InterruptSensitiveRestoreTest {
VoidPermanentBlobService.INSTANCE),
new TestingTaskManagerRuntimeInfo(),
UnregisteredMetricGroups.createUnregisteredTaskMetricGroup(),
- new NoOpResultPartitionConsumableNotifier(),
mock(PartitionProducerStateChecker.class),
mock(Executor.class));
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java
index 3423fa341a8..f3c660d5855 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java
@@ -138,8 +138,7 @@ public class StreamTaskITCase extends TestLogger {
new ResultPartitionDeploymentDescriptor(
PartitionDescriptorBuilder.newBuilder().build(),
NettyShuffleDescriptorBuilder.newBuilder().buildLocal(),
- 1,
- false);
+ 1);
return new TestTaskBuilder(shuffleEnvironment)
.setInvokable(NoOpStreamTask.class)
.setTaskConfig(taskConfiguration)
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskSystemExitTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskSystemExitTest.java
index 00cbaf81aa4..8fcfe04adfa 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskSystemExitTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskSystemExitTest.java
@@ -38,7 +38,6 @@ import org.apache.flink.runtime.filecache.FileCache;
import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
-import org.apache.flink.runtime.io.network.partition.NoOpResultPartitionConsumableNotifier;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider;
@@ -204,7 +203,6 @@ public class StreamTaskSystemExitTest extends TestLogger {
mock(FileCache.class),
taskManagerRuntimeInfo,
UnregisteredMetricGroups.createUnregisteredTaskMetricGroup(),
- new NoOpResultPartitionConsumableNotifier(),
mock(PartitionProducerStateChecker.class),
Executors.directExecutor());
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java
index 60f4bfbf704..f52d0b83614 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java
@@ -41,7 +41,6 @@ import org.apache.flink.runtime.filecache.FileCache;
import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
-import org.apache.flink.runtime.io.network.partition.NoOpResultPartitionConsumableNotifier;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider;
@@ -188,7 +187,6 @@ public class StreamTaskTerminationTest extends TestLogger {
mock(FileCache.class),
taskManagerRuntimeInfo,
UnregisteredMetricGroups.createUnregisteredTaskMetricGroup(),
- new NoOpResultPartitionConsumableNotifier(),
mock(PartitionProducerStateChecker.class),
Executors.directExecutor());
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SynchronousCheckpointITCase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SynchronousCheckpointITCase.java
index b96e95fcf12..5a106aaa230 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SynchronousCheckpointITCase.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SynchronousCheckpointITCase.java
@@ -42,8 +42,6 @@ import org.apache.flink.runtime.filecache.FileCache;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
-import org.apache.flink.runtime.io.network.partition.NoOpResultPartitionConsumableNotifier;
-import org.apache.flink.runtime.io.network.partition.ResultPartitionConsumableNotifier;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider;
import org.apache.flink.runtime.jobgraph.tasks.TaskInvokable;
@@ -224,8 +222,6 @@ public class SynchronousCheckpointITCase {
private Task createTask(Class<? extends TaskInvokable> invokableClass) throws Exception {
- ResultPartitionConsumableNotifier consumableNotifier =
- new NoOpResultPartitionConsumableNotifier();
PartitionProducerStateChecker partitionProducerStateChecker =
mock(PartitionProducerStateChecker.class);
Executor executor = mock(Executor.class);
@@ -278,7 +274,6 @@ public class SynchronousCheckpointITCase {
mock(FileCache.class),
new TestingTaskManagerRuntimeInfo(),
taskMetricGroup,
- consumableNotifier,
partitionProducerStateChecker,
executor);
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TaskCheckpointingBehaviourTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TaskCheckpointingBehaviourTest.java
index 517d6324c99..89785e97042 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TaskCheckpointingBehaviourTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TaskCheckpointingBehaviourTest.java
@@ -47,7 +47,6 @@ import org.apache.flink.runtime.filecache.FileCache;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
-import org.apache.flink.runtime.io.network.partition.NoOpResultPartitionConsumableNotifier;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider;
@@ -242,7 +241,6 @@ public class TaskCheckpointingBehaviourTest extends TestLogger {
VoidPermanentBlobService.INSTANCE),
new TestingTaskManagerRuntimeInfo(),
UnregisteredMetricGroups.createUnregisteredTaskMetricGroup(),
- new NoOpResultPartitionConsumableNotifier(),
mock(PartitionProducerStateChecker.class),
Executors.directExecutor());
}