[
https://issues.apache.org/jira/browse/FLINK-29678?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17619427#comment-17619427
]
luoyuxia commented on FLINK-29678:
----------------------------------
In stream mode, with auto compaction, the pipeline for writing is
CompactFileWriter , CompactCoordinator, CompactOperator, PartitionCommitter.
if the datastream is bounded, CompactFileWriter write file1, file2, then call
method endInput. `CompactCoordinator` is expected to pack fiel1, file2 to
downstream's CompactOperator. But `CompactCoordinator` won't do that since it
has no `endInput` method. So that, file1, file2 will never be compacted which
is always Invisible to user. So, the data in file1, file2 will loss.
To fix it, we need to add a method `endInput` in `CompactCoordinator` to pack
the remain files which are written between last snapshot and endinput.
> Data may loss when sink bounded stream into filesystem with auto compact
> enabled in streaming mode
> ---------------------------------------------------------------------------------------------------
>
> Key: FLINK-29678
> URL: https://issues.apache.org/jira/browse/FLINK-29678
> Project: Flink
> Issue Type: Bug
> Components: Connectors / FileSystem
> Affects Versions: 1.15.0
> Reporter: luoyuxia
> Priority: Major
>
> In stream mode, when writing bounded data stream into filesystem with auto
> compactation enabel, the data may loss.
> We can reproduce it by adding one code line `'auto-compaction'='true' ` in
> `FileSystemITCaseBase#open` to enable auto compact.
> {code:java}
> tableEnv.executeSql(
> s"""
> |create table partitionedTable (
> | x string,
> | y int,
> | a int,
> | b bigint,
> | c as b + 1
> |) partitioned by (a, b) with (
> | 'connector' = 'filesystem',
> | 'auto-compaction'='true', // added line to enable auto compaction.
> | 'path' = '$getScheme://$resultPath',
> | ${formatProperties().mkString(",\n")}
> |)
> """.stripMargin
> ) {code}
> Then the test `StreamFileSystemTestCsvITCase#testPartialDynamicPartition`
> will fail with the assert failure:
> {code:java}
> java.lang.AssertionError:
> Expected :List(x18,18)
> Actual :List() {code}
> There is no data has been written into the table.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
