[
https://issues.apache.org/jira/browse/FLINK-4190?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15387767#comment-15387767
]
ASF GitHub Bot commented on FLINK-4190:
---------------------------------------
Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/2269
Very good work!
I know we discussed before whether to check for inactivity in a different
thread or in `invoke()`. There's actually a third option that I'm showcasing in
the PR I did against your PR. 😃 The StreamTask already has a TimerService that
can be set for testing. If we use the appropriate methods in the bucketing sink
then we get testability with a settable clock for free.
I also added a `BucketSinkTest`. It think it would be good if the
`BucketSinkITCase` and `BucketingSinkMultipleActiveBucketsCase` could be folded
into this one because having the ITCases means having a lot of overhead and our
build is already taking quite long.
What do you think?
> Generalise RollingSink to work with arbitrary buckets
> -----------------------------------------------------
>
> Key: FLINK-4190
> URL: https://issues.apache.org/jira/browse/FLINK-4190
> Project: Flink
> Issue Type: Improvement
> Components: filesystem-connector, Streaming Connectors
> Reporter: Josh Forman-Gornall
> Assignee: Josh Forman-Gornall
> Priority: Minor
>
> The current RollingSink implementation appears to be intended for writing to
> directories that are bucketed by system time (e.g. minutely) and to only be
> writing to one file within one bucket at any point in time. When the system
> time determines that the current bucket should be changed, the current bucket
> and file are closed and a new bucket and file are created. The sink cannot be
> used for the more general problem of writing to arbitrary buckets, perhaps
> determined by an attribute on the element/tuple being processed.
> There are three limitations which prevent the existing sink from being used
> for more general problems:
> - Only bucketing by the current system time is supported, and not by e.g. an
> attribute of the element being processed by the sink.
> - Whenever the sink sees a change in the bucket being written to, it flushes
> the file and moves on to the new bucket. Therefore the sink cannot have more
> than one bucket/file open at a time. Additionally the checkpointing mechanics
> only support saving the state of one active bucket and file.
> - The sink determines that it should 'close' an active bucket and file when
> the bucket path changes. We need another way to determine when a bucket has
> become inactive and needs to be closed.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
