XComp commented on code in PR #4910:
URL: https://github.com/apache/flink/pull/4910#discussion_r922227177
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/TwoPhaseCommitSinkFunction.java:
##########
@@ -312,20 +347,104 @@ public void
initializeState(FunctionInitializationContext context) throws Except
}
this.pendingCommitTransactions.clear();
- currentTransaction = beginTransaction();
- LOG.debug("{} - started new transaction '{}'", name(),
currentTransaction);
+ currentTransactionHolder = beginTransactionInternal();
+ LOG.debug("{} - started new transaction '{}'", name(),
currentTransactionHolder);
+ }
+
+ /**
+ * This method must be the only place to call {@link
#beginTransaction()} to ensure that the
+ * {@link TransactionHolder} is created at the same time.
+ */
+ private TransactionHolder<TXN> beginTransactionInternal() throws
Exception {
+ return new TransactionHolder<>(beginTransaction(),
clock.millis());
+ }
+
+ /**
+ * This method must be the only place to call {@link
#recoverAndCommit(Object)} to ensure that
+ * the configuration parameters {@link #transactionTimeout} and
+ * {@link #ignoreFailuresAfterTransactionTimeout} are respected.
+ */
+ private void recoverAndCommitInternal(TransactionHolder<TXN>
transactionHolder) {
+ try {
+ logWarningIfTimeoutAlmostReached(transactionHolder);
+ recoverAndCommit(transactionHolder.handle);
+ } catch (final Exception e) {
Review Comment:
Cool thanks for clarification. That makes sense. And thanks for the pointer
to the sneaky throw example. 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
