Revert "[FLINK-21178][Runtime/Checkpointing] Task failure should trig…

…ger master hook's reset() (apache#14890)"

This reverts commit 816ce96

flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java

@@ -1536,13 +1536,13 @@ private OptionalLong restoreLatestCheckpointedStateInternal(
  throw new IllegalStateException("No completed checkpoint available");
  }
 
- LOG.debug("Resetting the master hooks.");
- MasterHooks.reset(masterHooks.values(), LOG);
-
  if (operatorCoordinatorRestoreBehavior
  == OperatorCoordinatorRestoreBehavior.RESTORE_OR_RESET) {
  // we let the JobManager-side components know that there was a recovery,
  // even if there was no checkpoint to recover from, yet
+ LOG.debug("Resetting the master hooks.");
+ MasterHooks.reset(masterHooks.values(), LOG);
+
  LOG.info("Resetting the Operator Coordinators to an empty state.");
  restoreStateToCoordinators(
  OperatorCoordinator.NO_CHECKPOINT, Collections.emptyMap());

flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java

@@ -3350,26 +3350,6 @@ public Integer deserialize(int version, byte[] serialized)
  assertTrue(checkpointCoordinator.getSuccessfulCheckpoints().isEmpty());
  }
 
- @Test
- public void testResetCalledInRegionRecovery() throws Exception {
- final JobID jobId = new JobID();
-
- // set up the coordinator
- CheckpointCoordinator checkpointCoordinator =
- new CheckpointCoordinatorBuilder()
- .setJobId(jobId)
- .setTimer(manuallyTriggeredScheduledExecutor)
- .build();
-
- TestResetHook hook = new TestResetHook("id");
-
- // Add a master hook
- checkpointCoordinator.addMasterHook(hook);
- assertFalse(hook.resetCalled);
- checkpointCoordinator.restoreLatestCheckpointedStateToSubtasks(Collections.emptySet());
- assertTrue(hook.resetCalled);
- }
-
  @Test
  public void testNotifyCheckpointAbortionInOperatorCoordinator() throws Exception {
  JobVertexID jobVertexID = new JobVertexID();
@@ -3612,42 +3592,4 @@ public int getInvokeCounter() {
  return invokeCounter;
  }
  }
-
- private static class TestResetHook implements MasterTriggerRestoreHook<String> {
-
- private final String id;
- boolean resetCalled;
-
- TestResetHook(String id) {
- this.id = id;
- this.resetCalled = false;
- }
-
- @Override
- public String getIdentifier() {
- return id;
- }
-
- @Override
- public void reset() throws Exception {
- resetCalled = true;
- }
-
- @Override
- public CompletableFuture<String> triggerCheckpoint(
- long checkpointId, long timestamp, Executor executor) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void restoreCheckpoint(long checkpointId, @Nullable String checkpointData)
- throws Exception {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public SimpleVersionedSerializer<String> createCheckpointDataSerializer() {
- throw new UnsupportedOperationException();
- }
- }
 }