AggregationChunker extends ForwardingStreamConsumer<T> { public static <C, T> AggregationChunker<C, T> create(AggregationStructure aggregation, List<String> fields, Class<T> recordClass, PartitionPredicate<T> partitionPredicate, AggregationChunkStorage<C> storage, DefiningClassLoader classLoader, int chunkSize) { StreamConsumerSwitcher<T> switcher = StreamConsumerSwitcher.create(); AggregationChunker<C, T> chunker = new AggregationChunker<>(switcher, aggregation, fields, recordClass, partitionPredicate, storage, classLoader, chunkSize); chunker.startNewChunk(); return chunker; } private  AggregationChunker(StreamConsumerSwitcher<T> switcher,
			AggregationStructure aggregation, List<String> fields,
			Class<T> recordClass, PartitionPredicate<T> partitionPredicate,
			AggregationChunkStorage<C> storage,
			DefiningClassLoader classLoader,
			int chunkSize); static AggregationChunker<C, T> create(AggregationStructure aggregation, List<String> fields,
			Class<T> recordClass, PartitionPredicate<T> partitionPredicate,
			AggregationChunkStorage<C> storage,
			DefiningClassLoader classLoader,
			int chunkSize); Promise<List<AggregationChunk>> getResult();  }
@Test public void testSupplierWithError() { List<StreamConsumer> listConsumers = new ArrayList<>(); AggregationChunkStorage<Long> aggregationChunkStorage = new AggregationChunkStorage<Long>() { long chunkId; final List items = new ArrayList(); @Override public <T> Promise<StreamSupplier<T>> read(AggregationStructure aggregation, List<String> fields, Class<T> recordClass, Long chunkId, DefiningClassLoader classLoader) { return Promise.of(StreamSupplier.ofIterable(items)); } @Override public <T> Promise<StreamConsumer<T>> write(AggregationStructure aggregation, List<String> fields, Class<T> recordClass, Long chunkId, DefiningClassLoader classLoader) { StreamConsumerToList<T> consumer = StreamConsumerToList.create(items); listConsumers.add(consumer); return Promise.of(consumer); } @Override public Promise<Long> createId() { return Promise.of(++chunkId); } @Override public Promise<Void> finish(Set<Long> chunkIds) { return Promise.complete(); } }; List<AggregationChunk> chunksToConsolidate = state.findChunksGroupWithMostOverlaps(); List<String> fields = new ArrayList<>(); for (AggregationChunk chunk : chunksToConsolidate) { for (String field : chunk.getMeasures()) { if (!fields.contains(field)) { fields.add(field); } } } Class<?> recordClass = createRecordClass(structure, structure.getKeys(), fields, classLoader); ExpectedException exception = new ExpectedException("Test Exception"); StreamSupplier<KeyValuePair> supplier = StreamSupplier.concat( StreamSupplier.of( new KeyValuePair(1, 1, 0), new KeyValuePair(1, 2, 1), new KeyValuePair(1, 1, 2)), StreamSupplier.of( new KeyValuePair(1, 1, 0), new KeyValuePair(1, 2, 1), new KeyValuePair(1, 1, 2)), StreamSupplier.of( new KeyValuePair(1, 1, 0), new KeyValuePair(1, 2, 1), new KeyValuePair(1, 1, 2)), StreamSupplier.of( new KeyValuePair(1, 1, 0), new KeyValuePair(1, 2, 1), new KeyValuePair(1, 1, 2)), StreamSupplier.closingWithError(exception) ); AggregationChunker chunker = AggregationChunker.create( structure, structure.getMeasures(), recordClass, (PartitionPredicate) singlePartition(), aggregationChunkStorage, classLoader, 1); Throwable e = awaitException(supplier.streamTo(chunker)); assertSame(exception, e); for (int i = 0; i < listConsumers.size() - 1; i++) { assertEndOfStream(listConsumers.get(i)); } assertClosedWithError(listConsumers.get(listConsumers.size() - 1)); }
@Test public void testStorageConsumerWithError() { List<StreamConsumer> listConsumers = new ArrayList<>(); AggregationChunkStorage<Long> aggregationChunkStorage = new AggregationChunkStorage<Long>() { long chunkId; final List items = new ArrayList(); @Override public <T> Promise<StreamSupplier<T>> read(AggregationStructure aggregation, List<String> fields, Class<T> recordClass, Long chunkId, DefiningClassLoader classLoader) { return Promise.of(StreamSupplier.ofIterable(items)); } @Override public <T> Promise<StreamConsumer<T>> write(AggregationStructure aggregation, List<String> fields, Class<T> recordClass, Long chunkId, DefiningClassLoader classLoader) { if (chunkId == 1) { StreamConsumerToList<T> toList = StreamConsumerToList.create(items); listConsumers.add(toList); return Promise.of(toList); } else { StreamConsumer<T> consumer = StreamConsumer.closingWithError(new Exception("Test Exception")); listConsumers.add(consumer); return Promise.of(consumer); } } @Override public Promise<Long> createId() { return Promise.of(++chunkId); } @Override public Promise<Void> finish(Set<Long> chunkIds) { return Promise.complete(); } }; List<AggregationChunk> chunksToConsolidate = state.findChunksGroupWithMostOverlaps(); List<String> fields = new ArrayList<>(); for (AggregationChunk chunk : chunksToConsolidate) { for (String field : chunk.getMeasures()) { if (!fields.contains(field)) { fields.add(field); } } } Class<?> recordClass = createRecordClass(structure, structure.getKeys(), fields, classLoader); StreamSupplier<KeyValuePair> supplier = StreamSupplier.of( new KeyValuePair(1, 1, 0), new KeyValuePair(1, 2, 1), new KeyValuePair(1, 1, 2), new KeyValuePair(1, 1, 2), new KeyValuePair(1, 1, 2)); AggregationChunker chunker = AggregationChunker.create( structure, structure.getMeasures(), recordClass, (PartitionPredicate) singlePartition(), aggregationChunkStorage, classLoader, 1); awaitException(supplier.streamTo(chunker)); for (int i = 0; i < listConsumers.size() - 1; i++) { assertEndOfStream(listConsumers.get(i)); } assertClosedWithError(listConsumers.get(listConsumers.size() - 1)); }