You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "gguptp (via GitHub)" <gi...@apache.org> on 2025/06/04 19:24:18 UTC
Re: [PR] [FLINK-37688] Implement Amazon CloudWatch Metric Sink Connector [flink-connector-aws]
gguptp commented on code in PR #202:
URL: https://github.com/apache/flink-connector-aws/pull/202#discussion_r2127234968
##########
flink-connector-aws/flink-connector-cloudwatch/src/main/java/org/apache/flink/connector/cloudwatch/sink/CloudWatchStateSerializer.java:
##########
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.cloudwatch.sink;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.base.sink.writer.AsyncSinkWriterStateSerializer;
+
+import software.amazon.awssdk.utils.StringUtils;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+/** CloudWatch implementation {@link AsyncSinkWriterStateSerializer}. */
+@Internal
+public class CloudWatchStateSerializer extends AsyncSinkWriterStateSerializer<MetricWriteRequest> {
+ @Override
Review Comment:
we should keep a version of the state serializer so that in case we have future edits to serialization logic, we can make it forward and backward compatible
##########
flink-connector-aws/flink-connector-cloudwatch/src/main/java/org/apache/flink/connector/cloudwatch/sink/MetricWriteRequest.java:
##########
@@ -0,0 +1,335 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.cloudwatch.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+import org.apache.commons.lang3.builder.EqualsBuilder;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+import software.amazon.awssdk.annotations.NotNull;
+import software.amazon.awssdk.services.cloudwatch.model.MetricDatum;
+import software.amazon.awssdk.services.cloudwatch.model.StatisticSet;
+
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/** Pojo used as sink input, containing information for a single CloudWatch MetricDatum Object. */
+@PublicEvolving
+public class MetricWriteRequest implements Serializable {
+
+ public String metricName;
+ public Dimension[] dimensions;
+ public Double[] values;
+ public Double[] counts;
+ public Instant timestamp;
+ public String unit;
+ public Integer storageResolution;
+ public Double statisticMax;
+ public Double statisticMin;
+ public Double statisticSum;
+ public Double statisticCount;
+
+ public MetricWriteRequest() {}
+
+ public MetricWriteRequest(
+ @NotNull String metricName,
+ Dimension[] dimensions,
+ Double[] values,
+ Double[] counts,
+ Instant timestamp,
+ String unit,
+ Integer storageResolution,
+ Double statisticMax,
+ Double statisticMin,
+ Double statisticSum,
+ Double statisticCount) {
+ this.metricName = metricName;
+ this.dimensions = dimensions;
+ this.values = values;
+ this.counts = counts;
+ this.timestamp = timestamp;
+ this.unit = unit;
+ this.storageResolution = storageResolution;
+ this.statisticMax = statisticMax;
+ this.statisticMin = statisticMin;
+ this.statisticSum = statisticSum;
+ this.statisticCount = statisticCount;
+ }
+
+ public Dimension[] getDimensions() {
+ return dimensions;
+ }
+
+ public Double[] getValues() {
+ return values;
+ }
+
+ public String getMetricName() {
+ return metricName;
+ }
+
+ public Double[] getCounts() {
+ return counts;
+ }
+
+ public Instant getTimestamp() {
+ return timestamp;
+ }
+
+ public String getUnit() {
+ return unit;
+ }
+
+ public Integer getStorageResolution() {
+ return storageResolution;
+ }
+
+ public Double getStatisticMax() {
+ return statisticMax;
+ }
+
+ public Double getStatisticMin() {
+ return statisticMin;
+ }
+
+ public Double getStatisticSum() {
+ return statisticSum;
+ }
+
+ public Double getStatisticCount() {
+ return statisticCount;
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ /** A single Dimension. */
+ public static class Dimension implements Serializable {
+ public String name;
+ public String value;
+
+ public Dimension() {}
+
+ public Dimension(String name, String value) {
+ this.name = name;
+ this.value = value;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public String getValue() {
+ return value;
+ }
+
+ public software.amazon.awssdk.services.cloudwatch.model.Dimension toCloudWatchDimension() {
+ return software.amazon.awssdk.services.cloudwatch.model.Dimension.builder()
+ .name(this.name)
+ .value(this.value)
+ .build();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Dimension label = (Dimension) o;
+ return new EqualsBuilder()
+ .append(name, label.name)
+ .append(value, label.value)
+ .isEquals();
+ }
+
+ @Override
+ public int hashCode() {
+ return new HashCodeBuilder(17, 37).append(name).append(value).toHashCode();
+ }
+ }
+
+ /** Builder for sink input pojo instance. */
+ public static final class Builder {
+ private final List<Dimension> dimensions = new ArrayList<>();
+ private final List<Double> values = new ArrayList<>();
+ private final List<Double> counts = new ArrayList<>();
+ private String metricName;
+ private Instant timestamp;
+ private String unit;
+ private Integer storageResolution;
+ private Double statisticMax;
+ private Double statisticMin;
+ private Double statisticSum;
+ private Double statisticCount;
+
+ private Builder() {}
+
+ public Builder withMetricName(String metricName) {
+ this.metricName = metricName;
+ return this;
+ }
+
+ public Builder addDimension(String dimensionName, String dimensionValue) {
+ dimensions.add(new Dimension(dimensionName, dimensionValue));
+ return this;
+ }
+
+ public Builder addValue(Double value) {
+ values.add(value);
+ return this;
+ }
+
+ public Builder addCount(Double count) {
+ counts.add(count);
+ return this;
+ }
+
+ public Builder withTimestamp(Instant timestamp) {
+ this.timestamp = timestamp;
+ return this;
+ }
+
+ public Builder withUnit(String unit) {
+ this.unit = unit;
+ return this;
+ }
+
+ public Builder withStorageResolution(Integer storageResolution) {
+ this.storageResolution = storageResolution;
+ return this;
+ }
+
+ public Builder withStatisticMax(Double statisticMax) {
+ this.statisticMax = statisticMax;
+ return this;
+ }
+
+ public Builder withStatisticMin(Double statisticMin) {
+ this.statisticMin = statisticMin;
+ return this;
+ }
+
+ public Builder withStatisticSum(Double statisticSum) {
+ this.statisticSum = statisticSum;
+ return this;
+ }
+
+ public Builder withStatisticCount(Double statisticCount) {
+ this.statisticCount = statisticCount;
+ return this;
+ }
+
+ public MetricWriteRequest build() {
+ return new MetricWriteRequest(
+ metricName,
+ dimensions.toArray(new Dimension[dimensions.size()]),
Review Comment:
could be replaced by `dimensions.toArray(new Dimension[0])`
##########
flink-connector-aws/flink-connector-cloudwatch/src/main/java/org/apache/flink/connector/cloudwatch/sink/client/CloudWatchAsyncClientProvider.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.cloudwatch.sink.client;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.aws.util.AWSClientUtil;
+import org.apache.flink.connector.aws.util.AWSGeneralUtil;
+import org.apache.flink.connector.cloudwatch.sink.CloudWatchConfigConstants;
+
+import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
+import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient;
+
+import java.util.Properties;
+
+/** Provides a {@link CloudWatchAsyncClient}. */
+@Internal
+public class CloudWatchAsyncClientProvider implements SdkClientProvider<CloudWatchAsyncClient> {
+
+ private final SdkAsyncHttpClient httpClient;
+ private final CloudWatchAsyncClient cloudWatchAsyncClient;
Review Comment:
do these need to be class level variables? could we just expose the buildClient method instead. WDYT?
##########
flink-connector-aws/flink-connector-cloudwatch/src/main/java/org/apache/flink/connector/cloudwatch/sink/CloudWatchSinkBuilder.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.cloudwatch.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.connector.base.sink.AsyncSinkBaseBuilder;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+
+import software.amazon.awssdk.http.Protocol;
+
+import java.util.Optional;
+import java.util.Properties;
+
+import static org.apache.flink.connector.aws.config.AWSConfigConstants.HTTP_PROTOCOL_VERSION;
+import static software.amazon.awssdk.http.Protocol.HTTP1_1;
+
+/**
+ * Builder to construct {@link CloudWatchSink}.
+ *
+ * <p>The following example shows the minimum setup to create a {@link CloudWatchSink} that writes
+ * String values to a CloudWatch named cloudWatchUrl.
+ *
+ * <pre>{@code
+ * Properties sinkProperties = new Properties();
+ * sinkProperties.put(AWSConfigConstants.AWS_REGION, "eu-west-1");
+ *
+ * CloudWatchSink<String> cloudWatchSink =
+ * CloudWatchSink.<String>builder()
+ * .setNamespace("namespace")
+ * .setCloudWatchClientProperties(sinkProperties)
+ * .build();
+ * }</pre>
+ *
+ * <p>If the following parameters are not set in this builder, the following defaults will be used:
+ *
+ * <ul>
+ * <li>{@code maxBatchSize} will be 100
+ * <li>{@code maxInFlightRequests} will be 50
+ * <li>{@code maxBufferedRequests} will be 5000
+ * <li>{@code maxBatchSizeInBytes} will be 100 KB
+ * <li>{@code maxTimeInBufferMs} will be 5000ms
+ * <li>{@code maxRecordSizeInBytes} will be 1 KB
+ * <li>{@code invalidMetricDataRetryMode} will be FAIL_ON_ERROR
+ * </ul>
+ *
+ * @param <InputT> type of elements that should be persisted in the destination
+ */
+@PublicEvolving
+public class CloudWatchSinkBuilder<InputT>
+ extends AsyncSinkBaseBuilder<InputT, MetricWriteRequest, CloudWatchSinkBuilder<InputT>> {
+
+ private static final int DEFAULT_MAX_BATCH_SIZE = 100;
+ private static final int DEFAULT_MAX_IN_FLIGHT_REQUESTS = 50;
+ private static final int DEFAULT_MAX_BUFFERED_REQUESTS = 5_000;
+ private static final long DEFAULT_MAX_BATCH_SIZE_IN_B = 100 * 1000;
+ private static final long DEFAULT_MAX_TIME_IN_BUFFER_MS = 5000;
+ private static final long DEFAULT_MAX_RECORD_SIZE_IN_B = 1000;
+ private static final InvalidMetricDataRetryMode DEFAULT_INVALID_METRIC_DATA_RETRY_MODE =
+ InvalidMetricDataRetryMode.FAIL_ON_ERROR;
+ private static final Protocol DEFAULT_HTTP_PROTOCOL = HTTP1_1;
Review Comment:
should we make these customer configurable? if yes, lets pull them from CloudWatchConfigConstants
##########
flink-connector-aws/flink-connector-cloudwatch/src/main/java/org/apache/flink/connector/cloudwatch/sink/MetricWriteRequest.java:
##########
@@ -0,0 +1,335 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.cloudwatch.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+import org.apache.commons.lang3.builder.EqualsBuilder;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+import software.amazon.awssdk.annotations.NotNull;
+import software.amazon.awssdk.services.cloudwatch.model.MetricDatum;
+import software.amazon.awssdk.services.cloudwatch.model.StatisticSet;
+
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/** Pojo used as sink input, containing information for a single CloudWatch MetricDatum Object. */
+@PublicEvolving
+public class MetricWriteRequest implements Serializable {
+
+ public String metricName;
+ public Dimension[] dimensions;
+ public Double[] values;
+ public Double[] counts;
+ public Instant timestamp;
+ public String unit;
+ public Integer storageResolution;
+ public Double statisticMax;
+ public Double statisticMin;
+ public Double statisticSum;
+ public Double statisticCount;
+
+ public MetricWriteRequest() {}
+
+ public MetricWriteRequest(
+ @NotNull String metricName,
+ Dimension[] dimensions,
+ Double[] values,
+ Double[] counts,
+ Instant timestamp,
+ String unit,
+ Integer storageResolution,
+ Double statisticMax,
+ Double statisticMin,
+ Double statisticSum,
+ Double statisticCount) {
+ this.metricName = metricName;
+ this.dimensions = dimensions;
+ this.values = values;
+ this.counts = counts;
+ this.timestamp = timestamp;
+ this.unit = unit;
+ this.storageResolution = storageResolution;
+ this.statisticMax = statisticMax;
+ this.statisticMin = statisticMin;
+ this.statisticSum = statisticSum;
+ this.statisticCount = statisticCount;
+ }
+
+ public Dimension[] getDimensions() {
+ return dimensions;
+ }
+
+ public Double[] getValues() {
+ return values;
+ }
+
+ public String getMetricName() {
+ return metricName;
+ }
+
+ public Double[] getCounts() {
+ return counts;
+ }
+
+ public Instant getTimestamp() {
+ return timestamp;
+ }
+
+ public String getUnit() {
+ return unit;
+ }
+
+ public Integer getStorageResolution() {
+ return storageResolution;
+ }
+
+ public Double getStatisticMax() {
+ return statisticMax;
+ }
+
+ public Double getStatisticMin() {
+ return statisticMin;
+ }
+
+ public Double getStatisticSum() {
+ return statisticSum;
+ }
+
+ public Double getStatisticCount() {
+ return statisticCount;
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ /** A single Dimension. */
+ public static class Dimension implements Serializable {
+ public String name;
Review Comment:
should we put a `toString()` here as well for debugging purposes
##########
flink-connector-aws/flink-connector-cloudwatch/src/main/java/org/apache/flink/connector/cloudwatch/sink/MetricWriteRequest.java:
##########
@@ -0,0 +1,335 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.cloudwatch.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+import org.apache.commons.lang3.builder.EqualsBuilder;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+import software.amazon.awssdk.annotations.NotNull;
+import software.amazon.awssdk.services.cloudwatch.model.MetricDatum;
+import software.amazon.awssdk.services.cloudwatch.model.StatisticSet;
+
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/** Pojo used as sink input, containing information for a single CloudWatch MetricDatum Object. */
+@PublicEvolving
+public class MetricWriteRequest implements Serializable {
+
+ public String metricName;
+ public Dimension[] dimensions;
+ public Double[] values;
+ public Double[] counts;
+ public Instant timestamp;
+ public String unit;
+ public Integer storageResolution;
+ public Double statisticMax;
+ public Double statisticMin;
+ public Double statisticSum;
+ public Double statisticCount;
+
+ public MetricWriteRequest() {}
+
+ public MetricWriteRequest(
+ @NotNull String metricName,
Review Comment:
i dont think we use the `@NotNull` annotation anywhere in the project, rather we use `Preconditions.checkNotNull`, so lets follow the same convention here as well
##########
flink-connector-aws-e2e-tests/flink-connector-cloudwatch-e2e-tests/pom.xml:
##########
@@ -0,0 +1,141 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one
+ ~ or more contributor license agreements. See the NOTICE file
+ ~ distributed with this work for additional information
+ ~ regarding copyright ownership. The ASF licenses this file
+ ~ to you under the Apache License, Version 2.0 (the
+ ~ "License"); you may not use this file except in compliance
+ ~ with the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <parent>
+ <artifactId>flink-connector-aws-e2e-tests-parent</artifactId>
+ <groupId>org.apache.flink</groupId>
+ <version>5.1-SNAPSHOT</version>
+ </parent>
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>flink-connector-aws-cloudwatch-e2e-tests</artifactId>
+ <name>Flink : Connectors : AWS : E2E Tests : Amazon CloudWatch</name>
+ <packaging>jar</packaging>
+
+ <properties>
+ <aws.sdkv2.version>2.31.18</aws.sdkv2.version>
Review Comment:
should we update this in the parent pom.xml. Having two different SDK versions doesnt make sense
##########
flink-connector-aws/flink-connector-cloudwatch/src/main/java/org/apache/flink/connector/cloudwatch/sink/client/SdkClientProvider.java:
##########
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.cloudwatch.sink.client;
+
+import org.apache.flink.annotation.Internal;
+
+import software.amazon.awssdk.core.SdkClient;
+import software.amazon.awssdk.utils.SdkAutoCloseable;
+
+/** Provides a {@link SdkClient}. */
+@Internal
+public interface SdkClientProvider<T extends SdkClient> extends SdkAutoCloseable {
Review Comment:
do we really need this interface?
##########
flink-connector-aws/flink-sql-connector-cloudwatch/pom.xml:
##########
@@ -0,0 +1,142 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-aws-parent</artifactId>
+ <version>5.1-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>flink-sql-connector-cloudwatch</artifactId>
+ <name>Flink : Connectors : AWS : SQL : Amazon CloudWatch</name>
+
+ <packaging>jar</packaging>
+
+ <properties>
+ <japicmp.skip>true</japicmp.skip>
+ <aws.sdkv2.version>2.31.18</aws.sdkv2.version>
Review Comment:
lets update this in the bom of the parent pom
##########
flink-connector-aws/flink-connector-cloudwatch/src/main/java/org/apache/flink/connector/cloudwatch/sink/DefaultMetricWriteRequestElementConverter.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.cloudwatch.sink;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+
+import java.util.UnknownFormatConversionException;
+
+/**
+ * An implementation of the {@link ElementConverter} that uses the AWS CloudWatch SDK v2. The user
+ * needs to provide the {@code InputT} in the form of {@link MetricWriteRequest}.
+ *
+ * <p>The InputT needs to be structured and only supports {@link MetricWriteRequest}
+ */
+@Internal
+public class DefaultMetricWriteRequestElementConverter<InputT>
+ extends MetricWriteRequestElementConverter<InputT> {
+
+ @Override
+ public MetricWriteRequest apply(InputT element, SinkWriter.Context context) {
+ if (!(element instanceof MetricWriteRequest)) {
+ throw new UnknownFormatConversionException(
+ "DefaultMetricWriteRequestElementConverter only supports MetricWriteRequest element.");
+ }
+
+ return (MetricWriteRequest) element;
Review Comment:
should we do
```
if (element instanceof MetricWriteRequest) {
return (MetricWriteRequest) element;
}
throw new UnknownFormatConversionException(
"DefaultMetricWriteRequestElementConverter only supports MetricWriteRequest element.");
```
?
##########
flink-connector-aws/flink-connector-cloudwatch/pom.xml:
##########
@@ -0,0 +1,176 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-aws-parent</artifactId>
+ <version>5.1-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>flink-connector-cloudwatch</artifactId>
+ <name>Flink : Connectors : AWS : Amazon Cloudwatch</name>
+ <packaging>jar</packaging>
+
+ <properties>
+ <aws.sdkv2.version>2.31.18</aws.sdkv2.version>
+ <flink.version>1.19.0</flink.version>
Review Comment:
lets not hard code flink version
##########
flink-connector-aws/flink-connector-cloudwatch/src/main/java/org/apache/flink/connector/cloudwatch/sink/MetricWriteRequest.java:
##########
@@ -0,0 +1,335 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.cloudwatch.sink;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+import org.apache.commons.lang3.builder.EqualsBuilder;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+import software.amazon.awssdk.annotations.NotNull;
+import software.amazon.awssdk.services.cloudwatch.model.MetricDatum;
+import software.amazon.awssdk.services.cloudwatch.model.StatisticSet;
+
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/** Pojo used as sink input, containing information for a single CloudWatch MetricDatum Object. */
+@PublicEvolving
+public class MetricWriteRequest implements Serializable {
+
+ public String metricName;
+ public Dimension[] dimensions;
+ public Double[] values;
+ public Double[] counts;
+ public Instant timestamp;
+ public String unit;
+ public Integer storageResolution;
+ public Double statisticMax;
+ public Double statisticMin;
+ public Double statisticSum;
+ public Double statisticCount;
+
+ public MetricWriteRequest() {}
+
+ public MetricWriteRequest(
+ @NotNull String metricName,
Review Comment:
also, i'm thinking should we have some additional validations for e.g maybe unit cannot be null or one of `statisticMax`, `statisticMin`, `statisticSum` or `statisticCount` cannot be null? just thinking out loud
##########
flink-connector-aws/flink-connector-cloudwatch/src/main/java/org/apache/flink/connector/cloudwatch/sink/CloudWatchSink.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.cloudwatch.sink;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.connector.base.sink.AsyncSinkBase;
+import org.apache.flink.connector.base.sink.writer.BufferedRequestState;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+import org.apache.flink.connector.cloudwatch.sink.client.CloudWatchAsyncClientProvider;
+import org.apache.flink.connector.cloudwatch.sink.client.SdkClientProvider;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.util.Preconditions;
+
+import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Properties;
+
+/**
+ * A CloudWatch Sink that performs async requests against a destination CloudWatch using the
+ * buffering protocol specified in {@link AsyncSinkBase}.
+ *
+ * <p>The sink internally uses a {@link CloudWatchAsyncClient} to communicate with the AWS endpoint.
+ *
+ * <p>Please see the writer implementation in {@link CloudWatchSinkWriter}
+ *
+ * <p>maxBatchSize is calculated in terms of requestEntries (MetricWriteRequest). In CloudWatch,
+ * each PutMetricDataRequest can have maximum of 1000 MetricWriteRequest, hence the maxBatchSize
+ * cannot be more than 1000.
+ *
+ * <p>maxBatchSizeInBytes is calculated in terms of size of requestEntries (MetricWriteRequest). In
+ * CloudWatch, each PutMetricDataRequest can have maximum of 1MB of payload, hence the
+ * maxBatchSizeInBytes cannot be more than 1 MB.
+ *
+ * @param <InputT> Type of the elements handled by this sink
+ */
+@PublicEvolving
+public class CloudWatchSink<InputT> extends AsyncSinkBase<InputT, MetricWriteRequest> {
+
+ private final String namespace;
+ private final Properties cloudWatchClientProperties;
+ private transient SdkClientProvider<CloudWatchAsyncClient> asyncClientSdkClientProviderOverride;
+ private final InvalidMetricDataRetryMode invalidMetricDataRetryMode;
+
+ CloudWatchSink(
+ ElementConverter<InputT, MetricWriteRequest> elementConverter,
+ Integer maxBatchSize,
+ Integer maxInFlightRequests,
+ Integer maxBufferedRequests,
+ Long maxBatchSizeInBytes,
+ Long maxTimeInBufferMS,
+ Long maxRecordSizeInBytes,
+ String namespace,
+ Properties cloudWatchClientProperties,
+ InvalidMetricDataRetryMode invalidMetricDataRetryMode) {
+ super(
+ elementConverter,
+ maxBatchSize,
+ maxInFlightRequests,
+ maxBufferedRequests,
+ maxBatchSizeInBytes,
+ maxTimeInBufferMS,
+ maxRecordSizeInBytes);
+ this.namespace =
+ Preconditions.checkNotNull(
+ namespace,
+ "The cloudWatch namespace must not be null when initializing the CloudWatch Sink.");
+ this.invalidMetricDataRetryMode = invalidMetricDataRetryMode;
+ Preconditions.checkArgument(
+ !this.namespace.isEmpty(),
+ "The cloudWatch namespace must be set when initializing the CloudWatch Sink.");
+
+ Preconditions.checkArgument(
+ (this.getMaxBatchSize() <= 1000),
+ "The cloudWatch MaxBatchSize must not be greater than 1,000.");
+
+ Preconditions.checkArgument(
+ (this.getMaxBatchSizeInBytes() <= 1000 * 1000),
+ "The cloudWatch MaxBatchSizeInBytes must not be greater than 1,000,000.");
Review Comment:
instead of failing, should we just cap these numbers to the max ones if customer provides a bigger value?
##########
flink-connector-aws/flink-connector-cloudwatch/src/main/java/org/apache/flink/connector/cloudwatch/sink/CloudWatchStateSerializer.java:
##########
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.cloudwatch.sink;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.base.sink.writer.AsyncSinkWriterStateSerializer;
+
+import software.amazon.awssdk.utils.StringUtils;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+/** CloudWatch implementation {@link AsyncSinkWriterStateSerializer}. */
+@Internal
+public class CloudWatchStateSerializer extends AsyncSinkWriterStateSerializer<MetricWriteRequest> {
+ @Override
Review Comment:
i mean we have a version, but we should serialize that version in the state to ensure backward compatibility
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org