private final ConcurrentMap<Long, GroupBatchBuilder> _batches =
new ConcurrentHashMap<>();
private final BatchSizeMetric _batchSizeMetric = new BatchSizeMetric();
private final BatchAggregationTimeMetric _batchAggregationTimeMetric = new BatchAggregationTimeMetric();
public Task<T> batchable(final String desc, final K key) {
<BUG>return Task.async(desc, ctx -> {
final BatchPromise<T> result = new BatchPromise<>();</BUG>
final Long planId = ctx.getPlanId();
final GroupBatchBuilder builder = _batches.computeIfAbsent(planId, k -> new GroupBatchBuilder());
final G group = classify(key);

private final ConcurrentMap<Long, GroupBatchBuilder> _batches =
new ConcurrentHashMap<>();
private final BatchSizeMetric _batchSizeMetric = new BatchSizeMetric();
private final BatchAggregationTimeMetric _batchAggregationTimeMetric = new BatchAggregationTimeMetric();
public Task<T> batchable(final String desc, final K key) {
Task<T> batchableTask = Task.async(desc, ctx -> {
final BatchPromise<T> result = new BatchPromise<>();
final Long planId = ctx.getPlanId();
final GroupBatchBuilder builder = _batches.computeIfAbsent(planId, k -> new GroupBatchBuilder());
final G group = classify(key);
