[flink]Added the function of generating labels when Flink Sink is completed … #2469
Conversation
…in batch processing scenarios.
siyangzeng
changed the title
| @@ -301,6 +301,13 @@ public class FlinkConnectorOptions { | |||
| .withDescription( | |||
| "Sink committer memory to control heap memory of global committer."); | |||
|
|
|||
| public static final ConfigOption<Boolean> SINK_FINISH_GENERATAR_TAG = | |||
| ConfigOptions.key("sink.finish.generatar-tag") | |||
There was a problem hiding this comment.
Can this feature be a part of tag.automatic-creation? We can introduce BATCH TagCreationMode.
The downside of this is that it cannot be orthogonal to the automatic creation of tags in streaming writing.
There was a problem hiding this comment.
Yes, we hope to use tags in batch processing scenarios and generate corresponding tags after the current task ends.
Do you mean to add a batch mode in tag.automatic-creation without adding new configuration? If that's the case, I think it's okay
There was a problem hiding this comment.
yes, tag.automatic-creation=batch
| * Commit {@link Committable} for snapshot using the {@link CommitterOperator}. | ||
| * When the task is completed, the corresponding tag is generated. | ||
| */ | ||
| public class SinkFinishGeneratorTagOperator<CommitT, GlobalCommitT> implements |
There was a problem hiding this comment.
Can you create a ITCase for this feature?
| * Commit {@link Committable} for snapshot using the {@link CommitterOperator}. | ||
| * When the task is completed, the corresponding tag is generated. | ||
| */ | ||
| public class SinkFinishGeneratorTagOperator<CommitT, GlobalCommitT> implements |
| SetupableStreamOperator, | ||
| BoundedOneInput { | ||
|
|
||
| private static final String SINK_FINISH_TAG_PREFIX = "sinkFinish-"; |
There was a problem hiding this comment.
Only concern is here, better name?
Maybe batch-write-2023-12-13?
…processing scenarios
|
Compile failed? |
…processing scenarios
|
+1 and merged |
This closes apache#2469 (cherry picked from commit 5a8b597)
Purpose
We need to enhance paimon's capabilities in flink batch processing scenarios, such as generating tags after flink batch processing is completed. This function has a very good practical effect when combined with changelog internally.
Tests
API and Format
Documentation