[FLINK-17467][task][checkpointing] Code cleanup

spotify · May 14, 2020 · d1db7be · d1db7be
1 parent 2c34a8c
commit d1db7be
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 8 deletions.
diff --git a/...ng-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierAligner.java b/...ng-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierAligner.java
@@ -134,7 +134,7 @@ public void processBarrier(CheckpointBarrier receivedBarrier, int channelIndex)
 
  // -- general code path for multiple input channels --
 
- if (numBarriersReceived > 0) {
+ if (isCheckpointPending()) {
  // this is only true if some alignment is already progress and was not canceled
 
  if (barrierId == currentCheckpointId) {
@@ -243,7 +243,7 @@ public void processCancellationBarrier(CancelCheckpointMarker cancelBarrier) thr
 
  // -- general code path for multiple input channels --
 
- if (numBarriersReceived > 0) {
+ if (isCheckpointPending()) {
  // this is only true if some alignment is in progress and nothing was canceled
 
  if (barrierId == currentCheckpointId) {
@@ -303,7 +303,7 @@ else if (barrierId > currentCheckpointId) {
  public void processEndOfPartition() throws Exception {
  numClosedChannels++;
 
- if (numBarriersReceived > 0) {
+ if (isCheckpointPending()) {
  // let the task know we skip a checkpoint
  notifyAbort(currentCheckpointId,
  new CheckpointException(CheckpointFailureReason.CHECKPOINT_DECLINED_INPUT_END_OF_STREAM));

diff --git a/...-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java b/...-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierUnaligner.java
@@ -118,7 +118,7 @@ public class CheckpointBarrierUnaligner extends CheckpointBarrierHandler {
 
  @Override
  public void releaseBlocksAndResetBarriers() {
- if (numBarrierConsumed > 0) {
+ if (isCheckpointPending()) {
  // make sure no additional data is persisted
  Arrays.fill(hasInflightBuffers, false);
  // the next barrier that comes must assume it is the first
@@ -150,7 +150,7 @@ public void processBarrier(
  CheckpointBarrier receivedBarrier,
  int channelIndex) throws Exception {
  long barrierId = receivedBarrier.getId();
- if (currentConsumedCheckpointId > barrierId || (currentConsumedCheckpointId == barrierId && numBarrierConsumed == 0)) {
+ if (currentConsumedCheckpointId > barrierId || (currentConsumedCheckpointId == barrierId && !isCheckpointPending())) {
  // ignore old and cancelled barriers
  return;
  }
@@ -173,11 +173,11 @@ public void processBarrier(
  public void processCancellationBarrier(CancelCheckpointMarker cancelBarrier) throws Exception {
  final long barrierId = cancelBarrier.getCheckpointId();
 
- if (currentConsumedCheckpointId >= barrierId && numBarrierConsumed == 0) {
+ if (currentConsumedCheckpointId >= barrierId && !isCheckpointPending()) {
  return;
  }
 
- if (numBarrierConsumed > 0) {
+ if (isCheckpointPending()) {
  LOG.warn("{}: Received cancellation barrier for checkpoint {} before completing current checkpoint {}. " +
  "Skipping current checkpoint.",
  taskName,
@@ -196,7 +196,7 @@ public void processCancellationBarrier(CancelCheckpointMarker cancelBarrier) thr
  public void processEndOfPartition() throws Exception {
  threadSafeUnaligner.onChannelClosed();
 
- if (numBarrierConsumed > 0) {
+ if (isCheckpointPending()) {
  // let the task know we skip a checkpoint
  notifyAbort(
  currentConsumedCheckpointId,