You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "iceiceiceCN (via GitHub)" <gi...@apache.org> on 2023/06/20 08:46:44 UTC
[GitHub] [flink] iceiceiceCN opened a new pull request, #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
iceiceiceCN opened a new pull request, #22831:
URL: https://github.com/apache/flink/pull/22831
## What is the purpose of the change
By allowing the passing of parameters, the custom PartitionCommitPolicy becomes more flexible and customizable. This enables user to enhance their custom PartitionCommitPolicy by including additional functionality, such as passing monitoring parameters to track the files associated with each commit.
## Brief change log
- *`SINK_PARTITION_COMMIT_POLICY_CLASS_PARAMETERS` parameter added
- *To create a custom commit policy through the PartitionCommitPolicyFactory, you have the option to include `SINK_PARTITION_COMMIT_POLICY_CLASS_PARAMETERS` as a parameter.
## Verifying this change
This change is a trivial rework / code cleanup without any test coverage.
## 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)`: (yes)
- 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, 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? (docs)
Key: **sink.partition-commit.policy.class**
Default: (none)
Type: String
Description: A (semicolon-separated) list of patterns that for custom policy class. Only work in custom commit policy
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #22831:
URL: https://github.com/apache/flink/pull/22831#issuecomment-1598379618
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "e6bb176227fede51898b4ca67189c3cccb357c29",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "e6bb176227fede51898b4ca67189c3cccb357c29",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e6bb176227fede51898b4ca67189c3cccb357c29 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #22831:
URL: https://github.com/apache/flink/pull/22831#discussion_r1243092293
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSink.java:
##########
@@ -456,7 +456,8 @@ private DataStreamSink<?> createBatchCompactSink(
new PartitionCommitPolicyFactory(
conf.get(HiveOptions.SINK_PARTITION_COMMIT_POLICY_KIND),
conf.get(HiveOptions.SINK_PARTITION_COMMIT_POLICY_CLASS),
- conf.get(HiveOptions.SINK_PARTITION_COMMIT_SUCCESS_FILE_NAME));
+ conf.get(HiveOptions.SINK_PARTITION_COMMIT_SUCCESS_FILE_NAME),
+ conf.get(HiveOptions.SINK_PARTITION_COMMIT_POLICY_CLASS_PARAMETERS));
Review Comment:
```suggestion
conf.get(FileSystemConnectorOptions.SINK_PARTITION_COMMIT_POLICY_CLASS_PARAMETERS));
```
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSink.java:
##########
@@ -610,7 +611,8 @@ private DataStreamSink<Row> createBatchNoCompactSink(
new PartitionCommitPolicyFactory(
conf.get(HiveOptions.SINK_PARTITION_COMMIT_POLICY_KIND),
conf.get(HiveOptions.SINK_PARTITION_COMMIT_POLICY_CLASS),
- conf.get(HiveOptions.SINK_PARTITION_COMMIT_SUCCESS_FILE_NAME)));
+ conf.get(HiveOptions.SINK_PARTITION_COMMIT_SUCCESS_FILE_NAME),
+ conf.get(HiveOptions.SINK_PARTITION_COMMIT_POLICY_CLASS_PARAMETERS)));
Review Comment:
dito
##########
docs/content.zh/docs/connectors/table/filesystem.md:
##########
@@ -451,6 +451,13 @@ public class HourPartTimeExtractor implements PartitionTimeExtractor {
<td>String</td>
<td> 实现 PartitionCommitPolicy 接口的分区提交策略类。只有在 custom 提交策略下才使用该类。</td>
</tr>
+ <tr>
+ <td><h5>sink.partition-commit.policy.class.parameters</h5></td>
+ <td style="word-wrap: break-word;">(无)</td>
+ <td>String</td>
+ <td> 传入 custom 提交策略类的字符串参数, 要求多个参数之间用分号分隔, 比如 'param1;param2',
Review Comment:
```suggestion
<td> 传入 custom 提交策略类构造器的参数, 多个参数之间用分号分隔, 比如 'param1;param2',
```
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionCommitPolicyFactory.java:
##########
@@ -63,11 +67,23 @@ public List<PartitionCommitPolicy> createPolicyChain(
successFileName, fsSupplier.get());
case PartitionCommitPolicy.CUSTOM:
try {
- return (PartitionCommitPolicy)
- cl.loadClass(customClass).newInstance();
- } catch (ClassNotFoundException
- | IllegalAccessException
- | InstantiationException e) {
+ if (parameters != null && !parameters.isEmpty()) {
+ String[] paramStrings = parameters.toArray(new String[parameters.size()]);
Review Comment:
```suggestion
String[] paramStrings = parameters.toArray(new String[0]);
```
##########
docs/content/docs/connectors/table/filesystem.md:
##########
@@ -470,6 +470,14 @@ The partition commit policy defines what action is taken when partitions are com
<td>String</td>
<td>The partition commit policy class for implement PartitionCommitPolicy interface. Only work in custom commit policy.</td>
</tr>
+ <tr>
+ <td><h5>sink.partition-commit.policy.class.parameters</h5></td>
+ <td>optional</td>
+ <td>yes</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The custom commit policy class can accept a string argument, which can include multiple arguments separated by semicolons. For example, 'param1;param2'. The string argument will be split into a list (['param1', 'param2']) and passed as constructor parameters to the custom commit policy class.</td>
Review Comment:
```suggestion
<td>The parameters passed to the constructor of the custom commit policy, with multiple parameters separated by semicolons, such as 'param1;param2'. For example, 'param1;param2'. The configuration value will be split into a list (['param1', 'param2']) and passed to the constructor of the custom commit policy class.</td>
```
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionCommitPolicyFactory.java:
##########
@@ -63,11 +67,23 @@ public List<PartitionCommitPolicy> createPolicyChain(
successFileName, fsSupplier.get());
case PartitionCommitPolicy.CUSTOM:
try {
- return (PartitionCommitPolicy)
- cl.loadClass(customClass).newInstance();
- } catch (ClassNotFoundException
- | IllegalAccessException
- | InstantiationException e) {
+ if (parameters != null && !parameters.isEmpty()) {
Review Comment:
```suggestion
if (!CollectionUtil.isNullOrEmpty(parameters)) {
```
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionCommitPolicyFactory.java:
##########
@@ -63,11 +67,23 @@ public List<PartitionCommitPolicy> createPolicyChain(
successFileName, fsSupplier.get());
case PartitionCommitPolicy.CUSTOM:
try {
- return (PartitionCommitPolicy)
- cl.loadClass(customClass).newInstance();
- } catch (ClassNotFoundException
- | IllegalAccessException
- | InstantiationException e) {
+ if (parameters != null && !parameters.isEmpty()) {
+ String[] paramStrings = parameters.toArray(new String[parameters.size()]);
+ Class<?>[] classes = new Class<?>[parameters.size()];
+ for (int i = 0; i < parameters.size(); i++) {
+ classes[i] = String.class;
+ }
+ return (PartitionCommitPolicy)
+ cl.loadClass(customClass)
+ .getConstructor(classes)
+ .newInstance(paramStrings);
Review Comment:
```suggestion
.newInstance((Object[]) paramStrings);
```
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveOptions.java:
##########
@@ -137,6 +138,9 @@ public class HiveOptions {
public static final ConfigOption<String> SINK_PARTITION_COMMIT_POLICY_CLASS =
FileSystemConnectorOptions.SINK_PARTITION_COMMIT_POLICY_CLASS;
+ public static final ConfigOption<List<String>> SINK_PARTITION_COMMIT_POLICY_CLASS_PARAMETERS =
Review Comment:
I don't think we need to introduce a option to HiveOption. We can reuse the `FileSystemConnectorOptions.SINK_PARTITION_COMMIT_POLICY_CLASS_PARAMETERS`
##########
docs/content.zh/docs/connectors/table/filesystem.md:
##########
@@ -451,6 +451,13 @@ public class HourPartTimeExtractor implements PartitionTimeExtractor {
<td>String</td>
<td> 实现 PartitionCommitPolicy 接口的分区提交策略类。只有在 custom 提交策略下才使用该类。</td>
</tr>
+ <tr>
+ <td><h5>sink.partition-commit.policy.class.parameters</h5></td>
+ <td style="word-wrap: break-word;">(无)</td>
+ <td>String</td>
+ <td> 传入 custom 提交策略类的字符串参数, 要求多个参数之间用分号分隔, 比如 'param1;param2',
+ 该字符串将被切分为列表(['param1','param2'])并传给 custom 提交策略类的构造器。该项为可选项。</td>
Review Comment:
```suggestion
该字符串将被切分为列表(['param1','param2'])并传给 custom 提交策略类的构造器。该项为可选项，不配置的话将使用类的默认构造方法。</td>
```
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemCommitterTest.java:
##########
@@ -258,6 +263,38 @@ void testPartitionPathNotExist() throws Exception {
assertThat(outputPath.toFile().list()).isEqualTo(new String[0]);
}
+ @Test
+ void testCustomCommitPolicyWithParameters() {
Review Comment:
Sorry, I mean IT not UT. You can refer to this pr https://github.com/apache/flink/pull/17245.
I think we can add test in `FileSystemTableSinkTest`.
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemConnectorOptions.java:
##########
@@ -220,6 +221,15 @@ public class FileSystemConnectorOptions {
"The partition commit policy class for implement"
+ " PartitionCommitPolicy interface. Only work in custom commit policy");
+ public static final ConfigOption<List<String>> SINK_PARTITION_COMMIT_POLICY_CLASS_PARAMETERS =
+ key("sink.partition-commit.policy.class.parameters")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ "A semicolon-separated string of parameters for the custom commit policy class"
Review Comment:
We can resue the doc in the above:
```
The parameters passed to the constructor of the custom commit policy, with multiple parameters separated by semicolons, such as 'param1;param2'.
For example, 'param1;param2'. The configuration value will be split into a list (['param1', 'param2']) and passed to the constructor of the custom commit policy class.
```
##########
flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveOptions.java:
##########
@@ -137,6 +138,9 @@ public class HiveOptions {
public static final ConfigOption<String> SINK_PARTITION_COMMIT_POLICY_CLASS =
FileSystemConnectorOptions.SINK_PARTITION_COMMIT_POLICY_CLASS;
+ public static final ConfigOption<List<String>> SINK_PARTITION_COMMIT_POLICY_CLASS_PARAMETERS =
Review Comment:
Btw, these two lines in HiveOption can also be delegated to Filesystem option:
```
public static final ConfigOption<String> SINK_PARTITION_COMMIT_POLICY_CLASS =
FileSystemConnectorOptions.SINK_PARTITION_COMMIT_POLICY_CLASS;
public static final ConfigOption<String> SINK_PARTITION_COMMIT_SUCCESS_FILE_NAME =
FileSystemConnectorOptions.SINK_PARTITION_COMMIT_SUCCESS_FILE_NAME;
```
Could you please also help remove them in this pr?
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemCommitterTest.java:
##########
@@ -258,6 +263,38 @@ void testPartitionPathNotExist() throws Exception {
assertThat(outputPath.toFile().list()).isEqualTo(new String[0]);
}
+ @Test
+ void testCustomCommitPolicyWithParameters() {
Review Comment:
Btw, I don't think we need the test in here if we have a IT.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #22831:
URL: https://github.com/apache/flink/pull/22831#discussion_r1241007352
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionCommitPolicyFactory.java:
##########
@@ -37,12 +39,22 @@ public class PartitionCommitPolicyFactory implements Serializable {
private final String policyKind;
private final String customClass;
private final String successFileName;
+ private final List<String> parameters;
public PartitionCommitPolicyFactory(
Review Comment:
We can remove this constructor.
Please also rememer to update the Hive part.
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemConnectorOptions.java:
##########
@@ -220,6 +221,15 @@ public class FileSystemConnectorOptions {
"The partition commit policy class for implement"
+ " PartitionCommitPolicy interface. Only work in custom commit policy");
+ public static final ConfigOption<List<String>> SINK_PARTITION_COMMIT_POLICY_CLASS_PARAMETERS =
+ key("sink.partition-commit.policy.class.parameters")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ "A (semicolon-separated) list of patterns that for custom policy class"
Review Comment:
May be we should highlight that the all the parameter(if have) of this policy class should be string.
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionCommitPolicyFactory.java:
##########
@@ -63,11 +75,23 @@ public List<PartitionCommitPolicy> createPolicyChain(
successFileName, fsSupplier.get());
case PartitionCommitPolicy.CUSTOM:
try {
- return (PartitionCommitPolicy)
- cl.loadClass(customClass).newInstance();
- } catch (ClassNotFoundException
- | IllegalAccessException
- | InstantiationException e) {
+ if (parameters != null && !parameters.isEmpty()) {
Review Comment:
I'm wondering if change it to
```
parameters != null
```
Can it also get the constructor with empty arguments?
If so, we should consider to change to `parameters != null` to make code more clear.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia merged pull request #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #22831:
URL: https://github.com/apache/flink/pull/22831
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] iceiceiceCN closed pull request #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
Posted by "iceiceiceCN (via GitHub)" <gi...@apache.org>.
iceiceiceCN closed pull request #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
URL: https://github.com/apache/flink/pull/22831
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] iceiceiceCN commented on pull request #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
Posted by "iceiceiceCN (via GitHub)" <gi...@apache.org>.
iceiceiceCN commented on PR #22831:
URL: https://github.com/apache/flink/pull/22831#issuecomment-1598399473
> @iceiceiceCN I think it makes sense first to discuss this on the mailing list, given that it would introduce a new configuration option. Besides that, this PR lacks documentation. Can you briefly discuss this on the Dev mailing list?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on pull request #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #22831:
URL: https://github.com/apache/flink/pull/22831#issuecomment-1620116201
@iceiceiceCN The ci fail. Please fix it, you can run 'mvn spotless:apply' to fix these violations. After that, i'll help merge..
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] iceiceiceCN commented on pull request #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
Posted by "iceiceiceCN (via GitHub)" <gi...@apache.org>.
iceiceiceCN commented on PR #22831:
URL: https://github.com/apache/flink/pull/22831#issuecomment-1598461375
> @iceiceiceCN I think it makes sense first to discuss this on the mailing list, given that it would introduce a new configuration option. Besides that, this PR lacks documentation. Can you briefly discuss this on the Dev mailing list?
Hi MartijnVisser, I've emailed the Dev mailing list to discuss it, thanks for your guidance. : )
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] iceiceiceCN commented on a diff in pull request #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
Posted by "iceiceiceCN (via GitHub)" <gi...@apache.org>.
iceiceiceCN commented on code in PR #22831:
URL: https://github.com/apache/flink/pull/22831#discussion_r1241576587
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionCommitPolicyFactory.java:
##########
@@ -63,11 +75,23 @@ public List<PartitionCommitPolicy> createPolicyChain(
successFileName, fsSupplier.get());
case PartitionCommitPolicy.CUSTOM:
try {
- return (PartitionCommitPolicy)
- cl.loadClass(customClass).newInstance();
- } catch (ClassNotFoundException
- | IllegalAccessException
- | InstantiationException e) {
+ if (parameters != null && !parameters.isEmpty()) {
Review Comment:
If users input
`'sink.partition-commit.policy.class.parameters'='',`
parameters != null but parameters will empty,
so yes, it maybe get the constructor with empty arguments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] iceiceiceCN commented on a diff in pull request #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
Posted by "iceiceiceCN (via GitHub)" <gi...@apache.org>.
iceiceiceCN commented on code in PR #22831:
URL: https://github.com/apache/flink/pull/22831#discussion_r1244996088
##########
flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/table/FileSystemCommitterTest.java:
##########
@@ -258,6 +263,38 @@ void testPartitionPathNotExist() throws Exception {
assertThat(outputPath.toFile().list()).isEqualTo(new String[0]);
}
+ @Test
+ void testCustomCommitPolicyWithParameters() {
Review Comment:
> Btw, I don't think we need the test in here if we have a IT.
Yeah, it's a good idea, thanks.
I have removed UT here and add IT in `FileSystemTableSinkTest`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #22831:
URL: https://github.com/apache/flink/pull/22831#discussion_r1247371004
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionCommitPolicyFactory.java:
##########
@@ -63,11 +70,26 @@ public List<PartitionCommitPolicy> createPolicyChain(
successFileName, fsSupplier.get());
case PartitionCommitPolicy.CUSTOM:
try {
- return (PartitionCommitPolicy)
- cl.loadClass(customClass).newInstance();
+ if (!CollectionUtil.isNullOrEmpty(parameters)) {
Review Comment:
https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=50618&view=results
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] MartijnVisser commented on pull request #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on PR #22831:
URL: https://github.com/apache/flink/pull/22831#issuecomment-1620080961
> @MartijnVisser Do you have any other concern? The discussion thread is in
I don't, except that the CI doesn't pass on the current PR code :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] iceiceiceCN commented on pull request #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
Posted by "iceiceiceCN (via GitHub)" <gi...@apache.org>.
iceiceiceCN commented on PR #22831:
URL: https://github.com/apache/flink/pull/22831#issuecomment-1621190771
@luoyuxia Hi, I fixed it with the command, and the ci was success. : )
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #22831:
URL: https://github.com/apache/flink/pull/22831#discussion_r1246196096
##########
docs/content/docs/connectors/table/filesystem.md:
##########
@@ -470,6 +470,14 @@ The partition commit policy defines what action is taken when partitions are com
<td>String</td>
<td>The partition commit policy class for implement PartitionCommitPolicy interface. Only work in custom commit policy.</td>
</tr>
+ <tr>
+ <td><h5>sink.partition-commit.policy.class.parameters</h5></td>
+ <td>optional</td>
+ <td>yes</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The parameters passed to the constructor of the custom commit policy, with multiple parameters separated by semicolons, such as 'param1;param2'. For example, 'param1;param2'. The configuration value will be split into a list (['param1', 'param2']) and passed to the constructor of the custom commit policy class. This option is optional, if not configured, the default constructor will be used.</td>
Review Comment:
```suggestion
<td>The parameters passed to the constructor of the custom commit policy, with multiple parameters separated by semicolons, such as 'param1;param2'. The configuration value will be split into a list (['param1', 'param2']) and passed to the constructor of the custom commit policy class. This option is optional, if not configured, the default constructor will be used.</td>
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/file/table/FileSystemTableSinkTest.java:
##########
@@ -149,4 +163,54 @@ private static String buildSinkTableSql(
private static String buildInsertIntoSql(String sinkTable, String sourceTable) {
return String.format("INSERT INTO %s SELECT * FROM %s", sinkTable, sourceTable);
}
+
+ @Test
+ public void testFileSystemTableSinkWithCustomCommitPolicy() throws Exception {
Review Comment:
Suggestion, I think the test can be simpiled as following:
```
final String outputTable = "outputTable";
final String customPartitionCommitPolicyClassName = TestCustomCommitPolicy.class.getName();
final TableEnvironment tEnv =
TableEnvironment.create(EnvironmentSettings.inStreamingMode());
String ddl =
"CREATE TABLE %s ("
+ " a INT,"
+ " b STRING,"
+ " d STRING,"
+ " e STRING"
+ ") PARTITIONED BY (d, e) WITH ("
+ "'connector'='filesystem',"
+ "'path'='/tmp',"
+ "'format'='testcsv',"
+ "'sink.partition-commit.delay'='0s',"
+ "'sink.partition-commit.policy.kind'='custom',"
+ "'sink.partition-commit.policy.class'='%s',"
+ "'sink.partition-commit.policy.class.parameters'='test1;test2'"
+ ")";
ddl = String.format(ddl, outputTable, customPartitionCommitPolicyClassName);
tEnv.executeSql(ddl);
tEnv.executeSql(
"insert into outputTable select *"
+ " from (values (1, 'a', '2020-05-03', '3'), "
+ "(2, 'x', '2020-05-03', '4'))")
.await();
Set<String> actualCommittedPaths =
TestCustomCommitPolicy.getCommittedPartitionPathsAndReset();
Set<String> expectedCommittedPaths =
new HashSet<>(
Arrays.asList(
"test1test2", "/tmp/d=2020-05-03/e=3", "/tmp/d=2020-05-03/e=4"));
assertEquals(expectedCommittedPaths, actualCommittedPaths);
```
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemConnectorOptions.java:
##########
@@ -220,6 +221,19 @@ public class FileSystemConnectorOptions {
"The partition commit policy class for implement"
+ " PartitionCommitPolicy interface. Only work in custom commit policy");
+ public static final ConfigOption<List<String>> SINK_PARTITION_COMMIT_POLICY_CLASS_PARAMETERS =
+ key("sink.partition-commit.policy.class.parameters")
+ .stringType()
+ .asList()
+ .noDefaultValue()
+ .withDescription(
+ "The parameters passed to the constructor of the custom commit policy, "
+ + " with multiple parameters separated by semicolons, such as 'param1;param2'."
+ + " For example, 'param1;param2'. The configuration value will be split"
Review Comment:
```suggestion
+ " The configuration value will be split"
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #22831:
URL: https://github.com/apache/flink/pull/22831#discussion_r1246244925
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionCommitPolicyFactory.java:
##########
@@ -63,11 +70,26 @@ public List<PartitionCommitPolicy> createPolicyChain(
successFileName, fsSupplier.get());
case PartitionCommitPolicy.CUSTOM:
try {
- return (PartitionCommitPolicy)
- cl.loadClass(customClass).newInstance();
+ if (!CollectionUtil.isNullOrEmpty(parameters)) {
Review Comment:
The ci fail in here.
```
Jun 29 04:30:25 04:30:25.403 [ERROR] Architecture Violation [Priority: MEDIUM] - Rule 'Connector production code must depend only on public API when outside of connector packages' was violated (1 times):
Jun 29 04:30:25 Method <org.apache.flink.connector.file.table.PartitionCommitPolicyFactory.createPolicyChain(java.lang.ClassLoader, java.util.function.Supplier)> calls method <org.apache.flink.util.CollectionUtil.isNullOrEmpty(java.util.Collection)> in (PartitionCommitPolicyFactory.java:73)
```
Seems we can't call `CollectionUtil.isNullOrEmpty` in here.
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionCommitPolicyFactory.java:
##########
@@ -63,11 +70,26 @@ public List<PartitionCommitPolicy> createPolicyChain(
successFileName, fsSupplier.get());
case PartitionCommitPolicy.CUSTOM:
try {
- return (PartitionCommitPolicy)
- cl.loadClass(customClass).newInstance();
+ if (!CollectionUtil.isNullOrEmpty(parameters)) {
Review Comment:
```
parameters != null && !parameters.isEmpty()
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] iceiceiceCN commented on pull request #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
Posted by "iceiceiceCN (via GitHub)" <gi...@apache.org>.
iceiceiceCN commented on PR #22831:
URL: https://github.com/apache/flink/pull/22831#issuecomment-1620932577
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] iceiceiceCN commented on a diff in pull request #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
Posted by "iceiceiceCN (via GitHub)" <gi...@apache.org>.
iceiceiceCN commented on code in PR #22831:
URL: https://github.com/apache/flink/pull/22831#discussion_r1241576587
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionCommitPolicyFactory.java:
##########
@@ -63,11 +75,23 @@ public List<PartitionCommitPolicy> createPolicyChain(
successFileName, fsSupplier.get());
case PartitionCommitPolicy.CUSTOM:
try {
- return (PartitionCommitPolicy)
- cl.loadClass(customClass).newInstance();
- } catch (ClassNotFoundException
- | IllegalAccessException
- | InstantiationException e) {
+ if (parameters != null && !parameters.isEmpty()) {
Review Comment:
If users input
`'sink.partition-commit.policy.class.parameters'='',`
parameters != null but parameters will empty,
so yes, it maybe get the constructor with empty arguments.
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionCommitPolicyFactory.java:
##########
@@ -63,11 +75,23 @@ public List<PartitionCommitPolicy> createPolicyChain(
successFileName, fsSupplier.get());
case PartitionCommitPolicy.CUSTOM:
try {
- return (PartitionCommitPolicy)
- cl.loadClass(customClass).newInstance();
- } catch (ClassNotFoundException
- | IllegalAccessException
- | InstantiationException e) {
+ if (parameters != null && !parameters.isEmpty()) {
Review Comment:
> I'm wondering if change it to
>
> ```
> parameters != null
> ```
>
> Can it also get the constructor with empty arguments? If so, we should consider to change to `parameters != null` to make code more clear.
If users input
`'sink.partition-commit.policy.class.parameters'='',`
parameters != null but parameters will empty,
so yes, it maybe get the constructor with empty arguments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] iceiceiceCN commented on a diff in pull request #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
Posted by "iceiceiceCN (via GitHub)" <gi...@apache.org>.
iceiceiceCN commented on code in PR #22831:
URL: https://github.com/apache/flink/pull/22831#discussion_r1247352456
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionCommitPolicyFactory.java:
##########
@@ -63,11 +70,26 @@ public List<PartitionCommitPolicy> createPolicyChain(
successFileName, fsSupplier.get());
case PartitionCommitPolicy.CUSTOM:
try {
- return (PartitionCommitPolicy)
- cl.loadClass(customClass).newInstance();
+ if (!CollectionUtil.isNullOrEmpty(parameters)) {
Review Comment:
Ok, and where can I see this ci fail? I hope to learn further : )
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] iceiceiceCN commented on a diff in pull request #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
Posted by "iceiceiceCN (via GitHub)" <gi...@apache.org>.
iceiceiceCN commented on code in PR #22831:
URL: https://github.com/apache/flink/pull/22831#discussion_r1247350845
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/file/table/FileSystemTableSinkTest.java:
##########
@@ -149,4 +163,54 @@ private static String buildSinkTableSql(
private static String buildInsertIntoSql(String sinkTable, String sourceTable) {
return String.format("INSERT INTO %s SELECT * FROM %s", sinkTable, sourceTable);
}
+
+ @Test
+ public void testFileSystemTableSinkWithCustomCommitPolicy() throws Exception {
Review Comment:
More concise! Thanks for your suggesion. 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] luoyuxia commented on a diff in pull request #22831: [FLINK-32388]Add the ability to pass parameters to CUSTOM PartitionCommitPolicy
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #22831:
URL: https://github.com/apache/flink/pull/22831#discussion_r1247371849
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionCommitPolicyFactory.java:
##########
@@ -63,11 +70,26 @@ public List<PartitionCommitPolicy> createPolicyChain(
successFileName, fsSupplier.get());
case PartitionCommitPolicy.CUSTOM:
try {
- return (PartitionCommitPolicy)
- cl.loadClass(customClass).newInstance();
+ if (!CollectionUtil.isNullOrEmpty(parameters)) {
Review Comment:
You can see the it in `CI report:`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org