@Override
  public boolean clearStream(StreamSpec streamSpec) {
    LOG.info("Creating Kafka topic: {} on system: {}", streamSpec.getPhysicalName(), streamSpec.getSystemName());

    String topicName = streamSpec.getPhysicalName();

    try {
      DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(ImmutableSet.of(topicName));
      deleteTopicsResult.all().get(KAFKA_ADMIN_OPS_TIMEOUT_MS, TimeUnit.MILLISECONDS);
    } catch (Exception e) {
      LOG.error("Failed to delete topic {} with exception {}.", topicName, e);
      return false;
    }

    return true;
  }
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