public Set<CoordinatorStreamMessage> getUnreadMessages(SystemStreamPartitionIterator iterator, String type) {
    LinkedHashSet<CoordinatorStreamMessage> messages = new LinkedHashSet<CoordinatorStreamMessage>();
    while (iterator.hasNext()) {
      IncomingMessageEnvelope envelope = iterator.next();
      Object[] keyArray = keySerde.fromBytes((byte[]) envelope.getKey()).toArray();
      Map<String, Object> valueMap = null;
      if (envelope.getMessage() != null) {
        valueMap = messageSerde.fromBytes((byte[]) envelope.getMessage());
      }
      CoordinatorStreamMessage coordinatorStreamMessage = new CoordinatorStreamMessage(keyArray, valueMap);
      if (type == null || type.equals(coordinatorStreamMessage.getType())) {
        messages.add(coordinatorStreamMessage);
      }
    }
    return messages;
  }
returns all unread messages of a specific type, after an iterator on the stream

@param iterator the iterator pointing to an offset in the coordinator stream. All unread messages after this iterator are returned
@param type     the type of the messages to be returned
@return a set of unread messages of a given type, after a given iterator
samza-core/src/main/java/org/apache/samza/coordinator/stream/CoordinatorStreamSystemConsumer.java