[GitHub] [flink] GJL 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>]

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

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