code: @SuppressWarnings("unchecked")
  	public <T, C, K extends Comparable> Promise<AggregationDiff> consume(StreamSupplier<T> supplier,
  			Class<T> inputClass, Map<String, String> keyFields, Map<String, String> measureFields) {
  		checkArgument(new HashSet<>(getKeys()).equals(keyFields.keySet()), "Expected keys: %s, actual keyFields: %s", getKeys(), keyFields);
  		checkArgument(getMeasureTypes().keySet().containsAll(measureFields.keySet()), "Unknown measures: %s", difference(measureFields.keySet(),
  				getMeasureTypes().keySet()));
  
  		logger.info("Started consuming data in aggregation {}. Keys: {} Measures: {}", this, keyFields.keySet(), measureFields.keySet());
  
  		Class<K> keyClass = createKeyClass(
  				keysToMap(getKeys().stream(), structure.getKeyTypes()::get),
  				classLoader);
  		Set<String> measureFieldKeys = measureFields.keySet();
  		List<String> measures = getMeasureTypes().keySet().stream().filter(measureFieldKeys::contains).collect(toList());
  
  		Class<T> recordClass = createRecordClass(structure, getKeys(), measures, classLoader);
  
  		Aggregate<T, Object> aggregate = createPreaggregator(structure, inputClass, recordClass,
  				keyFields, measureFields,
  				classLoader);
  
  		Function<T, K> keyFunction = createKeyFunction(inputClass, keyClass, getKeys(), classLoader);
  		AggregationGroupReducer<C, T, K> groupReducer = new AggregationGroupReducer<>(aggregationChunkStorage,
  				structure, measures,
  				recordClass,
  				createPartitionPredicate(recordClass, getPartitioningKey(), classLoader),
  				keyFunction,
  				aggregate, chunkSize, classLoader);
  
  		return supplier.streamTo(groupReducer)
  				.then($ -> groupReducer.getResult())
  				.map(chunks -> AggregationDiff.of(new HashSet<>(chunks)));
  	}
docs: Provides a {@link StreamConsumer} for streaming data to this aggregation.
  
  @param inputClass class of input records
  @param <T>        data records type
  @return consumer for streaming data to aggregation