Github user ryanpersaud commented on the issue:
https://github.com/apache/storm/pull/1044
Now you've piqued my curiosity - what is it that you don't like about
rotation actions?
Writers are only stored in the writers map, so once they are removed, there
is no way for FSDataOutputStream.close() to be invoked by Storm code.
Furthermore, there are no finalize() methods defined in HDFSWriter,
AbstractHDFSWiter, FSDataOutputStream, or DataOutputStream (relying on the
finalizer to close the writer would be a bad practice anyway).
I used code in the HiveBolt as a starting point for handling retiring
writers (see below). Please let me know what you think. I'll also create a
Jira tonight and a PR once we coalesce on a solution.
// Setting accessOrder to true to facilitate obtaining the LRU entry
writers = new LinkedHashMap<>(this.maxOpenFiles, 0.75f, true);
private AbstractHDFSWriter getOrCreateWriter(String writerKey, Tuple
tuple) throws IOException {
AbstractHDFSWriter writer = writers.get( writerKey );
if (writer == null) {
LOG.debug("Creating Writer for writerKey : " + writerKey);
Path pathForNextFile = getBasePathForNextFile(tuple);
writer = makeNewWriter(pathForNextFile, tuple);
if (writers.size() > (this.maxOpenFiles - 1)) {
LOG.info("cached writers size {} exceeded maxOpenFiles {}
", writers.size(), this.maxOpenFiles);
retireEldestWriter();
}
this.writers.put(writerKey, writer);
}
return writer;
}
/**
* Locate writer that has not been used for longest time and retire it
*/
private void retireEldestWriter() throws IOException {
LOG.info("Attempting to close eldest writer");
Iterator<String> iterator = this.writers.keySet().iterator();
if (iterator.hasNext()) {
String eldestKey = iterator.next();
doRotationAndRemoveWriter(eldestKey,
this.writers.get(eldestKey));
}
}
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
