[FLINK-37688] Implement Amazon CloudWatch Metric Sink Connector

[darenwkt]

Purpose of the change

Implement Amazon CloudWatch Metric Sink Connector based on FLIP-524 (https://cwiki.apache.org/confluence/display/FLINK/FLIP-524:+CloudWatch+Metric+Sink+Connector) which includes following:

• DataStream and TableAPI support
• Integration Test with LocalStack container

Verifying this change

This change added tests and can be verified as follows:

Unit Test

• Added unit tests

IT Test - Added integration tests for end-to-end deployment

• Ran IT test locally

mvn clean verify -Prun-end-to-end-tests -DdistDir=/Users/darenwkt/Downloads/flink-1.19.2

• Verified test pass for DataStream and TableAPI

[INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 71.197 s - in org.apache.flink.connector.cloudwatch.table.test.CloudWatchTableAPIITCase
[INFO] Running org.apache.flink.connector.cloudwatch.sink.test.CloudWatchSinkITCase
[INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 34.422 s - in org.apache.flink.connector.cloudwatch.sink.test.CloudWatchSinkITCase
[INFO] 
[INFO] Results:
[INFO] 
[INFO] Tests run: 2, Failures: 0, Errors: 0, Skipped: 0

DataStream API Manual Test - Manually verified by running the CloudWatch connector on a local Flink cluster.

• Create Flink job code

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        final Map<String, Properties> applicationParameters = loadApplicationProperties(env);

        env.fromSource(createDataGeneratorSource(applicationParameters.get("DataGen")), WatermarkStrategy.noWatermarks(), "DataGen")
                .uid("data-gen")
                .setParallelism(1)
                .disableChaining()
                .sinkTo(CloudWatchSink.<TemperatureSample>builder()
                        .setNamespace("CloudWatchSinkTest")
                        .setCloudWatchClientProperties(applicationParameters.get("CloudWatchSink"))
                        .setElementConverter(new MetricWriteRequestElementConverter<>() {
                            @Override
                            public MetricWriteRequest apply(TemperatureSample value, SinkWriter.Context context) {
                                return MetricWriteRequest.builder()
                                        .withMetricName("temperature")
                                        .addDimension("roomId", "testRoomId")
                                        .addDimension("sensorId", "testSensorId")
                                        .addValue(value.getTemperature())
                                        .addCount(1d)
                                        .withTimestamp(Instant.ofEpochMilli(value.getTimestamp()))
                                        .build();
                            }
                        })
                        .build())
                .uid("cloudwatch-sink").name("CloudWatchSink");

        env.enableCheckpointing(5000);

        env.execute("CloudWatch sink example");

• Build jar and submitted job

./bin/flink run cloudwatch-connector-example-1.0.jar
Job has been submitted with JobID f417eccc9d9204604ec51eeed24196e0

• Verified job is running and checkpointing successfully
  

• Verified CloudWatch received the metric successfully
  

TableAPI Manual Test - Manually verified by running the CloudWatch connector on a local Flink cluster.

• Built SQL jar and started sql-client with it

./bin/sql-client.sh --jar flink-sql-connector-cloudwatch-5.1-SNAPSHOT.jar

• Created Sink Table

Flink SQL>
> CREATE TABLE CloudWatchTable
> (
>     `my_metric_name`   STRING,
>     `my_dim` STRING,
>     `sample_value` DOUBLE,
>     `sample_count` DOUBLE,
>     `unit`             STRING,
>     `storage_res`      INT,
>     `stats_max` DOUBLE,
>     `stats_min` DOUBLE,
>     `stats_sum` DOUBLE,
>     `stats_count` DOUBLE
> )
>     WITH (
>         'connector' = 'cloudwatch',
>         'aws.region' = 'us-east-1',
>         'metric.namespace' = 'cw_connector_namespace',
>         'metric.name.key' = 'my_metric_name',
>         'metric.dimension.keys' = 'my_dim',
>         'metric.value.key' = 'sample_value',
>         'metric.count.key' = 'sample_count',
>         'metric.unit.key' = 'unit',
>         'metric.storage-resolution.key' = 'storage_res',
>         'metric.statistic.max.key' = 'stats_max',
>         'metric.statistic.min.key' = 'stats_min',
>         'metric.statistic.sum.key' = 'stats_sum',
>         'metric.statistic.sample-count.key' = 'stats_count',
>         'sink.invalid-metric.retry-mode' = 'RETRY'
>         );
[INFO] Execute statement succeed.

• Inserted sample value into table

Flink SQL> INSERT INTO CloudWatchTable VALUES ('test_metric', 'dim_1', 123, 1, 'Seconds', 60, 999, 1, 10, 1
1);
[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: 90d19a75e7326575b4ea3f4d883091a7

• Verified job entered FINISHED state successfully
  

• Verified Cloudwatch received the metric successfully
  

Significant changes

(Please check any boxes [x] if the answer is "yes". You can first publish the PR and check them afterwards, for convenience.)

• Dependencies have been added or upgraded
• Public API has been changed (Public API is any class annotated with @Public(Evolving))
• Serializers have been changed
• New feature has been introduced
  • If yes, how is this documented? (JavaDocs)

[gguptp]

Thanks for the PR @darenwkt! Left some comments

[gguptp]

do these need to be class level variables? could we just expose the buildClient method instead. WDYT?

[darenwkt]

I see what you mean, I think the benefit of using getClient is that the same cloudwatchClient instance can be reused many times in the same provider, so I am inclined to keep this for now, let me know what you think

[gguptp]

do we really need this interface?

[darenwkt]

We don't need it strictly, I just saw that we use this for DDB and SQS sink as well and thought of using it here to be consistent