this.recordSchema = recordSchema;
this.metricStore = metricStore;
}
@Override
public void onReceived(Iterator<FetchedMessage> messages) {
<BUG>final ByteBufferInputStream is = new ByteBufferInputStream(null);
List<MetricValues> records = ImmutableList.copyOf(
Iterators.filter(Iterators.transform(messages, new Function<FetchedMessage, MetricValues>() {
@Override
public MetricValues apply(FetchedMessage input) {</BUG>
try {

this.recordSchema = recordSchema;
this.metricStore = metricStore;
}
@Override
public void onReceived(Iterator<FetchedMessage> messages) {
final ByteBufferInputStream is = new ByteBufferInputStream(null);
List<MetricValues> records = Lists.newArrayList();
while (messages.hasNext()) {
FetchedMessage input = messages.next();
