Skip to content
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.

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

[core] Add a http-report action to reporting partition done to remote servers. #4862

Merged
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
30 changes: 27 additions & 3 deletions docs/content/flink/sql-write.md
Original file line number Diff line number Diff line change
Expand Up @@ -261,11 +261,13 @@ CREATE TABLE my_partitioned_table (
'partition.time-interval'='1 d',
'partition.idle-time-to-done'='15 m',
'partition.mark-done-action'='done-partition'
-- You can also customize a PartitionMarkDoneAction to mark the partition completed.
-- 'partition.mark-done-action'='done-partition,custom',
-- 'partition.mark-done-action.custom.class'='org.apache.paimon.CustomPartitionMarkDoneAction'
);
```

You can also customize a PartitionMarkDoneAction to mark the partition completed.
- partition.mark-done-action: custom
- partition.mark-done-action.custom.class: The partition mark done class for implement PartitionMarkDoneAction interface (e.g. org.apache.paimon.CustomPartitionMarkDoneAction).

Define a class CustomPartitionMarkDoneAction to implement the PartitionMarkDoneAction interface.
```java
package org.apache.paimon;
Expand All @@ -282,6 +284,28 @@ public class CustomPartitionMarkDoneAction implements PartitionMarkDoneAction {
}
```

Paimon also support http-report partition mark done action, this action will report the partition to the remote http server.
- partition.mark-done-action: http-report
- partition.mark-done-action.http.url : Action will report the partition to the remote http server.
- partition.mark-done-action.http.timeout : Http client connection timeout and default is 5s.
- partition.mark-done-action.http.params : Http client request params in the request body json.

Http Post request body :
```json
{
"table": "table fullName",
"path": "table location path",
"partition": "mark done partition",
"params" : "custom params"
}
```
Http Response body :
```json
{
"result": "success"
}
```

1. Firstly, you need to define the time parser of the partition and the time interval between partitions in order to
determine when the partition can be properly marked done.
2. Secondly, you need to define idle-time, which determines how long it takes for the partition to have no new data,
Expand Down
20 changes: 19 additions & 1 deletion docs/layouts/shortcodes/generated/core_configuration.html
Original file line number Diff line number Diff line change
Expand Up @@ -663,14 +663,32 @@
<td><h5>partition.mark-done-action</h5></td>
<td style="word-wrap: break-word;">"success-file"</td>
<td>String</td>
<td>Action to mark a partition done is to notify the downstream application that the partition has finished writing, the partition is ready to be read.<br />1. 'success-file': add '_success' file to directory.<br />2. 'done-partition': add 'xxx.done' partition to metastore.<br />3. 'mark-event': mark partition event to metastore.<br />4. 'custom': use policy class to create a mark-partition policy.<br />Both can be configured at the same time: 'done-partition,success-file,mark-event,custom'.</td>
<td>Action to mark a partition done is to notify the downstream application that the partition has finished writing, the partition is ready to be read.<br />1. 'success-file': add '_success' file to directory.<br />2. 'done-partition': add 'xxx.done' partition to metastore.<br />3. 'mark-event': mark partition event to metastore.<br />4. 'http-report': report partition mark done to remote http server.<br />5. 'custom': use policy class to create a mark-partition policy.<br />Both can be configured at the same time: 'done-partition,success-file,mark-event,custom'.</td>
</tr>
<tr>
<td><h5>partition.mark-done-action.custom.class</h5></td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
<td>The partition mark done class for implement PartitionMarkDoneAction interface. Only work in custom mark-done-action.</td>
</tr>
<tr>
<td><h5>partition.mark-done-action.http.params</h5></td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
<td>Http client request parameters will be written to the request body, this can only be used by http-report partition mark done action.</td>
</tr>
<tr>
<td><h5>partition.mark-done-action.http.timeout</h5></td>
<td style="word-wrap: break-word;">5 s</td>
<td>Duration</td>
<td>Http client connection timeout, this can only be used by http-report partition mark done action.</td>
</tr>
<tr>
<td><h5>partition.mark-done-action.http.url</h5></td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
<td>Mark done action will reports the partition to the remote http server, this can only be used by http-report partition mark done action.</td>
</tr>
<tr>
<td><h5>partition.timestamp-formatter</h5></td>
<td style="word-wrap: break-word;">(none)</td>
Expand Down
65 changes: 64 additions & 1 deletion paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
Expand Down Expand Up @@ -1220,7 +1221,10 @@ public class CoreOptions implements Serializable {
.text("3. 'mark-event': mark partition event to metastore.")
.linebreak()
.text(
"4. 'custom': use policy class to create a mark-partition policy.")
"4. 'http-report': report partition mark done to remote http server.")
.linebreak()
.text(
"5. 'custom': use policy class to create a mark-partition policy.")
.linebreak()
.text(
"Both can be configured at the same time: 'done-partition,success-file,mark-event,custom'.")
Expand All @@ -1234,6 +1238,27 @@ public class CoreOptions implements Serializable {
"The partition mark done class for implement"
+ " PartitionMarkDoneAction interface. Only work in custom mark-done-action.");

public static final ConfigOption<String> PARTITION_MARK_DONE_ACTION_URL =
LinMingQiang marked this conversation as resolved.
Show resolved Hide resolved
key("partition.mark-done-action.http.url")
.stringType()
.noDefaultValue()
.withDescription(
"Mark done action will reports the partition to the remote http server, this can only be used by http-report partition mark done action.");

public static final ConfigOption<Duration> PARTITION_MARK_DONE_ACTION_TIMEOUT =
key("partition.mark-done-action.http.timeout")
.durationType()
.defaultValue(Duration.ofSeconds(5))
.withDescription(
"Http client connection timeout, this can only be used by http-report partition mark done action.");

public static final ConfigOption<String> PARTITION_MARK_DONE_ACTION_PARAMS =
key("partition.mark-done-action.http.params")
.stringType()
.noDefaultValue()
.withDescription(
"Http client request parameters will be written to the request body, this can only be used by http-report partition mark done action.");

public static final ConfigOption<Boolean> METASTORE_PARTITIONED_TABLE =
key("metastore.partitioned-table")
.booleanType()
Expand Down Expand Up @@ -2262,10 +2287,28 @@ public String partitionTimestampPattern() {
return options.get(PARTITION_TIMESTAMP_PATTERN);
}

public String httpReportMarkDoneActionUrl() {
return options.get(PARTITION_MARK_DONE_ACTION_URL);
}

public Duration httpReportMarkDoneActionTimeout() {
return options.get(PARTITION_MARK_DONE_ACTION_TIMEOUT);
}

public String httpReportMarkDoneActionParams() {
return options.get(PARTITION_MARK_DONE_ACTION_PARAMS);
}

public String partitionMarkDoneCustomClass() {
return options.get(PARTITION_MARK_DONE_CUSTOM_CLASS);
}

public Set<PartitionMarkDoneAction> partitionMarkDoneActions() {
return Arrays.stream(options.get(PARTITION_MARK_DONE_ACTION).split(","))
.map(x -> PartitionMarkDoneAction.valueOf(x.replace('-', '_').toUpperCase()))
.collect(Collectors.toCollection(HashSet::new));
}

public String consumerId() {
String consumerId = options.get(CONSUMER_ID);
if (consumerId != null && consumerId.isEmpty()) {
Expand Down Expand Up @@ -3163,4 +3206,24 @@ public enum MaterializedTableRefreshStatus {
ACTIVATED,
SUSPENDED
}

/** Partition mark done actions. */
public enum PartitionMarkDoneAction {
SUCCESS_FILE("success-file"),
DONE_PARTITION("done-partition"),
MARK_EVENT("mark-event"),
HTTP_REPORT("http-report"),
CUSTOM("custom");

private final String value;

PartitionMarkDoneAction(String value) {
this.value = value;
}

@Override
public String toString() {
return value;
}
}
}
Loading
Loading