@Override
  public Set<SystemStream> getAllSystemStreams() {
    Map<String, List<PartitionInfo>> topicToPartitionInfoMap = threadSafeKafkaConsumer.execute(consumer -> consumer.listTopics());
    Set<SystemStream> systemStreams = topicToPartitionInfoMap.keySet()
                                                             .stream()
                                                             .map(topic -> new SystemStream(systemName, topic))
                                                             .collect(Collectors.toSet());
    return systemStreams;
  }
Delete records up to (and including) the provided ssp offsets for
all system stream partitions specified in the map.
This only works with Kafka cluster 0.11 or later. Otherwise it's a no-op.
@param offsets specifies up to what offsets the messages should be deleted
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemAdmin.java