[FLINK-20439][runtime] Rename scheduleOrUpdateConsumers to `notifyP…

…artitionDataAvailable` to avoid confusion

flink-runtime/src/main/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptor.java

@@ -45,19 +45,19 @@ public class ResultPartitionDeploymentDescriptor implements Serializable {
 
  private final int maxParallelism;
 
- /** Flag whether the result partition should send scheduleOrUpdateConsumer messages. */
- private final boolean sendScheduleOrUpdateConsumersMessage;
+ /** 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 sendScheduleOrUpdateConsumersMessage) {
+ boolean notifyPartitionDataAvailable) {
  this.partitionDescriptor = checkNotNull(partitionDescriptor);
  this.shuffleDescriptor = checkNotNull(shuffleDescriptor);
  KeyGroupRangeAssignment.checkParallelismPreconditions(maxParallelism);
  this.maxParallelism = maxParallelism;
- this.sendScheduleOrUpdateConsumersMessage = sendScheduleOrUpdateConsumersMessage;
+ this.notifyPartitionDataAvailable = notifyPartitionDataAvailable;
  }
 
  public IntermediateDataSetID getResultId() {
@@ -88,8 +88,8 @@ public ShuffleDescriptor getShuffleDescriptor() {
  return shuffleDescriptor;
  }
 
- public boolean sendScheduleOrUpdateConsumersMessage() {
- return sendScheduleOrUpdateConsumersMessage;
+ public boolean notifyPartitionDataAvailable() {
+ return notifyPartitionDataAvailable;
  }
 
  @Override

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java

@@ -397,12 +397,12 @@ public CompletableFuture<?> getReleaseFuture() {
 
  public CompletableFuture<Execution> registerProducedPartitions(
  TaskManagerLocation location,
- boolean sendScheduleOrUpdateConsumersMessage) {
+ boolean notifyPartitionDataAvailable) {
 
  assertRunningInJobMasterMainThread();
 
  return FutureUtils.thenApplyAsyncIfNotDone(
- registerProducedPartitions(vertex, location, attemptId, sendScheduleOrUpdateConsumersMessage),
+ registerProducedPartitions(vertex, location, attemptId, notifyPartitionDataAvailable),
  vertex.getExecutionGraph().getJobMasterMainThreadExecutor(),
  producedPartitionsCache -> {
  producedPartitions = producedPartitionsCache;
@@ -436,7 +436,7 @@ static CompletableFuture<Map<IntermediateResultPartitionID, ResultPartitionDeplo
  ExecutionVertex vertex,
  TaskManagerLocation location,
  ExecutionAttemptID attemptId,
- boolean sendScheduleOrUpdateConsumersMessage) {
+ boolean notifyPartitionDataAvailable) {
 
  ProducerDescriptor producerDescriptor = ProducerDescriptor.create(location, attemptId);
 
@@ -460,7 +460,7 @@ static CompletableFuture<Map<IntermediateResultPartitionID, ResultPartitionDeplo
  partitionDescriptor,
  shuffleDescriptor,
  maxParallelism,
- sendScheduleOrUpdateConsumersMessage));
+ notifyPartitionDataAvailable));
  partitionRegistrations.add(partitionRegistration);
  }
 

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java

@@ -453,11 +453,11 @@ public CompletableFuture<ExecutionState> requestPartitionState(
  }
 
  @Override
- public CompletableFuture<Acknowledge> scheduleOrUpdateConsumers(
+ public CompletableFuture<Acknowledge> notifyPartitionDataAvailable(
  final ResultPartitionID partitionID,
  final Time timeout) {
 
- schedulerNG.scheduleOrUpdateConsumers(partitionID);
+ schedulerNG.notifyPartitionDataAvailable(partitionID);
  return CompletableFuture.completedFuture(Acknowledge.get());
  }
 

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java

@@ -118,9 +118,9 @@ CompletableFuture<ExecutionState> requestPartitionState(
  *
  * @param partitionID The partition which has already produced data
  * @param timeout before the rpc call fails
- * @return Future acknowledge of the schedule or update operation
+ * @return Future acknowledge of the notification
  */
- CompletableFuture<Acknowledge> scheduleOrUpdateConsumers(
+ CompletableFuture<Acknowledge> notifyPartitionDataAvailable(
  final ResultPartitionID partitionID,
  @RpcTimeout final Time timeout);
 

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultScheduler.java

@@ -313,7 +313,7 @@ private CompletableFuture<?> cancelExecutionVertex(final ExecutionVertexID execu
  }
 
  @Override
- protected void scheduleOrUpdateConsumersInternal(final IntermediateResultPartitionID partitionId) {
+ protected void notifyPartitionDataAvailableInternal(final IntermediateResultPartitionID partitionId) {
  schedulingStrategy.onPartitionConsumable(partitionId);
  }
 
@@ -442,10 +442,12 @@ private BiFunction<LogicalSlot, Throwable, Void> assignResourceOrHandleError(fin
 
  if (throwable == null) {
  final ExecutionVertex executionVertex = getExecutionVertex(executionVertexId);
- final boolean sendScheduleOrUpdateConsumerMessage = deploymentHandle.getDeploymentOption().sendScheduleOrUpdateConsumerMessage();
+ final boolean notifyPartitionDataAvailable = deploymentHandle
+ .getDeploymentOption()
+ .notifyPartitionDataAvailable();
  executionVertex
  .getCurrentExecutionAttempt()
- .registerProducedPartitions(logicalSlot.getTaskManagerLocation(), sendScheduleOrUpdateConsumerMessage);
+ .registerProducedPartitions(logicalSlot.getTaskManagerLocation(), notifyPartitionDataAvailable);
  executionVertex.tryAssignResource(logicalSlot);
  } else {
  handleTaskDeploymentFailure(executionVertexId, maybeWrapWithNoResourceAvailableException(throwable));

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DeploymentOption.java

@@ -19,17 +19,17 @@
 package org.apache.flink.runtime.scheduler;
 
 /**
- * Deployment option which indicates whether the task should send scheduleOrUpdateConsumer message to master.
+ * Deployment option which indicates whether the task should notify master when its data is available.
  */
 public class DeploymentOption {
 
- private final boolean sendScheduleOrUpdateConsumerMessage;
+ private final boolean notifyPartitionDataAvailable;
 
- public DeploymentOption(boolean sendScheduleOrUpdateConsumerMessage) {
- this.sendScheduleOrUpdateConsumerMessage = sendScheduleOrUpdateConsumerMessage;
+ public DeploymentOption(boolean notifyPartitionDataAvailable) {
+ this.notifyPartitionDataAvailable = notifyPartitionDataAvailable;
  }
 
- public boolean sendScheduleOrUpdateConsumerMessage() {
- return sendScheduleOrUpdateConsumerMessage;
+ public boolean notifyPartitionDataAvailable() {
+ return notifyPartitionDataAvailable;
  }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java

@@ -685,15 +685,15 @@ public ExecutionState requestPartitionState(
  }
 
  @Override
- public final void scheduleOrUpdateConsumers(final ResultPartitionID partitionId) {
+ public final void notifyPartitionDataAvailable(final ResultPartitionID partitionId) {
  mainThreadExecutor.assertRunningInMainThread();
 
  executionGraph.notifyPartitionDataAvailable(partitionId);
 
- scheduleOrUpdateConsumersInternal(partitionId.getPartitionId());
+ notifyPartitionDataAvailableInternal(partitionId.getPartitionId());
  }
 
- protected void scheduleOrUpdateConsumersInternal(IntermediateResultPartitionID resultPartitionId) {
+ protected void notifyPartitionDataAvailableInternal(IntermediateResultPartitionID resultPartitionId) {
  }
 
  @Override

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerNG.java

@@ -95,7 +95,7 @@ default boolean updateTaskExecutionState(TaskExecutionState taskExecutionState)
 
  ExecutionState requestPartitionState(IntermediateDataSetID intermediateResultId, ResultPartitionID resultPartitionId) throws PartitionProducerDisposedException;
 
- void scheduleOrUpdateConsumers(ResultPartitionID partitionID);
+ void notifyPartitionDataAvailable(ResultPartitionID partitionID);
 
  ArchivedExecutionGraph requestJob();
 

flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/rpc/RpcResultPartitionConsumableNotifier.java

@@ -48,16 +48,17 @@ public RpcResultPartitionConsumableNotifier(
  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.scheduleOrUpdateConsumers(partitionId, timeout);
+ CompletableFuture<Acknowledge> acknowledgeFuture = jobMasterGateway.notifyPartitionDataAvailable(partitionId, timeout);
 
  acknowledgeFuture.whenCompleteAsync(
  (Acknowledge ack, Throwable throwable) -> {
  if (throwable != null) {
- LOG.error("Could not schedule or update consumers at the JobManager.", throwable);
+ LOG.error("Could not notify partition data available to JobManager.", throwable);
 
- taskActions.failExternally(new RuntimeException("Could not notify JobManager to schedule or update consumers.", throwable));
+ taskActions.failExternally(new RuntimeException("Could not notify partition data available to JobManager.", throwable));
  }
  },
  executor);

flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/ConsumableNotifyingResultPartitionWriterDecorator.java

@@ -49,7 +49,7 @@ public class ConsumableNotifyingResultPartitionWriterDecorator {
  /**
  * Optionally decorate the ResultPartitionWriter to call
  * {@link ResultPartitionConsumableNotifier#notifyPartitionConsumable(JobID, ResultPartitionID, TaskActions)}
- * on the first record, iff {@link ResultPartitionDeploymentDescriptor#sendScheduleOrUpdateConsumersMessage()}
+ * on the first record, iff {@link ResultPartitionDeploymentDescriptor#notifyPartitionDataAvailable()}
  * is true.
  */
  public static ResultPartitionWriter[] decorate(
@@ -62,7 +62,7 @@ public static ResultPartitionWriter[] decorate(
  ResultPartitionWriter[] consumableNotifyingPartitionWriters = new ResultPartitionWriter[partitionWriters.length];
  int counter = 0;
  for (ResultPartitionDeploymentDescriptor desc : descs) {
- if (desc.sendScheduleOrUpdateConsumersMessage() && desc.getPartitionType().isPipelined()) {
+ if (desc.notifyPartitionDataAvailable() && desc.getPartitionType().isPipelined()) {
  consumableNotifyingPartitionWriters[counter] = new ConsumableNotifyingResultPartitionWriter(
  taskActions,
  jobId,

flink-runtime/src/test/java/org/apache/flink/runtime/deployment/ResultPartitionDeploymentDescriptorTest.java

@@ -121,6 +121,6 @@ private static void verifyResultPartitionDeploymentDescriptorCopy(ResultPartitio
  assertThat(copy.getPartitionId(), is(partitionId));
  assertThat(copy.getPartitionType(), is(partitionType));
  assertThat(copy.getNumberOfSubpartitions(), is(numberOfSubpartitions));
- assertThat(copy.sendScheduleOrUpdateConsumersMessage(), is(true));
+ assertThat(copy.notifyPartitionDataAvailable(), is(true));
  }
 }

flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphPartitionReleaseTest.java

@@ -142,7 +142,7 @@ public void testStrategyNotifiedOfUnFinishedVertices() throws Exception {
  final Execution operator1Execution = 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.scheduleOrUpdateConsumers(new ResultPartitionID(partitionId, operator1Execution.getAttemptId()));
+ scheduler.notifyPartitionDataAvailable(new ResultPartitionID(partitionId, operator1Execution.getAttemptId()));
  }
  scheduler.updateTaskExecutionState(new TaskExecutionState(executionGraph.getJobID(), operator1Execution.getAttemptId(), ExecutionState.FINISHED));
  assertThat(releasedPartitions, empty());

flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphVariousFailuesTest.java

@@ -37,11 +37,11 @@
 public class ExecutionGraphVariousFailuesTest extends TestLogger {
 
  /**
- * Tests that a failing scheduleOrUpdateConsumers call with a non-existing execution attempt
+ * Tests that a failing notifyPartitionDataAvailable call with a non-existing execution attempt
  * id, will not fail the execution graph.
  */
  @Test
- public void testFailingScheduleOrUpdateConsumers() throws Exception {
+ public void testFailingNotifyPartitionDataAvailable() throws Exception {
  final SchedulerBase scheduler = SchedulerTestingUtils.newSchedulerBuilder(new JobGraph()).build();
  scheduler.initialize(ComponentMainThreadExecutorServiceAdapter.forMainThread());
  scheduler.startScheduling();
@@ -55,11 +55,11 @@ public void testFailingScheduleOrUpdateConsumers() throws Exception {
  ExecutionAttemptID producerId = new ExecutionAttemptID();
  ResultPartitionID resultPartitionId = new ResultPartitionID(intermediateResultPartitionId, producerId);
 
- // The execution attempt id does not exist and thus the scheduleOrUpdateConsumers call
+ // The execution attempt id does not exist and thus the notifyPartitionDataAvailable call
  // should fail
 
  try {
- scheduler.scheduleOrUpdateConsumers(resultPartitionId);
+ scheduler.notifyPartitionDataAvailable(resultPartitionId);
  fail("Error expected.");
  } catch (IllegalStateException e) {
  // we've expected this exception to occur

flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexDeploymentTest.java

@@ -294,7 +294,7 @@ public void testTddProducedPartitionsLazyScheduling() throws Exception {
 
  assertEquals(1, producedPartitions.size());
  ResultPartitionDeploymentDescriptor desc = producedPartitions.iterator().next();
- assertEquals(scheduleMode.allowLazyDeployment(), desc.sendScheduleOrUpdateConsumersMessage());
+ assertEquals(scheduleMode.allowLazyDeployment(), desc.notifyPartitionDataAvailable());
  }
  }
 

flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionTest.java

@@ -99,10 +99,10 @@ public void testResultSubpartitionInfo() {
  }
 
  /**
- * Tests the schedule or update consumers message sending behaviour depending on the relevant flags.
+ * Tests notifyPartitionDataAvailable behaviour depending on the relevant flags.
  */
  @Test
- public void testSendScheduleOrUpdateConsumersMessage() throws Exception {
+ public void testNotifyPartitionDataAvailable() throws Exception {
  FutureConsumerWithException[] notificationCalls = new FutureConsumerWithException[] {
  writer -> ((ResultPartitionWriter) writer).finish(),
  writer -> ((ResultPartitionWriter) writer).emitRecord(ByteBuffer.allocate(bufferSize), 0),
@@ -111,11 +111,11 @@ public void testSendScheduleOrUpdateConsumersMessage() throws Exception {
  };
 
  for (FutureConsumerWithException notificationCall: notificationCalls) {
- testSendScheduleOrUpdateConsumersMessage(notificationCall);
+ testNotifyPartitionDataAvailable(notificationCall);
  }
  }
 
- private void testSendScheduleOrUpdateConsumersMessage(
+ private void testNotifyPartitionDataAvailable(
  FutureConsumerWithException<ResultPartitionWriter, Exception> notificationCall) throws Exception {
  JobID jobId = new JobID();
  TaskActions taskActions = new NoOpTaskActions();

flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java

@@ -95,7 +95,7 @@ public class TestingJobMasterGateway implements JobMasterGateway {
  private final BiFunction<IntermediateDataSetID, ResultPartitionID, CompletableFuture<ExecutionState>> requestPartitionStateFunction;
 
  @Nonnull
- private final Function<ResultPartitionID, CompletableFuture<Acknowledge>> scheduleOrUpdateConsumersFunction;
+ private final Function<ResultPartitionID, CompletableFuture<Acknowledge>> notifyPartitionDataAvailableFunction;
 
  @Nonnull
  private final Function<ResourceID, CompletableFuture<Acknowledge>> disconnectTaskManagerFunction;
@@ -172,7 +172,7 @@ public TestingJobMasterGateway(
  @Nonnull Function<TaskExecutionState, CompletableFuture<Acknowledge>> updateTaskExecutionStateFunction,
  @Nonnull BiFunction<JobVertexID, ExecutionAttemptID, CompletableFuture<SerializedInputSplit>> requestNextInputSplitFunction,
  @Nonnull BiFunction<IntermediateDataSetID, ResultPartitionID, CompletableFuture<ExecutionState>> requestPartitionStateFunction,
- @Nonnull Function<ResultPartitionID, CompletableFuture<Acknowledge>> scheduleOrUpdateConsumersFunction,
+ @Nonnull Function<ResultPartitionID, CompletableFuture<Acknowledge>> notifyPartitionDataAvailableFunction,
  @Nonnull Function<ResourceID, CompletableFuture<Acknowledge>> disconnectTaskManagerFunction,
  @Nonnull Consumer<ResourceManagerId> disconnectResourceManagerConsumer,
  @Nonnull BiFunction<ResourceID, Collection<SlotOffer>, CompletableFuture<Collection<SlotOffer>>> offerSlotsFunction,
@@ -202,7 +202,7 @@ public TestingJobMasterGateway(
  this.updateTaskExecutionStateFunction = updateTaskExecutionStateFunction;
  this.requestNextInputSplitFunction = requestNextInputSplitFunction;
  this.requestPartitionStateFunction = requestPartitionStateFunction;
- this.scheduleOrUpdateConsumersFunction = scheduleOrUpdateConsumersFunction;
+ this.notifyPartitionDataAvailableFunction = notifyPartitionDataAvailableFunction;
  this.disconnectTaskManagerFunction = disconnectTaskManagerFunction;
  this.disconnectResourceManagerConsumer = disconnectResourceManagerConsumer;
  this.offerSlotsFunction = offerSlotsFunction;
@@ -249,8 +249,8 @@ public CompletableFuture<ExecutionState> requestPartitionState(IntermediateDataS
  }
 
  @Override
- public CompletableFuture<Acknowledge> scheduleOrUpdateConsumers(ResultPartitionID partitionID, Time timeout) {
- return scheduleOrUpdateConsumersFunction.apply(partitionID);
+ public CompletableFuture<Acknowledge> notifyPartitionDataAvailable(ResultPartitionID partitionID, Time timeout) {
+ return notifyPartitionDataAvailableFunction.apply(partitionID);
  }
 
  @Override

flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGatewayBuilder.java

@@ -83,7 +83,7 @@ public class TestingJobMasterGatewayBuilder {
  private Function<TaskExecutionState, CompletableFuture<Acknowledge>> updateTaskExecutionStateFunction = ignored -> CompletableFuture.completedFuture(Acknowledge.get());
  private BiFunction<JobVertexID, ExecutionAttemptID, CompletableFuture<SerializedInputSplit>> requestNextInputSplitFunction = (ignoredA, ignoredB) -> CompletableFuture.completedFuture(new SerializedInputSplit(null));
  private BiFunction<IntermediateDataSetID, ResultPartitionID, CompletableFuture<ExecutionState>> requestPartitionStateFunction = (ignoredA, ignoredB) -> CompletableFuture.completedFuture(ExecutionState.RUNNING);
- private Function<ResultPartitionID, CompletableFuture<Acknowledge>> scheduleOrUpdateConsumersFunction = ignored -> CompletableFuture.completedFuture(Acknowledge.get());
+ 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 -> {};
  private BiFunction<ResourceID, Collection<SlotOffer>, CompletableFuture<Collection<SlotOffer>>> offerSlotsFunction = (ignoredA, ignoredB) -> CompletableFuture.completedFuture(Collections.emptyList());
@@ -138,8 +138,8 @@ public TestingJobMasterGatewayBuilder setRequestPartitionStateFunction(BiFunctio
  return this;
  }
 
- public TestingJobMasterGatewayBuilder setScheduleOrUpdateConsumersFunction(Function<ResultPartitionID, CompletableFuture<Acknowledge>> scheduleOrUpdateConsumersFunction) {
- this.scheduleOrUpdateConsumersFunction = scheduleOrUpdateConsumersFunction;
+ public TestingJobMasterGatewayBuilder setNotifyPartitionDataAvailableFunction(Function<ResultPartitionID, CompletableFuture<Acknowledge>> notifyPartitionDataAvailableFunction) {
+ this.notifyPartitionDataAvailableFunction = notifyPartitionDataAvailableFunction;
  return this;
  }
 
@@ -266,7 +266,7 @@ public TestingJobMasterGateway build() {
  updateTaskExecutionStateFunction,
  requestNextInputSplitFunction,
  requestPartitionStateFunction,
- scheduleOrUpdateConsumersFunction,
+ notifyPartitionDataAvailableFunction,
  disconnectTaskManagerFunction,
  disconnectResourceManagerConsumer,
  offerSlotsFunction,

flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/TestingSchedulerNG.java

@@ -131,7 +131,7 @@ public ExecutionState requestPartitionState(
  }
 
  @Override
- public void scheduleOrUpdateConsumers(ResultPartitionID partitionID) {
+ public void notifyPartitionDataAvailable(ResultPartitionID partitionID) {
  failOperation();
  }