MartijnVisser commented on code in PR #259:
URL:
https://github.com/apache/flink-connector-jdbc/pull/259#discussion_r4543266147
##########
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:
The lazy open is new in this PR, so I fixed it here: the committer closes
the provider if `open()` fails. `testCloseConnectionOfFailedOpen` and
`testFailToConnect` fail without it. aa48f483
##########
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:
`getGlobalTransactionId()` builds a new array on every call, so it was only
`getJobId()` that handed out the internal one. It returns a copy now, pinned by
`testGetJobIdReturnsACopy`. 8940a09e
##########
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:
I've deprecated the builder method and the getter, and I'll add it to the
release note. It only still matters when a writer restores a 4.1 state with
more than one prepared transaction: if the first commit fails again, it then
still tries the others. 570badc9
##########
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:
Added to `XaTransactionTest`, on Derby and PostgreSQL, with states of the
released 4.1.0 writer after one and two transient commit failures. 2dfa8b36
##########
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:
Added to the docs, but without `discoverAndRollbackOnRecovery`: its lookup
never matches a driver Xid, so it rolls nothing back. At another parallelism
the writer still rolls back the transaction the restored state names; only
those of later checkpoints stay
(`testOpenAtOtherParallelismKeepsLaterLeftovers`, 8940a09e). FLINK-41101 (#262)
covers them. 627b6dde
##########
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:
Agreed, it fails the task now. MySQL then needs `XA_RECOVER_ADMIN` and
Oracle `SELECT` on `DBA_PENDING_TRANSACTIONS` and `EXECUTE` on `SYS.DBMS_XA`,
also with `withRecoveredAndRollback(false)`, which goes in the release note. A
writer that fails to start also closes its connection now (30d7f3ab). fc188bff
--
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]
