XComp commented on code in PR #4910:
URL: https://github.com/apache/flink/pull/4910#discussion_r921872724
##########
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:
@GJL @pnowojski I came across this method today and was initially confused
that this is actually possible, i.e. catching and then possibly re-throwing a
checked `Exception` without mentioning it in the method signature. But
apparently, this is valid Java as a special case of `Exception` handling where
the checked `Exception` is actually a `RuntimeException`.
That said, this `catch` block should actually focus on `RuntimeExceptions`
because `logWarningIfTimeoutAlmostReached` and `recoverAndCommit` both are not
considering checked `Exceptions`. I'm just asking because I got curious whether
I miss something here and this `catch` was done intentionally or whether this
minor thing just slipped through...
--
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]
