@Override
  public Integer offsetComparator(String offset1, String offset2) {
    if (offset1 == null || offset2 == null) {
      return -1;
    }

    return Long.valueOf(offset1).compareTo(Long.valueOf(offset2));
  }
Fetch SystemStreamMetadata for each topic with the consumer
@param topics set of topics to get metadata info for
@return map of topic to SystemStreamMetadata
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemAdmin.java