[GitHub] [flink] XComp commented on a diff in pull request #4910: [FLINK-7784] [kafka-producer] Don't fail TwoPhaseCommitSinkFunction when failing to commit during job recovery

[GitBox <gi...@apache.org>]

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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org