@Override
  public Map<String, String> toConfig() {
    Map<String, String> configs = new HashMap<>(super.toConfig());
    if (!consumerZkConnect.isEmpty()) {
      configs.put(String.format(CONSUMER_ZK_CONNECT_CONFIG_KEY, getSystemName()), String.join(",", consumerZkConnect));
    }
    consumerAutoOffsetResetOptional.ifPresent(consumerAutoOffsetReset ->
        configs.put(String.format(CONSUMER_AUTO_OFFSET_RESET_CONFIG_KEY, getSystemName()), consumerAutoOffsetReset));
    consumerFetchThresholdOptional.ifPresent(consumerFetchThreshold ->
        configs.put(String.format(CONSUMER_FETCH_THRESHOLD_CONFIG_KEY, getSystemName()), Integer.toString(consumerFetchThreshold)));
    consumerFetchThresholdBytesOptional.ifPresent(consumerFetchThresholdBytes ->
        configs.put(String.format(CONSUMER_FETCH_THRESHOLD_BYTES_CONFIG_KEY, getSystemName()), Long.toString(consumerFetchThresholdBytes)));
    consumerFetchMessageMaxBytesOptional.ifPresent(consumerFetchMessageMaxBytes ->
        configs.put(String.format(CONSUMER_FETCH_MESSAGE_MAX_BYTES_KEY, getSystemName()), Long.toString(consumerFetchMessageMaxBytes)));
    consumerConfigs.forEach((key, value) -> configs.put(String.format(CONSUMER_CONFIGS_CONFIG_KEY, getSystemName(), key), value));
    if (!producerBootstrapServers.isEmpty()) {
      configs.put(String.format(PRODUCER_BOOTSTRAP_SERVERS_CONFIG_KEY, getSystemName()), String.join(",", producerBootstrapServers));
    }
    producerConfigs.forEach((key, value) -> configs.put(String.format(PRODUCER_CONFIGS_CONFIG_KEY, getSystemName(), key), value));
    return configs;
  }
Any Kafka producer configuration can be included here. For example, to change the request timeout,
you can set timeout.ms. (There is no need to configure client.id as it is automatically configured by Samza.)

@param producerConfigs additional producer configuration
@return this system descriptor
samza-kafka/src/main/java/org/apache/samza/system/kafka/descriptors/KafkaSystemDescriptor.java