[FLINK-20389][tests] Fix UnalignedCheckpointITCase to work with unass…

…igned splits.

The test currently assumed that when induced failures happen, splits have been assigned to readers, which works fine for the planned snapshot-failure-recovery sequence.
However, when unexpected failures happen, this assumption does not necessarily hold resulting in failures that may impede investigations.
The test would still fail as the number of failures would be different from the expected numbers of failures, but investigation can focus on the unexpected failure then.

flink-tests/src/test/java/org/apache/flink/test/checkpointing/UnalignedCheckpointTestBase.java

@@ -224,8 +224,8 @@ public List<LongSplit> snapshotState(long checkpointId) {
 
 			@Override
 			public void notifyCheckpointComplete(long checkpointId) {
-				LOG.info("notifyCheckpointComplete {} @ {} subtask (? attempt)", split.numCompletedCheckpoints, split.nextNumber % split.increment);
 				if (split != null) {
+					LOG.info("notifyCheckpointComplete {} @ {} subtask (? attempt)", split.numCompletedCheckpoints, split.nextNumber % split.increment);
 					split.numCompletedCheckpoints++;
 				}
 			}
@@ -330,6 +330,11 @@ public void addReader(int subtaskId) {
 				}
 			}
 
+			@Override
+			public void notifyCheckpointComplete(long checkpointId) {
+				unassignedSplits.forEach(s -> s.numCompletedCheckpoints++);
+			}
+
 			@Override
 			public List<LongSplit> snapshotState() throws Exception {
 				LOG.info("snapshotState {}", unassignedSplits);
@@ -673,6 +678,5 @@ public void close() throws Exception {
 			LOG.info("Last state {} @ {} subtask ({} attempt)", state, getRuntimeContext().getIndexOfThisSubtask(), getRuntimeContext().getAttemptNumber());
 			super.close();
 		}
-
 	}
 }