[FLINK-17587][filesystem] Filesystem streaming sink support commit success file #12062

JingsongLi · 2020-05-11T02:10:15Z

What is the purpose of the change

Committing a partition is to notify the downstream application that the partition has finished writing, the partition is ready to be read.
Add “.succes” file to directory (success file name is configurable too)

Brief change log

This PR is based on [FLINK-17590][connector/filesystem] Support bucket lifecycle listener in streaming file sink buckets #12053
StreamingFileWriter(parallelism) -> StreamingFileCommitter(Single task to commit partition)

Verifying this change

StreamFileSystemTestCsvITCase

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? JavaDocs

flinkbot · 2020-05-11T02:12:23Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit e0a38b8 (Fri Oct 16 10:31:47 UTC 2020)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2020-05-11T02:27:00Z

CI report:

89b4355 UNKNOWN
1965797 Azure: FAILURE
e0a38b8 UNKNOWN

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

lirui-apache

@JingsongLi Thanks for the great efforts. I have left some comments.

...rc/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/StreamingFileSink.java

...k-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOptions.java

...ime-blink/src/main/java/org/apache/flink/table/filesystem/stream/StreamingFileCommitter.java

...ime-blink/src/main/java/org/apache/flink/table/filesystem/stream/PartitionCommitManager.java

...time-blink/src/main/java/org/apache/flink/table/filesystem/stream/PartitionCommitPolicy.java

...ime-blink/src/main/java/org/apache/flink/table/filesystem/stream/StreamingFileCommitter.java

lirui-apache · 2020-05-11T12:20:25Z

...untime-blink/src/main/java/org/apache/flink/table/filesystem/stream/StreamingFileWriter.java

+	}
+
+	@Override
+	public void close() throws Exception {


Is it possible that there's some pending data between close and the last notifyCheckpointComplete?

I'll modify to endInput and support commit pending data.

...me-blink/src/main/java/org/apache/flink/table/filesystem/stream/SuccessFileCommitPolicy.java

pnowojski · 2020-05-12T12:16:06Z

...untime-blink/src/main/java/org/apache/flink/table/filesystem/stream/StreamingFileWriter.java

+ *
+ * <p>See {@link StreamingFileCommitter}.
+ */
+public class StreamingFileWriter extends AbstractStreamOperator<CommitMessage> implements


This class is copy pasting quite a bit of code/fields from StreamingFileSink. Can not we extract a common abstraction or re-use one in the another?

I've discussed with @gaoyunhaii offline, mainly because feel that there is too little code, so don't reuse it.
But +1 for abstraction. I've extracted a helper from StreamingFileSink. See
36fa3d2

lirui-apache

Thanks @JingsongLi for the update. Left some minor comments.

lirui-apache · 2020-05-15T12:57:40Z

...k-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOptions.java

+							" 'partition-time': extract time from partition," +
+							" if 'watermark' > 'partition-time' + 'delay', will commit the partition." +
+							" 'process-time': use processing time, if 'current processing time' > " +
+							"'partition creation time' + 'delay', will commit the partition.");


I think we should explain how "partition creation time" is determined.

"partition directory creation time"

lirui-apache · 2020-05-15T13:12:32Z

...k-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOptions.java

+			key("sink.partition-commit.policy.class")
+					.stringType()
+					.noDefaultValue()
+					.withDescription("The partition commit policy class for implement PartitionCommitPolicy interface.");


Does this only work custom commit policy? If so it should be mentioned in the description and reflected in the config name.

I will add comment, it is OK, has been mentioned in policy kind.

lirui-apache · 2020-05-15T13:21:19Z

...ble-runtime-blink/src/main/java/org/apache/flink/table/filesystem/PartitionCommitPolicy.java

