feat: Max Compute Sink

build.gradle

@@ -22,13 +22,15 @@ plugins {
 }
 
 group 'com.gotocompany'
-version '0.9.2'
+version '0.10.1'
 
 repositories {
     mavenLocal()
     mavenCentral()
 }
 
+configurations.configureEach { exclude group: 'com.google.guava', module: 'listenablefuture' }
+
 dependencies {
     implementation group: 'com.google.protobuf', name: 'protobuf-java', version: '3.25.0'
     implementation group: 'com.datadoghq', name: 'java-dogstatsd-client', version: '2.13.0'
@@ -44,6 +46,10 @@ dependencies {
     implementation(group: 'com.google.cloud', name: 'google-cloud-bigtable', version: '2.24.1') {
         exclude group: "io.grpc"
     }
+    implementation (group: 'com.aliyun.odps', name: 'odps-sdk-core', version: '0.51.0-public.rc0') {
+        exclude group: "com.google"
+        exclude group: "io.grpc"
+    }
     implementation 'io.grpc:grpc-all:1.55.1'
     implementation group: 'org.slf4j', name: 'jul-to-slf4j', version: '1.7.35'
     implementation group: 'redis.clients', name: 'jedis', version: '3.10.0'
@@ -53,7 +59,9 @@ dependencies {
     implementation group: 'com.jayway.jsonpath', name: 'json-path', version: '2.8.0'
     implementation group: 'com.squareup.okhttp3', name: 'okhttp', version: '4.9.1'
     implementation group: 'joda-time', name: 'joda-time', version: '2.10.2'
+    implementation('com.google.guava:guava:32.0.1-jre') { force = true }
     testImplementation group: 'junit', name: 'junit', version: '4.13.1'
+    testImplementation group: 'org.assertj', name: 'assertj-core', version: '3.26.3'
     testImplementation 'org.hamcrest:hamcrest-all:1.3'
     testImplementation 'org.mockito:mockito-core:4.5.1'
     testImplementation 'com.github.tomakehurst:wiremock:2.16.0'
@@ -201,7 +209,10 @@ jacocoTestCoverageVerification {
                               '**/serializer/**',
                               '**/cortexpb/**',
                               '**/Clock**',
-                              '**/GoGoProtos**',])
+                              '**/GoGoProtos**',
+                              '**/MaxComputeClient**',
+                              '**/MaxComputeSinkFactory**',
+                              ''])
         })
     }
     violationRules {

docs/reference/configuration/maxcompute.md

@@ -0,0 +1,135 @@
+# MaxCompute Sink
+
+A MaxCompute sink requires these configurations to be passed on alongside with generic ones
+
+## SINK_MAXCOMPUTE_ODPS_URL
+
+Contains the URL of the MaxCompute endpoint. Further documentation on MaxCompute [ODPS URL](https://www.alibabacloud.com/help/en/maxcompute/user-guide/endpoints).
+* Example value: `http://service.ap-southeast-5.maxcompute.aliyun.com/api`
+* Type: `required`
+
+
+## SINK_MAXCOMPUTE_ACCESS_ID
+
+Contains the access id of the MaxCompute project. Further documentation on MaxCompute [Access ID](https://www.alibabacloud.com/help/en/tablestore/support/obtain-an-accesskey-pair).
+* Example value: `access-id`
+* Type: `required`
+
+## SINK_MAXCOMPUTE_ACCESS_KEY
+
+Contains the access key of the MaxCompute project. Further documentation on MaxCompute [Access Key](https://www.alibabacloud.com/help/en/tablestore/support/obtain-an-accesskey-pair).
+* Example value: `access-key`
+* Type: `required`
+
+## SINK_MAXCOMPUTE_PROJECT_ID
+
+Contains the identifier of a MaxCompute project. Further documentation on MaxCompute [Project ID](https://www.alibabacloud.com/help/en/maxcompute/product-overview/project).
+* Example value: `project-id`
+* Type: `required`
+
+## SINK_MAXCOMPUTE_ADD_METADATA_ENABLED
+
+Configuration for enabling metadata in top of the record. This config will be used for adding metadata information to the record. Metadata information will be added to the record in the form of key-value pair.
+* Example value: `true`
+* Type: `optional`
+* Default value: `true`
+
+## SINK_MAXCOMPUTE_METADATA_NAMESPACE
+
+Configuration for wrapping the metadata fields under a specific namespace. This will result in the metadata fields contained in a struct. 
+Empty string will result in the metadata fields being added directly to the root level.
+* Example value: `__kafka_metadata`
+* Type: `optional`
+
+## SINK_MAXCOMPUTE_METADATA_COLUMNS_TYPES
+
+Configuration for defining the metadata columns and their types. This config will be used for defining the metadata columns and their types. The format of this config is `column1=type1,column2=type2`.
+Supported types are `string`, `integer`, `long`, `timestamp`, `float`, `double`, `boolean`.
+
+* Example value: `topic=string,partition=integer,offset=integer,timestamp=timestamp`
+* Type: `optional`
+
+## SINK_MAXCOMPUTE_SCHEMA
+
+Contains the schema of the MaxCompute table. Schema is a dataset grouping of table columns. Further documentation on MaxCompute [Schema](https://www.alibabacloud.com/help/en/maxcompute/user-guide/schemas).
+* Example value: `column1=string,column2=integer,column3=double`
+* Type: `required`
+* Default value: `default`
+
+## SINK_MAXCOMPUTE_TABLE_PARTITIONING_ENABLE
+
+Configuration for enabling partitioning in the MaxCompute table. This config will be used for enabling partitioning in the MaxCompute table.
+* Example value: `true`
+* Type: `optional`
+* Default value: `false`
+
+## SINK_MAXCOMPUTE_TABLE_PARTITION_KEY
+
+Contains the partition key of the MaxCompute table. Partition key is referring to the payload field that will be used as partition key in the MaxCompute table.
+Supported MaxCompute type for partition key is `string`, `tinyint`, `smallint`, `int`, `bigint`, `timestamp_ntz`.
+* Example value: `column1`
+* Type: `optional`
+* Default value: `default`
+
+## SINK_MAXCOMPUTE_TABLE_PARTITION_COLUMN_NAME
+
+Contains the partition column name of the MaxCompute table. This could be the same as the partition key or different. This will reflect the column name in the MaxCompute table.
+* Example value: `column1`
+* Type: `required`
+
+## SINK_MAXCOMPUTE_TABLE_NAME
+
+Contains the name of the MaxCompute table. Further documentation on MaxCompute [Table Name](https://www.alibabacloud.com/help/en/maxcompute/user-guide/tables).
+* Example value: `table_name`
+* Type: `required`
+
+## SINK_MAXCOMPUTE_TABLE_LIFECYCLE_DAYS
+
+Contains the lifecycle of the MaxCompute table. This config will be used for setting the lifecycle of the MaxCompute table.
+Not setting this config will result in table with lifecycle. Lifecycle is applied at partition level. Further documentation on MaxCompute [Table Lifecycle](https://www.alibabacloud.com/help/en/maxcompute/product-overview/lifecycle).
+* Example value: `30`
+* Type: `optional`
+
+## SINK_MAXCOMPUTE_RECORD_PACK_FLUSH_TIMEOUT_MS
+
+Contains the timeout for flushing the record pack in milliseconds. This config will be used for setting the timeout for flushing the record pack.
+* Example value: `1000`
+* Type: `optional`
+* Default value: `-1`
+
+## SINK_MAXCOMPUTE_STREAMING_INSERT_COMPRESSION_ENABLED
+
+Configuration for enabling compression in the streaming insert operation. This config will be used for enabling compression in the streaming insert operation.
+* Example value: `true`
+* Type: `optional`
+* Default value: `false`
+
+## SINK_MAXCOMPUTE_STREAMING_INSERT_COMPRESSION_ALGORITHM
+
+Configuration for defining the compression algorithm in the streaming insert operation. This config will be used for defining the compression algorithm in the streaming insert operation.
+Supported algorithms are ODPS_RAW, ODPS_ZLIB, ODPS_LZ4_FRAME, ODPS_ARROW_LZ4_FRAME, ODPS_ARROW_ZSTD
+* Example value: `ODPS_ZLIB`
+* Type: `optional`
+* Default value: `ODPS_LZ4_FRAME`
+
+## SINK_MAXCOMPUTE_STREAMING_INSERT_COMPRESSION_LEVEL
+
+Configuration for defining the compression level in the streaming insert operation. This config will be used for defining the compression level in the streaming insert operation.
+* Example value: `1`
+* Type: `optional`
+* Default value: `1`
+
+## SINK_MAXCOMPUTE_STREAMING_INSERT_COMPRESSION_STRATEGY
+
+Configuration for defining the compression strategy in the streaming insert operation. This config will be used for defining the compression strategy in the streaming insert operation.
+* Example value: `1`
+* Type: `optional`
+* Default value: `0`
+
+## SINK_MAXCOMPUTE_ZONE_ID
+
+Contains ZoneID used for parsing the timestamp in the record. This config will be used for parsing the timestamp in the record.
+
+* Example value: `Asia/Bangkok`
+* Type: `optional`
+* Default value: `Asia/Bangkok`

docs/sinks/maxcompute.md

@@ -0,0 +1,54 @@
+# MaxCompute sink
+
+### Datatype Protobuf
+
+MaxCompute sink has several responsibilities, including :
+
+1. Creation of MaxCompute table if it does not exist.
+2. Updating the MaxCompute table schema based on the latest protobuf schema.
+3. Translating protobuf messages into MaxCompute compatible records and inserting them into MaxCompute tables.
+
+## MaxCompute Table Schema Update
+
+### Protobuf
+
+MaxCompute Sink update the MaxCompute table schema on separate table update operation. MaxCompute
+utilise [Stencil](https://github.com/goto/stencil) to parse protobuf messages generate schema and update MaxCompute
+tables with the latest schema.
+The stencil client periodically reload the descriptor cache. Table schema update happened after the descriptor caches
+uploaded.
+
+#### Supported Protobuf - MaxCompute Table Type Mapping
+
+| Protobuf Type                                                                      | MaxCompute Type             |
+|------------------------------------------------------------------------------------|-----------------------------|
+| bytes                                                                              | BINARY                      |
+| string                                                                             | STRING                      |
+| enum                                                                               | STRING                      |
+| float                                                                              | FLOAT                       |
+| double                                                                             | DOUBLE                      |
+| bool                                                                               | BOOLEAN                     |
+| int64, uint64, int32, uint32, fixed64, fixed32, sfixed64, sfixed32, sint64, sint32 | BIGINT                      |
+| message                                                                            | STRUCT                      |
+| .google.protobuf.Timestamp                                                         | TIMESTAMP_NTZ               |
+| .google.protobuf.Struct                                                            | STRING (Json Serialised)    |
+| .google.protobuf.Duration                                                          | STRUCT                      |
+| map<k,v>                                                                           | ARRAY<STRUCT<key:k, value:v |
+
+## Partitioning
+
+MaxCompute Sink supports creation of table with partition configuration. Currently, MaxCompute Sink supports primitive field(STRING, TINYINT, SMALLINT, BIGINT) 
+and timestamp field based partitioning. Timestamp based partitioning strategy introduces a pseudo-partition column with the value of the timestamp field truncated to the nearest start of day.
+
+## Clustering
+
+MaxCompute Sink currently does not support clustering.
+
+## Metadata
+
+For data quality checking purposes, sometimes some metadata need to be added on the record.
+if `SINK_MAXCOMPUTE_ADD_METADATA_ENABLED` is true then the metadata will be added.
+`SINK_MAXCOMPUTE_METADATA_NAMESPACE` is used for another namespace to add columns
+if namespace is empty, the metadata columns will be added in the root level.
+`SINK_MAXCOMPUTE_METADATA_COLUMNS_TYPES` is set with kafka metadata column and their type,
+An example of metadata columns that can be added for kafka records.

src/main/java/com/gotocompany/depot/config/MaxComputeSinkConfig.java

@@ -0,0 +1,90 @@
+package com.gotocompany.depot.config;
+
+import com.aliyun.odps.tunnel.io.CompressOption;
+import com.gotocompany.depot.common.TupleString;
+import com.gotocompany.depot.config.converter.ConfToListConverter;
+import com.gotocompany.depot.config.converter.MaxComputeCompressionAlgorithmConverter;
+import com.gotocompany.depot.config.converter.ZoneIdConverter;
+import org.aeonbits.owner.Config;
+
+import java.time.ZoneId;
+import java.util.List;
+
+public interface MaxComputeSinkConfig extends Config {
+
+    @Key("SINK_MAXCOMPUTE_ODPS_URL")
+    String getMaxComputeOdpsUrl();
+
+    @Key("SINK_MAXCOMPUTE_TUNNEL_URL")
+    String getMaxComputeTunnelUrl();
+
+    @Key("SINK_MAXCOMPUTE_ACCESS_ID")
+    String getMaxComputeAccessId();
+
+    @Key("SINK_MAXCOMPUTE_ACCESS_KEY")
+    String getMaxComputeAccessKey();
+
+    @Key("SINK_MAXCOMPUTE_PROJECT_ID")
+    String getMaxComputeProjectId();
+
+    @Key("SINK_MAXCOMPUTE_METADATA_NAMESPACE")
+    @DefaultValue("")
+    String getMaxcomputeMetadataNamespace();
+
+    @DefaultValue("true")
+    @Key("SINK_MAXCOMPUTE_ADD_METADATA_ENABLED")
+    boolean shouldAddMetadata();
+
+    @DefaultValue("")
+    @Key("SINK_MAXCOMPUTE_METADATA_COLUMNS_TYPES")
+    @ConverterClass(ConfToListConverter.class)
+    @Separator(ConfToListConverter.ELEMENT_SEPARATOR)
+    List<TupleString> getMetadataColumnsTypes();
+
+    @Key("SINK_MAXCOMPUTE_SCHEMA")
+    @DefaultValue("default")
+    String getMaxComputeSchema();
+
+    @Key("SINK_MAXCOMPUTE_TABLE_PARTITIONING_ENABLE")
+    @DefaultValue("false")
+    Boolean isTablePartitioningEnabled();
+
+    @Key("SINK_MAXCOMPUTE_TABLE_PARTITION_KEY")
+    String getTablePartitionKey();
+
+    @Key("SINK_MAXCOMPUTE_TABLE_PARTITION_COLUMN_NAME")
+    String getTablePartitionColumnName();
+
+    @Key("SINK_MAXCOMPUTE_TABLE_NAME")
+    String getMaxComputeTableName();
+
+    @Key("SINK_MAXCOMPUTE_TABLE_LIFECYCLE_DAYS")
+    Long getMaxComputeTableLifecycleDays();
+
+    @Key("SINK_MAXCOMPUTE_RECORD_PACK_FLUSH_TIMEOUT_MS")
+    @DefaultValue("-1")
+    Long getMaxComputeRecordPackFlushTimeoutMs();
+
+    @Key("SINK_MAXCOMPUTE_STREAMING_INSERT_COMPRESSION_ENABLED")
+    @DefaultValue("false")
+    boolean isStreamingInsertCompressEnabled();
+
+    @Key("SINK_MAXCOMPUTE_STREAMING_INSERT_COMPRESSION_ALGORITHM")
+    @ConverterClass(MaxComputeCompressionAlgorithmConverter.class)
+    @DefaultValue("ODPS_LZ4_FRAME")
+    CompressOption.CompressAlgorithm getMaxComputeCompressionAlgorithm();
+
+    @Key("SINK_MAXCOMPUTE_STREAMING_INSERT_COMPRESSION_LEVEL")
+    @DefaultValue("1")
+    int getMaxComputeCompressionLevel();
+
+    @Key("SINK_MAXCOMPUTE_STREAMING_INSERT_COMPRESSION_STRATEGY")
+    @DefaultValue("0")
+    int getMaxComputeCompressionStrategy();
+
+    @Key("SINK_MAXCOMPUTE_ZONE_ID")
+    @ConverterClass(ZoneIdConverter.class)
+    @DefaultValue("Asia/Bangkok")
+    ZoneId getZoneId();
+
+}

src/main/java/com/gotocompany/depot/config/SinkConfig.java

@@ -1,11 +1,13 @@
 package com.gotocompany.depot.config;
 
 import com.gotocompany.depot.config.converter.ConfToListConverter;
+import com.gotocompany.depot.config.converter.ProtoUnknownFiledValidationTypeConverter;
 import com.gotocompany.depot.config.converter.SchemaRegistryHeadersConverter;
 import com.gotocompany.depot.config.converter.SchemaRegistryRefreshConverter;
 import com.gotocompany.depot.config.converter.SinkConnectorSchemaDataTypeConverter;
 import com.gotocompany.depot.config.converter.SinkConnectorSchemaMessageModeConverter;
 import com.gotocompany.depot.config.enums.SinkConnectorSchemaDataType;
+import com.gotocompany.depot.message.ProtoUnknownFieldValidationType;
 import com.gotocompany.depot.message.SinkConnectorSchemaMessageMode;
 import com.gotocompany.depot.common.TupleString;
 import com.gotocompany.stencil.cache.SchemaRefreshStrategy;
@@ -99,4 +101,9 @@ public interface SinkConfig extends Config {
     @DefaultValue("true")
     boolean getSinkDefaultFieldValueEnable();
 
+    @Key("SINK_CONNECTOR_SCHEMA_PROTO_UNKNOWN_FIELDS_VALIDATION")
+    @DefaultValue("MESSAGE")
+    @ConverterClass(ProtoUnknownFiledValidationTypeConverter.class)
+    ProtoUnknownFieldValidationType getSinkConnectorSchemaProtoUnknownFieldsValidation();
+
 }

src/main/java/com/gotocompany/depot/config/converter/MaxComputeCompressionAlgorithmConverter.java

@@ -0,0 +1,25 @@
+package com.gotocompany.depot.config.converter;
+
+import com.aliyun.odps.tunnel.io.CompressOption;
+import com.gotocompany.depot.exception.ConfigurationException;
+import org.aeonbits.owner.Converter;
+
+import java.lang.reflect.Method;
+import java.util.Arrays;
+
+public class MaxComputeCompressionAlgorithmConverter implements Converter<CompressOption.CompressAlgorithm> {
+
+    private static final String INVALID_ENUM_MESSAGE_FORMAT = "Invalid compression algorithm: %s valid values are: %s";
+
+    @Override
+    public CompressOption.CompressAlgorithm convert(Method method, String s) {
+        try {
+            return CompressOption.CompressAlgorithm.valueOf(s.toUpperCase());
+        } catch (IllegalArgumentException e) {
+            throw new ConfigurationException(String.format(INVALID_ENUM_MESSAGE_FORMAT, s,
+                    Arrays.toString(CompressOption.CompressAlgorithm.values())), e);
+        }
+
+    }
+
+}

src/main/java/com/gotocompany/depot/config/converter/ProtoUnknownFiledValidationTypeConverter.java

@@ -0,0 +1,15 @@
+package com.gotocompany.depot.config.converter;
+
+import com.gotocompany.depot.message.ProtoUnknownFieldValidationType;
+import org.aeonbits.owner.Converter;
+
+import java.lang.reflect.Method;
+
+public class ProtoUnknownFiledValidationTypeConverter implements Converter<ProtoUnknownFieldValidationType> {
+
+    @Override
+    public ProtoUnknownFieldValidationType convert(Method method, String s) {
+        return ProtoUnknownFieldValidationType.valueOf(s);
+    }
+
+}