Trigger closing time partition after time some event time have been observed. about kafka-connect-hdfs

[ewencp]

@krisskross This seems like basically what the Partitioner classes and partitioner.class config was designed for. They don't close the files explicitly, but they allow you to divide data across files, and once you hit rotation, the file will be closed anyway.

Is there a specific case you're thinking of where you need to close the file sooner? I think one common example might be a time-based partitioner where timestamps are guaranteed to be monotonically increasing, in which case you know when it is safe to rotate a file (although note that the requirement that it is monotonically increasing is harder to guarantee in practice than you may think). Is there some other example you're thinking of that isn't addressed by the existing approach, or are you just trying to reduce the latency of delivery for the final file?

[krisskross]

The use case i'm seeking is closing files earlier so that they can be processed by batch jobs as early as possible. Also it might be possible to get larger files (ideally one per partition) which have the benefit of more efficient compression and faster batch processing.

We use an modified version of HourlyPartitioner that looks at event time (monotonically increasing) of the log and we can determine when no more events will arrive for a certain partition/hour by looking at the event time (with exceptions). But there is no mechanism to trigger the close and move the file to the final directory.

[ewencp]

Sure, so maybe a way to implement this would be to add a method to the Partitioner interface to allow it to indicate what files are now safe to close, and have the connector invoke it after each record? I think it'll make tracking the necessary state a bit more complicated (we normally commit all outstanding data, but in this case we'd change several states to handle only a subset of the outstanding data), but probably won't make things much more complicated.

[krisskross]

Yes, sounds like reasonable way forward. Maybe as default method on the interface in order to not break backward compatibility?