[FLINK-18063][checkpointing] Fix the invalid implementation of Altern…

…atingCheckpointBarrierHandler#getAlignmentDurationNanos

We should take the value from active handler instead of aligned handler, because aligned handler is only used for savepoint and in
most cases the unaligned alignment duration should always be 0.

This cloese apache#12460.

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/AlternatingCheckpointBarrierHandler.java

@@ -96,7 +96,7 @@ public long getLatestCheckpointId() {
 
  @Override
  public long getAlignmentDurationNanos() {
- return alignedHandler.getAlignmentDurationNanos();
+ return activeHandler.getAlignmentDurationNanos();
  }
 
  @Override

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierHandler.java

@@ -76,7 +76,9 @@ public void close() throws IOException {
 
  public abstract long getLatestCheckpointId();
 
- public abstract long getAlignmentDurationNanos();
+ public long getAlignmentDurationNanos() {
+ return 0;
+ }
 
  public long getCheckpointStartDelayNanos() {
  return latestCheckpointStartDelayNanos;

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierTracker.java

@@ -217,10 +217,6 @@ public long getLatestCheckpointId() {
  return pendingCheckpoints.isEmpty() ? -1 : pendingCheckpoints.peekLast().checkpointId();
  }
 
- public long getAlignmentDurationNanos() {
- return 0;
- }
-
  public boolean isCheckpointPending() {
  return !pendingCheckpoints.isEmpty();
  }

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java

@@ -199,11 +199,6 @@ public long getLatestCheckpointId() {
  return currentConsumedCheckpointId;
  }
 
- @Override
- public long getAlignmentDurationNanos() {
- return 0;
- }
-
  @Override
  public String toString() {
  return String.format("%s: last checkpoint: %d", taskName, currentConsumedCheckpointId);