+/**
+ * Policy for commit a partition.
+ *
+ * <p>The implemented commit method needs to be reentrant because the same partition may be


Suggested change

* <p>The implemented commit method needs to be reentrant because the same partition may be

* <p>The implemented commit method needs to be idempotent because the same partition may be

lirui-apache · 2020-05-15T13:27:55Z

...k-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOptions.java

+					.withDescription("Policy to commit a partition is to notify the downstream" +
+							" application that the partition has finished writing, the partition" +
+							" is ready to be read." +
+							" metastore: add partition to metastore." +


If users choose metastore policy, don't they need to specify/provide a TableMetaStoreFactory implementation?

I will add comment: Only work with hive table, it is empty implementation for file system table.

lirui-apache

LGTM

…FileSink

…ccess file

rmetzger added the review=description? label May 11, 2020

JingsongLi force-pushed the successFile branch from 38d4e66 to 8ea1504 Compare May 11, 2020 06:11

rmetzger added the component=<none> label May 11, 2020

JingsongLi force-pushed the successFile branch from 8ea1504 to e5127c6 Compare May 11, 2020 08:54

lirui-apache reviewed May 11, 2020

View reviewed changes

JingsongLi force-pushed the successFile branch 2 times, most recently from 7c0c835 to e7040a4 Compare May 12, 2020 10:10

pnowojski reviewed May 12, 2020

View reviewed changes

JingsongLi force-pushed the successFile branch 5 times, most recently from 0d278fa to e2b2ede Compare May 14, 2020 12:19

lirui-apache reviewed May 15, 2020

View reviewed changes

JingsongLi force-pushed the successFile branch from e2b2ede to b5c0419 Compare May 15, 2020 14:24

lirui-apache approved these changes May 16, 2020

View reviewed changes

JingsongLi force-pushed the successFile branch from b5c0419 to 1965797 Compare May 16, 2020 03:07

JingsongLi added 2 commits May 16, 2020 13:33

[FLINK-17587][runtime] Extract StreamingFileSinkHelper from Streaming…

27349ac

…FileSink

[FLINK-17587][filesystem] Filesystem streaming sink support commit su…

e0a38b8

…ccess file

JingsongLi force-pushed the successFile branch from 1965797 to e0a38b8 Compare May 16, 2020 05:33

JingsongLi merged commit b843480 into apache:master May 16, 2020

JingsongLi deleted the successFile branch November 5, 2020 09:46

flinkbot added component=Connectors/FileSystem and removed component=<none> labels Mar 17, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-17587][filesystem] Filesystem streaming sink support commit success file #12062

[FLINK-17587][filesystem] Filesystem streaming sink support commit success file #12062

JingsongLi commented May 11, 2020

flinkbot commented May 11, 2020 •

edited

flinkbot commented May 11, 2020 •

edited

lirui-apache left a comment

lirui-apache May 11, 2020

JingsongLi May 12, 2020

pnowojski May 12, 2020

JingsongLi May 13, 2020 •

edited

lirui-apache left a comment

lirui-apache May 15, 2020

JingsongLi May 15, 2020

lirui-apache May 15, 2020

JingsongLi May 15, 2020 •

edited

lirui-apache May 15, 2020

lirui-apache May 15, 2020

JingsongLi May 15, 2020

lirui-apache left a comment

	* <p>The implemented commit method needs to be reentrant because the same partition may be
	* <p>The implemented commit method needs to be idempotent because the same partition may be

[FLINK-17587][filesystem] Filesystem streaming sink support commit success file #12062

[FLINK-17587][filesystem] Filesystem streaming sink support commit success file #12062

Conversation

JingsongLi commented May 11, 2020

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented May 11, 2020 • edited

Automated Checks

Review Progress

flinkbot commented May 11, 2020 • edited

CI report:

lirui-apache left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

JingsongLi May 13, 2020 • edited

Choose a reason for hiding this comment

lirui-apache left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

JingsongLi May 15, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lirui-apache left a comment

Choose a reason for hiding this comment

flinkbot commented May 11, 2020 •

edited

flinkbot commented May 11, 2020 •

edited

JingsongLi May 13, 2020 •

edited

JingsongLi May 15, 2020 •

edited