@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)); }
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; }
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; } }
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); }
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(); }
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();  }