offsetsToCommit property

Iterable<ConsumerOffset> get offsetsToCommit

Implementation

Iterable<ConsumerOffset> get offsetsToCommit {
  var grouped = Map<TopicPartition, int>();
  for (var envelope in items) {
    var key = TopicPartition(envelope.topicName, envelope.partitionId);
    if (!grouped.containsKey(key)) {
      grouped[key] = envelope.offset;
    } else if (grouped[key]! < envelope.offset) {
      grouped[key] = envelope.offset;
    }
  }

  List<ConsumerOffset> offsets = [];
  for (var key in grouped.keys) {
    offsets.add(ConsumerOffset(key.topicName, key.partitionId, grouped[key]!, commitMetadata));
  }

  return offsets;
}