You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by GitBox <gi...@apache.org> on 2023/01/06 05:02:42 UTC
[GitHub] [inlong] e-mhui opened a new pull request, #7173: [INLONG-7172][Sort] Fix newly table write into iceberg failed
e-mhui opened a new pull request, #7173:
URL: https://github.com/apache/inlong/pull/7173
### Prepare a Pull Request
[INLONG-7172][Sort] Fix newly table write into iceberg failed
- Fixes #7172
### Motivation
When a newly table is added in the Flink job, and then restart the job with status. The newly table can be successfully written to iceberg.
### Modifications
Whether the job has the restored job id when it is restarted.
```java
if (context.isRestored()) {
// Newly table doesn't have restored flink job id.
if (!jobIdState.get().iterator().hasNext()) {
return;
}
/**
* ...
*/
}
```
--
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: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] e-mhui commented on a diff in pull request #7173: [INLONG-7172][Sort] Fix new table write into iceberg failed
Posted by GitBox <gi...@apache.org>.
e-mhui commented on code in PR #7173:
URL: https://github.com/apache/inlong/pull/7173#discussion_r1063193835
##########
inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleFileCommiter.java:
##########
@@ -160,7 +160,7 @@ public void initializeState(FunctionInitializationContext context) throws Except
this.checkpointsState = context.getOperatorStateStore().getListState(stateDescriptor);
this.jobIdState = context.getOperatorStateStore().getListState(jobIdDescriptor);
- if (context.isRestored()) {
+ if (context.isRestored() && jobIdState.get().iterator().hasNext()) {
Review Comment:
done.
--
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: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] e-mhui commented on a diff in pull request #7173: [INLONG-7172][Sort] Fix new table write into iceberg failed
Posted by GitBox <gi...@apache.org>.
e-mhui commented on code in PR #7173:
URL: https://github.com/apache/inlong/pull/7173#discussion_r1063151341
##########
inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleFileCommiter.java:
##########
@@ -161,6 +161,10 @@ public void initializeState(FunctionInitializationContext context) throws Except
this.checkpointsState = context.getOperatorStateStore().getListState(stateDescriptor);
this.jobIdState = context.getOperatorStateStore().getListState(jobIdDescriptor);
if (context.isRestored()) {
+ // New table doesn't have restored flink job id.
+ if (!jobIdState.get().iterator().hasNext()) {
Review Comment:
done.
--
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: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] EMsnap commented on a diff in pull request #7173: [INLONG-7172][Sort] Fix new table write into iceberg failed
Posted by GitBox <gi...@apache.org>.
EMsnap commented on code in PR #7173:
URL: https://github.com/apache/inlong/pull/7173#discussion_r1063184873
##########
inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleFileCommiter.java:
##########
@@ -160,7 +160,7 @@ public void initializeState(FunctionInitializationContext context) throws Except
this.checkpointsState = context.getOperatorStateStore().getListState(stateDescriptor);
this.jobIdState = context.getOperatorStateStore().getListState(jobIdDescriptor);
- if (context.isRestored()) {
+ if (context.isRestored() && jobIdState.get().iterator().hasNext()) {
Review Comment:
keep the comment you submit before pls
--
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: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] EMsnap commented on a diff in pull request #7173: [INLONG-7172][Sort] Fix new table write into iceberg failed
Posted by GitBox <gi...@apache.org>.
EMsnap commented on code in PR #7173:
URL: https://github.com/apache/inlong/pull/7173#discussion_r1063149606
##########
inlong-sort/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleFileCommiter.java:
##########
@@ -161,6 +161,10 @@ public void initializeState(FunctionInitializationContext context) throws Except
this.checkpointsState = context.getOperatorStateStore().getListState(stateDescriptor);
this.jobIdState = context.getOperatorStateStore().getListState(jobIdDescriptor);
if (context.isRestored()) {
+ // New table doesn't have restored flink job id.
+ if (!jobIdState.get().iterator().hasNext()) {
Review Comment:
if (context.isRestored() && jobIdState.get().iterator().hasNext())
--
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: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] EMsnap merged pull request #7173: [INLONG-7172][Sort] Fix new table write into iceberg failed
Posted by GitBox <gi...@apache.org>.
EMsnap merged PR #7173:
URL: https://github.com/apache/inlong/pull/7173
--
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: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org