eskabetxe commented on code in PR #259:
URL:
https://github.com/apache/flink-connector-jdbc/pull/259#discussion_r4426314007
##########
flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/JdbcExactlyOnceOptions.java:
##########
@@ -27,15 +27,18 @@
* JDBC exactly once sink options.
*
* <p><b>maxCommitAttempts</b> - maximum number of commit attempts to make per
transaction; must be
- * > 0; state size is proportional to the product of max number of in-flight
snapshots and this
- * number.
+ * > 0. The committer retries a transient failure right away, and Flink stops
after {@code
+ * sink.committer.retries} retries, so a transaction gets at most the smaller
of maxCommitAttempts
+ * and {@code sink.committer.retries} + 1 attempts.
*
- * <p><b>allowOutOfOrderCommits</b> - If true, all prepared transactions will
be attempted to commit
- * regardless of any transient failures during this operation. This may lead
to inconsistency.
- * Default: false.
+ * <p><b>allowOutOfOrderCommits</b> - If true, all prepared transactions
restored from a writer
Review Comment:
allowOutOfOrderCommits has become almost a no-op with this PR. Since this is
a @PublicEvolving option, users who set it get no signal that its contract
silently narrowed.
Could we mark withWithAllowOutOfOrderCommits (and the getter) @Deprecated
pointing at this change, or at least call it out in the release notes?
Javadoc-only, but a compile-time hint feels right.
##########
flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/datasource/transactions/xa/XaTransaction.java:
##########
@@ -141,6 +151,52 @@ public void checkState() {
"inconsistent internal state");
}
+ /**
+ * Rolls back the prepared transactions of this writer from the given
checkpoint on. A failed
+ * attempt may have left them prepared, the restart replays their records,
and the committer
+ * only commits transactions of earlier checkpoints. It runs whether or
not {@code
+ * discoverAndRollbackOnRecovery} is set.
+ */
+ public void rollbackPreparedTxs(long checkpointId) {
+ Collection<Xid> recovered;
+ try {
+ recovered = xaConnectionProvider.recover();
+ } catch (Exception e) {
+ LOG.warn("unable to recover prepared transactions", e);
Review Comment:
If recover() fails we continue and the writer will most likely block on the
locks of exactly the transactions we couldn't enumerate, which is the hang this
method exists to prevent.
Not a regression (pre-PR was silent too), but could we escalate this to
LOG.error or even fail the task here?
##########
flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/sink/committer/JdbcCommitter.java:
##########
@@ -21,40 +21,51 @@
import org.apache.flink.api.connector.sink2.Committer;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions;
-import
org.apache.flink.connector.jdbc.core.datastream.sink.writer.JdbcWriterState;
import
org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
import
org.apache.flink.connector.jdbc.datasource.connections.xa.XaConnectionProvider;
-import
org.apache.flink.connector.jdbc.datasource.transactions.xa.XaTransaction;
import
org.apache.flink.connector.jdbc.datasource.transactions.xa.domain.TransactionId;
+import
org.apache.flink.connector.jdbc.datasource.transactions.xa.exceptions.TransientXaException;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkRuntimeException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.transaction.xa.Xid;
+
import java.io.IOException;
import java.util.Collection;
+import java.util.OptionalLong;
/** The responsible for committing the {@link JdbcCommitable}. */
@Internal
public class JdbcCommitter implements Committer<JdbcCommitable> {
- private transient XaTransaction jdbcTransaction;
+ private static final Logger LOG =
LoggerFactory.getLogger(JdbcCommitter.class);
+
+ private transient XaConnectionProvider xaConnectionProvider;
+ private transient boolean opened;
Review Comment:
Nit: if xaConnectionProvider.open() throws (line 79-83), opened stays false
and close() skips provider.close(), leaking a partially established connection.
Consider setting opened = true before calling open(), or closing in the
catch. Pre-existing pattern elsewhere, so fine to leave.
##########
flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/datasource/transactions/xa/domain/TransactionId.java:
##########
@@ -143,6 +143,14 @@ public TransactionId withAttemptsIncremented() {
jobId, subtaskId, numberOfSubtasks, checkpointId, attempts +
1, restored);
}
+ public byte[] getJobId() {
Review Comment:
Nit: returns the internal mutable array. Consistent with
getGlobalTransactionId(), so not asking for a clone.
But recoverState() and isOwnFromCheckpoint() now depend on this for
correctness, so maybe a // do not modify note at least.
##########
flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/datasource/transactions/xa/XaTransaction.java:
##########
@@ -51,7 +52,7 @@ public class XaTransaction implements Serializable,
AutoCloseable {
private final XaConnectionProvider xaConnectionProvider;
private final JdbcExactlyOnceOptions exactlyOnceOptions;
- // checkpoints and the corresponding transactions waiting for completion
notification from JM
+ // prepared xids from the restored writer state, committed on open
Review Comment:
Coverage suggestion: this migration path (prepared entries in a pre-PR
writer state being committed by commitTx() in open()) doesn't seem to have a
test.
I only see the empty-list behavior in ExactlyOnceJdbcWriterTest.
A unit test that feeds XaTransaction.open() a JdbcWriterState with a
non-empty prepared list (prepared transaction in the DB) and asserts it lands
after restore would close the last upgrade-loop cell.
##########
flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/datasource/transactions/xa/XaTransaction.java:
##########
@@ -141,6 +151,52 @@ public void checkState() {
"inconsistent internal state");
}
+ /**
+ * Rolls back the prepared transactions of this writer from the given
checkpoint on. A failed
+ * attempt may have left them prepared, the restart replays their records,
and the committer
+ * only commits transactions of earlier checkpoints. It runs whether or
not {@code
+ * discoverAndRollbackOnRecovery} is set.
+ */
+ public void rollbackPreparedTxs(long checkpointId) {
+ Collection<Xid> recovered;
+ try {
+ recovered = xaConnectionProvider.recover();
+ } catch (Exception e) {
+ LOG.warn("unable to recover prepared transactions", e);
+ return;
+ }
+ for (Xid xid : recovered) {
+ if (isOwnFromCheckpoint(xid, checkpointId)) {
+ LOG.info("rolling back transaction of a failed attempt,
xid={}", xid);
+ xaConnectionProvider.rollback(xid);
+ }
+ }
+ }
+
+ /**
+ * A transaction id names the job, the subtask and the parallelism, but
not the sink. Another
+ * exactly-once sink of the job at the same parallelism can't run, as it
starts the same xids.
+ * At another parallelism its xids differ only in the parallelism, so that
has to match too.
+ */
+ private boolean isOwnFromCheckpoint(Xid xid, long checkpointId) {
+ if (xid.getFormatId() != baseTransaction.getFormatId()
+ || !Arrays.equals(
+ xid.getGlobalTransactionId(),
baseTransaction.getGlobalTransactionId())) {
+ return false;
+ }
+ try {
+ TransactionId transactionId =
+ TransactionId.createFromXid(
+ xid.getFormatId(),
+ xid.getGlobalTransactionId(),
+ xid.getBranchQualifier());
+ return transactionId.getNumberOfSubtasks() ==
baseTransaction.getNumberOfSubtasks()
Review Comment:
The consequence for users is that a job that fails at parallelism P and is
restarted at P' will not roll back the failed attempt's prepared transactions
(as their numberOfSubtasks doesn't match), so they keep holding locks and
showing up in recover() until someone cleans them up.
The javadoc and testRollbackPreparedTxsKeepsOtherParallelism pin the
behavior, but the user-facing docs don't mention it. One sentence in the new
docs paragraph ("after a rescale, prepared transactions of the failed attempt
may need discoverAndRollbackOnRecovery or a manual rollback") would make the
sharp edge visible.
--
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]
