You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ga...@apache.org on 2021/09/14 07:04:29 UTC
[flink] branch release-1.14 updated (574e80f -> 753947a)
This is an automated email from the ASF dual-hosted git repository.
gaoyunhaii pushed a change to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 574e80f [FLINK-24276][python] Avoid confusing output when executing in loopback mode
new 83ab9eb [hotfix][connector] Use null to replace empty pending transaction in 2pc sink
new 753947a [FLINK-24266][checkpoint] Log improvement for aborting checkpoint due to tasks are finishing
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../runtime/checkpoint/CheckpointCoordinator.java | 6 ++
.../checkpoint/CheckpointFailureReason.java | 3 +-
.../functions/sink/TwoPhaseCommitSinkFunction.java | 94 +++++++++-------------
3 files changed, 44 insertions(+), 59 deletions(-)
[flink] 02/02: [FLINK-24266][checkpoint] Log improvement for
aborting checkpoint due to tasks are finishing
Posted by ga...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gaoyunhaii pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 753947ac639664470d49812a14afb5c47075a773
Author: Yun Gao <ga...@gmail.com>
AuthorDate: Mon Sep 13 14:39:18 2021 +0800
[FLINK-24266][checkpoint] Log improvement for aborting checkpoint due to tasks are finishing
This closes #17250.
---
.../org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java | 6 ++++++
.../apache/flink/runtime/checkpoint/CheckpointFailureReason.java | 3 +--
2 files changed, 7 insertions(+), 2 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
index a51a63f..47575ea 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
@@ -650,6 +650,12 @@ public class CheckpointCoordinator {
triggerTasks(request, timestamp, checkpoint)
.exceptionally(
failure -> {
+ LOG.info(
+ "Triggering Checkpoint {} for job {} failed due to {}",
+ checkpoint.getCheckpointID(),
+ job,
+ failure);
+
final CheckpointException cause;
if (failure instanceof CheckpointException) {
cause = (CheckpointException) failure;
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointFailureReason.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointFailureReason.java
index 45a2c9d..75a70c0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointFailureReason.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointFailureReason.java
@@ -47,8 +47,7 @@ public enum CheckpointFailureReason {
CHECKPOINT_DECLINED_TASK_NOT_READY(false, "Checkpoint was declined (tasks not ready)"),
- CHECKPOINT_DECLINED_TASK_CLOSING(
- false, "Checkpoint was declined (task's operators partially closed)"),
+ CHECKPOINT_DECLINED_TASK_CLOSING(false, "Checkpoint was declined (task is closing)"),
CHECKPOINT_DECLINED_TASK_NOT_CHECKPOINTING(false, "Task does not support checkpointing"),
[flink] 01/02: [hotfix][connector] Use null to replace empty
pending transaction in 2pc sink
Posted by ga...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gaoyunhaii pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 83ab9eb1ca4705d3046d4eade3d25a87094afba0
Author: Yun Gao <ga...@gmail.com>
AuthorDate: Mon Sep 13 15:26:22 2021 +0800
[hotfix][connector] Use null to replace empty pending transaction in 2pc sink
This closes #17251.
---
.../functions/sink/TwoPhaseCommitSinkFunction.java | 94 +++++++++-------------
1 file changed, 37 insertions(+), 57 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/TwoPhaseCommitSinkFunction.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/TwoPhaseCommitSinkFunction.java
index d16f4dd..f18f96a 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/TwoPhaseCommitSinkFunction.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/TwoPhaseCommitSinkFunction.java
@@ -54,6 +54,7 @@ import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.stream.Stream;
@@ -91,12 +92,10 @@ public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT> extends RichS
private final ListStateDescriptor<State<TXN, CONTEXT>> stateDescriptor;
/**
- * Current Transaction Holder, including three states: 1. Normal Transaction: created when a new
- * snapshot is taken during normal task running 2. Empty Transaction: created when a new
- * snapshot is taken after the task is finished. At this point, there is no need to initiate
- * real transactions due to no more input data. 3. null: After task/function is closed.
+ * Current Transaction Holder, including two states: 1. Normal Transaction: created when a new
+ * snapshot is taken during normal task running 2. null: After task/function is finished.
*/
- private TransactionHolder<TXN> currentTransactionHolder;
+ @Nullable private TransactionHolder<TXN> currentTransactionHolder;
/** Specifies the maximum time a transaction should remain open. */
private long transactionTimeout = Long.MAX_VALUE;
@@ -335,10 +334,6 @@ public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT> extends RichS
// this is like the pre-commit of a 2-phase-commit transaction
// we are ready to commit and remember the transaction
- checkState(
- currentTransactionHolder != null,
- "bug: no transaction object when performing state snapshot");
-
long checkpointId = context.getCheckpointId();
LOG.debug(
"{} - checkpoint {} triggered, flushing transaction '{}'",
@@ -346,7 +341,7 @@ public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT> extends RichS
context.getCheckpointId(),
currentTransactionHolder);
- if (!currentTransactionHolder.equals(TransactionHolder.empty())) {
+ if (currentTransactionHolder != null) {
preCommit(currentTransactionHolder.handle);
pendingCommitTransactions.put(checkpointId, currentTransactionHolder);
LOG.debug("{} - stored pending transactions {}", name(), pendingCommitTransactions);
@@ -356,7 +351,7 @@ public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT> extends RichS
if (!finished) {
currentTransactionHolder = beginTransactionInternal();
} else {
- currentTransactionHolder = TransactionHolder.empty();
+ currentTransactionHolder = null;
}
LOG.debug("{} - started new transaction '{}'", name(), currentTransactionHolder);
@@ -403,9 +398,8 @@ public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT> extends RichS
}
{
- TXN transaction = operatorState.getPendingTransaction().handle;
-
- if (transaction != null) {
+ if (operatorState.getPendingTransaction() != null) {
+ TXN transaction = operatorState.getPendingTransaction().handle;
recoverAndAbort(transaction);
handledTransactions.add(transaction);
LOG.info(
@@ -545,28 +539,28 @@ public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT> extends RichS
@VisibleForTesting
@Internal
public static final class State<TXN, CONTEXT> {
- protected TransactionHolder<TXN> pendingTransaction;
+ @Nullable protected TransactionHolder<TXN> pendingTransaction;
protected List<TransactionHolder<TXN>> pendingCommitTransactions = new ArrayList<>();
protected Optional<CONTEXT> context;
public State() {}
public State(
- TransactionHolder<TXN> pendingTransaction,
+ @Nullable TransactionHolder<TXN> pendingTransaction,
List<TransactionHolder<TXN>> pendingCommitTransactions,
Optional<CONTEXT> context) {
- this.context = requireNonNull(context, "context is null");
- this.pendingTransaction =
- requireNonNull(pendingTransaction, "pendingTransaction is null");
+ this.pendingTransaction = pendingTransaction;
this.pendingCommitTransactions =
requireNonNull(pendingCommitTransactions, "pendingCommitTransactions is null");
+ this.context = requireNonNull(context, "context is null");
}
+ @Nullable
public TransactionHolder<TXN> getPendingTransaction() {
return pendingTransaction;
}
- public void setPendingTransaction(TransactionHolder<TXN> pendingTransaction) {
+ public void setPendingTransaction(@Nullable TransactionHolder<TXN> pendingTransaction) {
this.pendingTransaction = pendingTransaction;
}
@@ -598,17 +592,9 @@ public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT> extends RichS
State<?, ?> state = (State<?, ?>) o;
- if (pendingTransaction != null
- ? !pendingTransaction.equals(state.pendingTransaction)
- : state.pendingTransaction != null) {
- return false;
- }
- if (pendingCommitTransactions != null
- ? !pendingCommitTransactions.equals(state.pendingCommitTransactions)
- : state.pendingCommitTransactions != null) {
- return false;
- }
- return context != null ? context.equals(state.context) : state.context == null;
+ return Objects.equals(pendingTransaction, state.pendingTransaction)
+ && Objects.equals(pendingCommitTransactions, state.pendingCommitTransactions)
+ && Objects.equals(context, state.context);
}
@Override
@@ -633,8 +619,6 @@ public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT> extends RichS
private final TXN handle;
- private static final TransactionHolder<?> EMPTY = new TransactionHolder<>(null, -1);
-
/**
* The system time when {@link #handle} was created. Used to determine if the current
* transaction has exceeded its timeout specified by {@link #transactionTimeout}.
@@ -651,11 +635,6 @@ public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT> extends RichS
return clock.millis() - transactionStartTime;
}
- @SuppressWarnings("unchecked")
- public static <TXN> TransactionHolder<TXN> empty() {
- return (TransactionHolder<TXN>) EMPTY;
- }
-
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -667,10 +646,8 @@ public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT> extends RichS
TransactionHolder<?> that = (TransactionHolder<?>) o;
- if (transactionStartTime != that.transactionStartTime) {
- return false;
- }
- return handle != null ? handle.equals(that.handle) : that.handle == null;
+ return transactionStartTime == that.transactionStartTime
+ && Objects.equals(handle, that.handle);
}
@Override
@@ -701,7 +678,7 @@ public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT> extends RichS
private final TypeSerializer<TXN> transactionSerializer;
private final TypeSerializer<CONTEXT> contextSerializer;
- private final boolean supportEmptyPendingTransaction;
+ private final boolean supportNullPendingTransaction;
public StateSerializer(
TypeSerializer<TXN> transactionSerializer,
@@ -712,10 +689,10 @@ public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT> extends RichS
public StateSerializer(
TypeSerializer<TXN> transactionSerializer,
TypeSerializer<CONTEXT> contextSerializer,
- boolean supportEmptyPendingTransaction) {
+ boolean supportNullPendingTransaction) {
this.transactionSerializer = checkNotNull(transactionSerializer);
this.contextSerializer = checkNotNull(contextSerializer);
- this.supportEmptyPendingTransaction = supportEmptyPendingTransaction;
+ this.supportNullPendingTransaction = supportNullPendingTransaction;
}
@Override
@@ -728,7 +705,7 @@ public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT> extends RichS
return new StateSerializer<>(
transactionSerializer.duplicate(),
contextSerializer.duplicate(),
- supportEmptyPendingTransaction);
+ supportNullPendingTransaction);
}
@Override
@@ -740,9 +717,11 @@ public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT> extends RichS
public State<TXN, CONTEXT> copy(State<TXN, CONTEXT> from) {
final TransactionHolder<TXN> pendingTransaction = from.getPendingTransaction();
final TransactionHolder<TXN> copiedPendingTransaction =
- new TransactionHolder<>(
- transactionSerializer.copy(pendingTransaction.handle),
- pendingTransaction.transactionStartTime);
+ pendingTransaction == null
+ ? null
+ : new TransactionHolder<>(
+ transactionSerializer.copy(pendingTransaction.handle),
+ pendingTransaction.transactionStartTime);
final List<TransactionHolder<TXN>> copiedPendingCommitTransactions = new ArrayList<>();
for (TransactionHolder<TXN> txn : from.getPendingCommitTransactions()) {
@@ -812,17 +791,18 @@ public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT> extends RichS
}
private void serializePendingTransaction(
- TransactionHolder<TXN> pendingTransaction, DataOutputView target)
+ @Nullable TransactionHolder<TXN> pendingTransaction, DataOutputView target)
throws IOException {
- if (supportEmptyPendingTransaction) {
- target.writeBoolean(pendingTransaction.handle != null);
+ if (supportNullPendingTransaction) {
+ target.writeBoolean(pendingTransaction != null);
} else {
checkState(
- pendingTransaction.handle != null,
- "Received empty pending transaction while the serializer does not support empty pending transction.");
+ pendingTransaction != null,
+ "Received null pending transaction while the serializer "
+ + "does not support null pending transaction.");
}
- if (pendingTransaction.handle != null) {
+ if (pendingTransaction != null) {
transactionSerializer.serialize(pendingTransaction.handle, target);
target.writeLong(pendingTransaction.transactionStartTime);
}
@@ -831,10 +811,10 @@ public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT> extends RichS
private TransactionHolder<TXN> deserializePendingTransaction(DataInputView source)
throws IOException {
boolean hasPendingTransaction =
- supportEmptyPendingTransaction ? source.readBoolean() : true;
+ supportNullPendingTransaction ? source.readBoolean() : true;
if (!hasPendingTransaction) {
- return TransactionHolder.empty();
+ return null;
}
TXN pendingTxnHandle = transactionSerializer.deserialize(source);