Add deletion of temporal folders in Aggregation
@@ -285,17 +285,19 @@ public class Aggregation implements IAggregation, Initializable<Aggregation>, Ev
Comparator<T> keyComparator = createKeyComparator(resultClass, allKeys, classLoader);
BufferSerializer<T> bufferSerializer = createBufferSerializer(structure, resultClass,
getKeys(), measures, classLoader);
- if (temporarySortDir == null) {
- try {
- temporarySortDir = Files.createTempDirectory("aggregation_sort_dir");
- } catch (IOException e) {
- throw new UncheckedIOException(e);
- }
- }
- return unsortedStream
+ Path sortDir = (temporarySortDir != null) ? temporarySortDir : createSortDir();
+ StreamProducer<T> stream = unsortedStream
.with(StreamSorter.create(
- StreamSorterStorageImpl.create(executorService, bufferSerializer, temporarySortDir),
+ StreamSorterStorageImpl.create(executorService, bufferSerializer, sortDir),
Function.identity(), keyComparator, false, sorterItemsInMemory));
+
+ stream.getEndOfStream()
+ .whenComplete(($, e) -> {
+ if (temporarySortDir == null) {
+ deleteSortDirSilent(sortDir);
+ }
+ });
+ return stream;
}
private Stage<List<AggregationChunk>> doConsolidation(List<AggregationChunk> chunksToConsolidate) {
@@ -516,6 +518,22 @@ public class Aggregation implements IAggregation, Initializable<Aggregation>, Ev
.thenApply(removedChunks -> AggregationDiff.of(new LinkedHashSet<>(removedChunks), new LinkedHashSet<>(chunks)));
}
+ private Path createSortDir() {
+ try {
+ return Files.createTempDirectory("aggregation_sort_dir");
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ private void deleteSortDirSilent(Path sortDir) {
+ try {
+ Files.delete(sortDir);
+ } catch (IOException e) {
+ logger.warn("Could not delete temporal directory {} : {}", temporarySortDir, e.toString());
+ }
+ }
+
public static String getChunkIds(Iterable<AggregationChunk> chunks) {
List<Long> ids = new ArrayList<>();
for (AggregationChunk chunk : chunks) {
