@Override
  public Map<String, String> toConfig() {
    Map<String, String> config = new HashMap<>(super.toConfig());

    String systemName = getSystemName();
    String streamId = getStreamId();
    String clientConfigPrefix =
        String.format(KinesisConfig.CONFIG_STREAM_KINESIS_CLIENT_LIB_CONFIG, systemName, streamId);

    region.ifPresent(val -> config.put(String.format(KinesisConfig.CONFIG_STREAM_REGION, systemName, streamId), val));
    accessKey.ifPresent(
      val -> config.put(String.format(KinesisConfig.CONFIG_STREAM_ACCESS_KEY, systemName, streamId), val));
    secretKey.ifPresent(
      val -> config.put(String.format(KinesisConfig.CONFIG_STREAM_SECRET_KEY, systemName, streamId), val));
    kclConfig.forEach((k, v) -> config.put(clientConfigPrefix + k, v));

    return config;
  }
KCL (Kinesis Client Library) config for the system stream. This is not required by default.
@param kclConfig A map of specified KCL configs
@return this input descriptor
samza-aws/src/main/java/org/apache/samza/system/kinesis/descriptors/KinesisInputDescriptor.java