spuru9 commented on code in PR #259:
URL:
https://github.com/apache/flink-connector-jdbc/pull/259#discussion_r4572643913
##########
flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/datasource/connections/xa/SimpleXaConnectionProvider.java:
##########
@@ -222,6 +226,16 @@ public void rollback(Xid xid) {
this::buildRollbackErrorDesc));
}
+ private void rollbackIfKnown(Xid xid) {
+ try {
+ xaResource.rollback(xid);
+ } catch (XAException e) {
+ if (e.errorCode != XAException.XAER_NOTA) {
+ LOG.warn(formatErrorMessage("rollback", xid, e.errorCode));
+ }
+ }
+ }
Review Comment:
**major:** This only logs when the rollback fails (anything but
`XAER_NOTA`), and `failOrRollback` then calls `result.succeeded(x)`, so the xid
is never retried. A transient `XAER_RMFAIL` leaves the transaction prepared
with its locks, which is what this PR is trying to avoid. Can it rethrow like
`rollback(xid)` does, so transient failures land in the retry list?
```suggestion
private void rollbackIfKnown(Xid xid) {
try {
xaResource.rollback(xid);
} catch (XAException e) {
if (e.errorCode != XAException.XAER_NOTA) {
throw wrapException("rollback", xid, e);
}
}
}
```
##########
flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/JdbcExactlyOnceOptions.java:
##########
@@ -27,15 +27,16 @@
* 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> - deprecated; see {@link
+ * JDBCExactlyOnceOptionsBuilder#withAllowOutOfOrderCommits(boolean)}.
Default: false.
*
* <p><b>recoveredAndRollback</b> - whether to rollback prepared transactions
known to XA RM on
- * startup (after committing <b>known</b> transactions, i.e. restored from
state).
+ * startup (after committing <b>known</b> transactions, i.e. restored from
state). Transactions of
+ * checkpoints before the restored one belong to the committer.
Review Comment:
**minor:** This reads as if the flag controls the rollback of leftovers, but
`recoverAndRollback()` filters with `belongsTo`, which never matches the driver
xids `recover()` returns (FLINK-41101), so it's a no-op right now. The writer
always rolls back its own leftovers in `rollbackPreparedTxs`. Could you say
that here? The NOTE below it describes the old behavior too, so maybe mark it
as legacy.
```suggestion
* <p><b>recoveredAndRollback</b> - currently has no effect on the sink
(FLINK-41101). The writer
* always rolls back its own prepared transactions of the restored
checkpoint or later on start.
* Transactions of checkpoints before the restored one belong to the
committer.
```
##########
flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/datasource/connections/xa/PoolingXaConnectionProvider.java:
##########
@@ -44,8 +45,10 @@
/**
* A "pooling" implementation of {@link XaConnectionProvider}. Some database
implement XA such that
* one connection is limited to a single transaction. As a workaround, this
implementation creates a
- * new XA resource after each xa_start call is made (and associates it with
the xid to commit
- * later).
+ * new XA resource after each xa_start call is made (and associates it with
the xid until the
+ * transaction is prepared). A prepared transaction is committed or rolled
back on any idle XA
+ * resource, so the database must allow that from another connection (for
MySQL, {@code
+ * xa_detach_on_prepare}).
Review Comment:
**nit:** "creates a new XA resource after each xa_start" isn't true anymore,
`start()` takes an idle one from the pool and only creates one when the pool is
empty.
```suggestion
* one connection is limited to a single transaction. As a workaround, this
implementation takes an
* idle XA resource from a pool (or creates one) on each xa_start call and
associates it with the
* xid until the transaction is prepared. A prepared transaction is
committed or rolled back on any
* idle XA resource, so the database must allow that from another connection
(for MySQL, {@code
* xa_detach_on_prepare}).
```
##########
flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/sink/ExactlyOnceJdbcSinkTest.java:
##########
@@ -17,17 +17,286 @@
package org.apache.flink.connector.jdbc.core.datastream.sink;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import
org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.RestartStrategyOptions;
+import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
+import
org.apache.flink.connector.jdbc.datasource.connections.xa.SimpleXaConnectionProvider;
+import org.apache.flink.connector.jdbc.testutils.TableManaged;
+import org.apache.flink.connector.jdbc.testutils.tables.TableRow;
+import org.apache.flink.connector.jdbc.testutils.tables.templates.BooksTable;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.util.RestartStrategyUtils;
+import org.apache.flink.testutils.junit.SharedObjectsExtension;
+import org.apache.flink.testutils.junit.SharedReference;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import javax.transaction.xa.Xid;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static
org.apache.flink.connector.jdbc.testutils.tables.TableBuilder.dbType;
+import static
org.apache.flink.connector.jdbc.testutils.tables.TableBuilder.field;
+import static
org.apache.flink.connector.jdbc.testutils.tables.TableBuilder.tableRow;
+import static org.apache.flink.table.api.DataTypes.DOUBLE;
+import static org.apache.flink.table.api.DataTypes.INT;
+import static org.apache.flink.table.api.DataTypes.VARCHAR;
+import static org.assertj.core.api.Assertions.assertThat;
/** Smoke tests for the {@link JdbcSink} and the underlying classes. */
class ExactlyOnceJdbcSinkTest extends BaseJdbcSinkTest {
+ /** The books table without its key, so that a record written twice shows
up twice. */
+ private static final TableRow RECOVERY_TABLE =
+ tableRow(
+ "RecoverySinkTable",
+ field("id", INT().notNull()),
+ field("title", VARCHAR(50)),
+ field("author", VARCHAR(50)),
+ field("price", dbType("FLOAT"), DOUBLE()),
+ field("qty", INT()));
+
+ private static final long RECOVERY_RECORDS = 100;
+
+ private static final BooksTable UPDATE_TABLE = new
BooksTable("UpdateSinkTable");
+
+ private static final int UPDATES = 10_000;
+
+ @RegisterExtension
+ private final SharedObjectsExtension sharedObjects =
SharedObjectsExtension.create();
+
@Override
protected <T> JdbcSink<T> finishSink(JdbcSinkBuilder<T> builder) {
return builder.withExecutionOptions(
JdbcExecutionOptions.builder().withMaxRetries(0).build())
.buildExactlyOnce(
JdbcExactlyOnceOptions.defaults(),
getMetadata().getXaSourceSupplier());
}
+
+ @Override
+ public List<TableManaged> getManagedTables() {
+ List<TableManaged> tables = new ArrayList<>(super.getManagedTables());
+ tables.add(RECOVERY_TABLE);
+ tables.add(UPDATE_TABLE);
+ return tables;
+ }
+
+ /** A failed test can leave a prepared transaction behind, which holds its
locks. */
+ @AfterEach
+ void rollbackPreparedTransactions() throws SQLException {
+ try (SimpleXaConnectionProvider xa =
+
SimpleXaConnectionProvider.from(getMetadata().buildXaDataSource())) {
+ xa.open();
+ for (Xid xid : xa.recover()) {
+ xa.rollback(xid);
+ }
+ }
+ }
+
+ /**
+ * An operator ahead of the sink fails the second checkpoint after the
sink has prepared its
+ * transaction for it. The restart from the first checkpoint replays the
records of that
+ * transaction, so it must not have been committed.
+ */
+ @Test
+ void testRecoveryAfterAFailedCheckpoint() throws Exception {
+ testRecoveryAfterFailedCheckpoints(2, 1, false);
+ }
+
+ /**
+ * The first checkpoint fails after the sink has prepared its first
transaction. The restart has
+ * no state and starts the same xid again, so it must have been rolled
back.
+ */
+ @Test
+ void testRecoveryAfterAFailedFirstCheckpoint() throws Exception {
+ testRecoveryAfterFailedCheckpoints(1, 1, false);
+ }
+
+ /**
+ * The second checkpoint expires, which is tolerated, and the third one
expires too, which fails
+ * the job. The sink has prepared a transaction for each of them, and the
restart from the first
+ * checkpoint must roll back both.
+ */
+ @Test
+ void testRecoveryAfterATolerableFailedCheckpoint() throws Exception {
+ testRecoveryAfterFailedCheckpoints(2, 2, true);
+ }
+
+ private void testRecoveryAfterFailedCheckpoints(
+ long firstFailingCheckpointId, int failingCheckpoints, boolean
expire)
+ throws Exception {
+ Configuration config = new Configuration();
+ config.set(RestartStrategyOptions.RESTART_STRATEGY, "fixed-delay");
+
config.set(RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS, 1);
+ config.set(RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_DELAY,
Duration.ZERO);
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(config);
+ env.setParallelism(1);
+ env.enableCheckpointing(200L);
+ if (expire) {
+ env.getCheckpointConfig().setCheckpointTimeout(300L);
+
env.getCheckpointConfig().setTolerableCheckpointFailureNumber(failingCheckpoints
- 1);
+ }
+ SharedReference<AtomicInteger> failed = sharedObjects.add(new
AtomicInteger());
+ DataGeneratorSource<BooksTable.BookEntry> books =
+ new DataGeneratorSource<>(
+ ExactlyOnceJdbcSinkTest::book,
+ RECOVERY_RECORDS,
+ // records keep coming while checkpoints expire
+ expire
+ ? RateLimiterStrategy.perSecond(40)
+ : RateLimiterStrategy.perCheckpoint(10),
+ TypeInformation.of(BooksTable.BookEntry.class));
+
+ env.fromSource(books, WatermarkStrategy.noWatermarks(), "books")
+ .map(
+ new FailingSnapshotMapper(
+ failed, firstFailingCheckpointId,
failingCheckpoints, expire))
+ .sinkTo(
+ finishSink(
+ new JdbcSinkBuilder<BooksTable.BookEntry>()
+ .withQueryStatement(
+
RECOVERY_TABLE.getInsertIntoQuery(),
+ (ps, book) -> {
+ ps.setInt(1, book.id);
+ ps.setString(2,
book.title);
+ ps.setString(3,
book.author);
+ ps.setDouble(4,
book.price);
+ ps.setInt(5, book.qty);
+ })));
+ env.execute();
+
+ assertThat(failed.get().get()).isEqualTo(failingCheckpoints);
+ try (SimpleXaConnectionProvider xa =
+
SimpleXaConnectionProvider.from(getMetadata().buildXaDataSource())) {
+ xa.open();
+ assertThat(xa.recover()).isEmpty();
+ }
+ List<Integer> written =
+ RECOVERY_TABLE.selectAllTable(getMetadata()).stream()
+ .map(row -> (Integer) row.getField(0))
+ .collect(Collectors.toList());
+ assertThat(written)
+ .containsExactlyInAnyOrderElementsOf(
+ IntStream.range(0, (int) RECOVERY_RECORDS)
+ .boxed()
+ .collect(Collectors.toList()));
+ }
+
+ /**
+ * Every record updates the same row, so after each checkpoint the writer
needs the lock that
+ * the transaction prepared for that checkpoint holds until the committer
commits it.
+ */
+ @Test
+ void testUpdateRowOfThePreviousCheckpoint() throws Exception {
+ BooksTable.BookEntry book = new BooksTable.BookEntry(0, "Book 0",
"Writer", 0D, 0);
+ try (Connection connection = getMetadata().getConnection();
+ PreparedStatement ps =
+
connection.prepareStatement(UPDATE_TABLE.getInsertIntoQuery())) {
+ UPDATE_TABLE.getStatementBuilder().accept(ps, book);
+ ps.execute();
+ }
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ RestartStrategyUtils.configureNoRestartStrategy(env);
+ env.setParallelism(1);
+ env.enableCheckpointing(50L);
+
+ env.fromSequence(1, UPDATES)
+ .sinkTo(
+ new JdbcSinkBuilder<Long>()
+ .withQueryStatement(
+ "UPDATE UpdateSinkTable SET qty = ?
WHERE id = 0",
+ (ps, qty) -> ps.setInt(1,
qty.intValue()))
+ .withExecutionOptions(
+ JdbcExecutionOptions.builder()
+ .withBatchSize(1)
+ .withMaxRetries(0)
+ .build())
+ .buildExactlyOnce(
+ JdbcExactlyOnceOptions.defaults(),
+ getMetadata().getXaSourceSupplier()));
+ env.execute();
+
+ try (Connection connection = getMetadata().getConnection()) {
+ assertThat(UPDATE_TABLE.selectAllTable(connection))
+ .containsExactly(
+ new BooksTable.BookEntry(
+ book.id, book.title, book.author,
book.price, UPDATES));
+ }
+ }
+
+ private static BooksTable.BookEntry book(Long id) {
+ return new BooksTable.BookEntry(id.intValue(), "Book " + id, "Writer",
id.doubleValue(), 1);
+ }
+
+ /**
+ * Fails the snapshots of the given number of checkpoints, or lets them
expire. A chain takes
+ * its snapshots from the tail to the head, so the sink has already
prepared its transaction for
+ * those checkpoints.
+ */
+ private static class FailingSnapshotMapper
+ implements MapFunction<BooksTable.BookEntry, BooksTable.BookEntry>,
+ CheckpointedFunction {
+
+ private final SharedReference<AtomicInteger> failed;
+ private final long firstFailingCheckpointId;
+ private final int failingCheckpoints;
+ private final boolean expire;
+ private boolean restored;
+
+ FailingSnapshotMapper(
+ SharedReference<AtomicInteger> failed,
+ long firstFailingCheckpointId,
+ int failingCheckpoints,
+ boolean expire) {
+ this.failed = failed;
+ this.firstFailingCheckpointId = firstFailingCheckpointId;
+ this.failingCheckpoints = failingCheckpoints;
+ this.expire = expire;
+ }
+
+ @Override
+ public BooksTable.BookEntry map(BooksTable.BookEntry book) {
+ return book;
+ }
+
+ @Override
+ public void snapshotState(FunctionSnapshotContext context) throws
Exception {
+ if (!restored
+ && context.getCheckpointId() >= firstFailingCheckpointId
+ && failed.get().get() < failingCheckpoints) {
+ failed.get().incrementAndGet();
+ if (expire) {
+ // longer than the checkpoint timeout, short enough for
the next barrier to
+ // reach the sink before the job fails
+ Thread.sleep(400L);
Review Comment:
**minor:** Not a blocker, but this relies on 400 ms outlasting the
checkpoint timeout while ending before the next barrier reaches the sink, which
can flake on a slow CI box. Could it wait on a latch released from
`notifyCheckpointAborted` instead?
--
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]
