[
https://issues.apache.org/jira/browse/FLINK-41060?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Martijn Visser updated FLINK-41060:
-----------------------------------
Description:
With the exactly-once {{JdbcSink}}, {{JdbcWriter.prepareCommit()}} prepares the
XA branch, resets {{pendingRecords}} and returns the committable, after which
{{JdbcWriter.snapshotState()}} sees {{pendingRecords == false}} and calls
{{commitTxUntil(lastCheckpointId)}}, which commits the branch that was just
prepared. The data is committed at the barrier, before the checkpoint
completes. A checkpoint that fails after the barrier passed the sink leaves its
records committed, and the restart replays and writes them again.
The committer does not compensate: {{JdbcCommitableSerializer}} only writes the
{{Xid}}, so a deserialized committable carries no transaction and
{{JdbcCommitter.commit()}} falls back to {{commitTx()}} on its own empty
{{XaTransaction}}. With writer and committer in one chain the committable still
carries the writer's transaction object, which is why
{{ExactlyOnceJdbcSinkTest}} passes; with {{pipeline.operator-chaining=false}}
the same job leaves its only branch in-doubt and commits no row (Derby: 0 of 10
rows, 1 in-doubt branch).
The writer should only prepare the branch at the barrier and hand its xid to
the committer, which commits it once the checkpoint completed, on its own
connection. A prepared branch keeps its locks until then, and a writer chained
to the committer can block on them and keep the committer from running. So the
exactly-once committer runs in its own task: the committables pass a no-op map
chained to the writer and go to the first committer subtask with {{global()}},
which keeps the writer's operator id for savepoints without a uid.
was:
With the exactly-once {{JdbcSink}}, {{JdbcWriter.prepareCommit()}} prepares the
XA branch, resets {{pendingRecords}} and returns the committable, after which
{{JdbcWriter.snapshotState()}} sees {{pendingRecords == false}} and calls
{{commitTxUntil(lastCheckpointId)}}, which commits the branch that was just
prepared. The data is committed at the barrier, before the checkpoint
completes. A checkpoint that fails after the barrier passed the sink leaves its
records committed, and the restart replays and writes them again.
The committer does not compensate: {{JdbcCommitableSerializer}} only writes the
{{Xid}}, so a deserialized committable carries no transaction and
{{JdbcCommitter.commit()}} falls back to {{commitTx()}} on its own empty
{{XaTransaction}}. With writer and committer in one chain the committable still
carries the writer's transaction object, which is why
{{ExactlyOnceJdbcSinkTest}} passes; with {{pipeline.operator-chaining=false}}
the same job leaves its only branch in-doubt and commits no row (Derby: 0 of 10
rows, 1 in-doubt branch).
We should let {{snapshotState}} commit only when {{prepareCommit}} found
nothing to prepare, and let {{JdbcCommitter.commit}} commit the handed xid
through {{xaConnectionProvider.commit(xid, true)}}.
{{ExactlyOnceJdbcWriterTest.testCheckpoint}} asserts the barrier-time
visibility today and needs to change with it.
> Exactly-once JdbcSink commits its XA branch at the checkpoint barrier
> ---------------------------------------------------------------------
>
> Key: FLINK-41060
> URL: https://issues.apache.org/jira/browse/FLINK-41060
> Project: Flink
> Issue Type: Bug
> Components: Connectors / JDBC
> Affects Versions: jdbc-3.2.0, jdbc-3.3.0, jdbc-4.0.0, jdbc-4.1.0,
> jdbc-3.4.0
> Reporter: Martijn Visser
> Assignee: Martijn Visser
> Priority: Critical
> Fix For: jdbc-5.0.0
>
>
> With the exactly-once {{JdbcSink}}, {{JdbcWriter.prepareCommit()}} prepares
> the XA branch, resets {{pendingRecords}} and returns the committable, after
> which {{JdbcWriter.snapshotState()}} sees {{pendingRecords == false}} and
> calls {{commitTxUntil(lastCheckpointId)}}, which commits the branch that was
> just prepared. The data is committed at the barrier, before the checkpoint
> completes. A checkpoint that fails after the barrier passed the sink leaves
> its records committed, and the restart replays and writes them again.
> The committer does not compensate: {{JdbcCommitableSerializer}} only writes
> the {{Xid}}, so a deserialized committable carries no transaction and
> {{JdbcCommitter.commit()}} falls back to {{commitTx()}} on its own empty
> {{XaTransaction}}. With writer and committer in one chain the committable
> still carries the writer's transaction object, which is why
> {{ExactlyOnceJdbcSinkTest}} passes; with {{pipeline.operator-chaining=false}}
> the same job leaves its only branch in-doubt and commits no row (Derby: 0 of
> 10 rows, 1 in-doubt branch).
> The writer should only prepare the branch at the barrier and hand its xid to
> the committer, which commits it once the checkpoint completed, on its own
> connection. A prepared branch keeps its locks until then, and a writer
> chained to the committer can block on them and keep the committer from
> running. So the exactly-once committer runs in its own task: the committables
> pass a no-op map chained to the writer and go to the first committer subtask
> with {{global()}}, which keeps the writer's operator id for savepoints
> without a uid.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
