protected Properties createAdminClientProperties() {
    // populate brokerList from either consumer or producer configs
    Properties props = new Properties();
    // included SSL settings if needed

    props.putAll(config.subset(String.format("systems.%s.consumer.", systemName), true));

    //validate brokerList
    String brokerList = config.get(
        String.format(KafkaConfig.CONSUMER_CONFIGS_CONFIG_KEY(), systemName, ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG));
    if (brokerList == null) {
      brokerList = config.get(String.format(KafkaConfig.PRODUCER_CONFIGS_CONFIG_KEY(), systemName,
          ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG));
    }
    if (brokerList == null) {
      throw new SamzaException(
          ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG + " is required for systemAdmin for system " + systemName);
    }
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);

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