GJL commented on code in PR #4910:
URL: https://github.com/apache/flink/pull/4910#discussion_r922073308
##########
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:
I think changing `Exception` to `RuntimeException` won't be semantically
equivalent due to the possibility of [sneaky
throws](https://www.baeldung.com/java-sneaky-throws#sneaky). My memory about
this code is a bit fuzzy but I assume there can be sneaky throws from within
`recoverAndCommit` because the `commit` callback is public API.
--
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]
