Kinesis Producer - Support separate Cloudwatch credentials

[Diem, Chase via user]

Hey Team,

Looking for some thoughts here:

  *   We have a Kinesis Producer that produces to a topic in another AWS account
  *   The producer allows for configurations to set credentials for that 
account: 
https://github.com/apache/flink-connector-aws/blob/main/flink-connector-aws/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/util/KinesisConfigUtil.java#L494
  *   We DO NOT have access produce to their Cloudwatch.  We would prefer to 
produce the metrics to our own account instead.
  *   The AWS KinesisProducerConfiguration supports setting separate 
credentials for both the producer and Cloudwatch, but the KinesisConfigUtil 
does not support it: 
https://github.com/awslabs/amazon-kinesis-producer/blob/master/java/amazon-kinesis-producer/src/main/java/com/amazonaws/services/kinesis/producer/KinesisProducerConfiguration.java#L139

It would be nice to support properties for say:
     *     metrics.aws.credentials.provider for custom Cloudwatch metrics 
credentials provider,
     *     metrics.aws.credentials.provider.role.provider for custom Cloudwatch 
credentials provider for assuming a role,


AWS KinesisProducerConfiguration.java:
    /**
     * {@link AWSCredentialsProvider} that supplies credentials used to upload
     * metrics to CloudWatch.
     * <p>
     * If not given, the credentials used to put records
     * to Kinesis are also used for CloudWatch.
     *
     * @see #setCredentialsProvider(AWSCredentialsProvider)
     */
    public KinesisProducerConfiguration 
setMetricsCredentialsProvider(AWSCredentialsProvider 
metricsCredentialsProvider) {
        this.metricsCredentialsProvider = metricsCredentialsProvider;
        return this;
    }

Flink KinesisConfigUtil.java:
        KinesisProducerConfiguration kpc = 
KinesisProducerConfiguration.fromProperties(config);
        kpc.setRegion(config.getProperty(AWSConfigConstants.AWS_REGION));

        kpc.setCredentialsProvider(AWSUtil.getCredentialsProvider(config));

        // we explicitly lower the credential refresh delay (default is 5 
seconds)
        // to avoid an ignorable interruption warning that occurs when shutting 
down the
        // KPL client. See 
https://github.com/awslabs/amazon-kinesis-producer/issues/10.
        kpc.setCredentialsRefreshDelay(100);