Skip to content

Flink - Log warning when in table upsert mode for Flink 1.12 due to correctness issues #4519

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
kbendick wants to merge 16 commits into from
Closed

Flink - Log warning when in table upsert mode for Flink 1.12 due to correctness issues #4519

Changes from all commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
9f1603d
Flink - Disallow upsert mode for Flink 1.12 due to correctness issues
kbendick Apr 6, 2022
afbfd65
Use proper sentence structure
kbendick Apr 11, 2022
a0319f6
Remove upsert related tests in Flink 1.12 and add Assume statement th…
kbendick Apr 11, 2022
27bc9c5
Use Assert and not Assume so we don't skip tests that have upsert ena…
kbendick Apr 11, 2022
1330708
Revert V1 FlinkSink from throwing to logging an error message
kbendick May 12, 2022
3d9483b
Remove test for checking if upsert mode is disabled
kbendick May 12, 2022
57f41d0
Remove test assumption that upsert mode is not enabled in 1.12
kbendick May 12, 2022
9c0e69b
Add back in original tests and line them up
kbendick May 12, 2022
d7b7a37
checkstyle
kbendick May 12, 2022
6e014f4
finish checkstyle
kbendick May 12, 2022
4cb0939
Final checkstyle
kbendick May 12, 2022
e35cdb7
Change warning message to be more direct
kbendick May 12, 2022
f711cde
Merge old warning message and new
kbendick May 12, 2022
161e16d
Use table sink in place of job
kbendick May 12, 2022
211fa7a
Dont use a variable for logging as it breaks checkstyle
kbendick May 12, 2022
15319b5
Fix typo in comment
kbendick May 12, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 16 additions & 1 deletion flink/v1.12/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -375,8 +375,23 @@ private SingleOutputStreamOperator<WriteResult> appendWriter(DataStream<RowData>
boolean upsertMode = upsert || PropertyUtil.propertyAsBoolean(table.properties(),
UPSERT_ENABLED, UPSERT_ENABLED_DEFAULT);

// Validate the equality fields and partition fields if we enable the upsert mode.
// `upsert` mode should not be used in Flink 1.12 due to correctness issues.
// As part of a patch release, apache-iceberg-flink-runtime_1.12:0.13.2,
// it has been decided the best course of action would be to log a warning
// asking people to upgrade, as Flink 1.12 has been deprecated in upstream Apache Flink
// for some time as well as will be removed in the next major Iceberg release, Iceberg 0.14.0.
//
// But we allow the configuration given that it's a patch release and the change would be otherwise
// too breaking for a patch release.
//
// See in https://github.com/apache/iceberg/pull/4364 for more information.
if (upsertMode) {
LOG.error(
"This table sink is running in upsert mode. Upsert mode should not be used with Flink 1.12 because " +
"it will write incorrect delete file metadata, which could prevent deletes from being correctly" +
"applied. Upgrading to Flink 1.13+ is recommended. " +
"To safely use Flink 1.12, set manifest metrics to counts only.");

Preconditions.checkState(!overwrite,
"OVERWRITE mode shouldn't be enable when configuring to use UPSERT data stream.");
Preconditions.checkState(!equalityFieldIds.isEmpty(),
Expand Down