public KinesisClientLibConfiguration getKinesisClientLibConfig(String system, String stream, String appName) {
    ClientConfiguration clientConfig = getAWSClientConfig(system);
    String workerId = appName + "-" + UUID.randomUUID();
    InitialPositionInStream startPos = InitialPositionInStream.LATEST;
    AWSCredentialsProvider provider = credentialsProviderForStream(system, stream);
    KinesisClientLibConfiguration kinesisClientLibConfiguration =
        new KinesisClientLibConfiguration(appName, stream, provider, workerId)
            .withRegionName(getRegion(system, stream).getName())
            .withKinesisClientConfig(clientConfig)
            .withCloudWatchClientConfig(clientConfig)
            .withDynamoDBClientConfig(clientConfig)
            .withInitialPositionInStream(startPos)
            .withCallProcessRecordsEvenForEmptyRecordList(true); // For health monitoring metrics.
    // First, get system scoped configs for KCL and override with configs set at stream scope.
    setKinesisClientLibConfigs(
        subset(String.format(CONFIG_SYSTEM_KINESIS_CLIENT_LIB_CONFIG, system)), kinesisClientLibConfiguration);
    setKinesisClientLibConfigs(subset(String.format(CONFIG_STREAM_KINESIS_CLIENT_LIB_CONFIG, system, stream)),
        kinesisClientLibConfiguration);
    return kinesisClientLibConfiguration;
  }
Get KCL config for a given system stream.
@param system name of the system
@param stream name of the stream
@param appName name of the application
@return Stream scoped KCL configs required to build
        {@link KinesisClientLibConfiguration}
samza-aws/src/main/java/org/apache/samza/system/kinesis/KinesisConfig.java