[
https://issues.apache.org/jira/browse/FLINK-4534?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16596461#comment-16596461
]
ASF GitHub Bot commented on FLINK-4534:
---------------------------------------
azagrebin commented on issue #4482: [FLINK-4534] Fix synchronization issue in
BucketingSink
URL: https://github.com/apache/flink/pull/4482#issuecomment-416993974
Taking into account the last comment from [@kl0u in Jira
issue](https://issues.apache.org/jira/browse/FLINK-4534?focusedCommentId=16550597&page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#comment-16550597),
we can close this PR and create another if needed to remove all
`synchronized (state.bucketStates)` in `BucketingSink`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Lack of synchronization in BucketingSink#restoreState()
> -------------------------------------------------------
>
> Key: FLINK-4534
> URL: https://issues.apache.org/jira/browse/FLINK-4534
> Project: Flink
> Issue Type: Bug
> Components: Streaming Connectors
> Reporter: Ted Yu
> Assignee: zhangminglei
> Priority: Major
> Labels: pull-request-available
>
> Iteration over state.bucketStates is protected by synchronization in other
> methods, except for the following in restoreState():
> {code}
> for (BucketState<T> bucketState : state.bucketStates.values()) {
> {code}
> and following in close():
> {code}
> for (Map.Entry<String, BucketState<T>> entry :
> state.bucketStates.entrySet()) {
> closeCurrentPartFile(entry.getValue());
> {code}
> w.r.t. bucketState.pendingFilesPerCheckpoint , there is similar issue
> starting line 752:
> {code}
> Set<Long> pastCheckpointIds =
> bucketState.pendingFilesPerCheckpoint.keySet();
> LOG.debug("Moving pending files to final location on restore.");
> for (Long pastCheckpointId : pastCheckpointIds) {
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
