Feat max compute sink payload mappers

build.gradle

@@ -44,6 +44,7 @@ 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.48.8-public'
     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'
@@ -54,6 +55,7 @@ dependencies {
     implementation group: 'com.squareup.okhttp3', name: 'okhttp', version: '4.9.1'
     implementation group: 'joda-time', name: 'joda-time', version: '2.10.2'
     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'

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

@@ -0,0 +1,71 @@
+package com.gotocompany.depot.config;
+
+import com.gotocompany.depot.common.TupleString;
+import com.gotocompany.depot.config.converter.ConfToListConverter;
+import org.aeonbits.owner.Config;
+
+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_PARTITION_BY_TIMESTAMP_TIMEZONE")
+    @DefaultValue("UTC+7")
+    String getTablePartitionByTimestampTimezone();
+
+    @Key("SINK_MAX_COMPUTE_TABLE_PARTITION_BY_TIMESTAMP_ZONE_OFFSET")
+    @DefaultValue("+07:00")
+    String getTablePartitionByTimestampZoneOffset();
+
+    @Key("SINK_MAXCOMPUTE_TABLE_NAME")
+    String getMaxComputeTableName();
+
+    @Key("SINK_MAXCOMPUTE_TABLE_LIFECYCLE_DAYS")
+    Long getMaxComputeTableLifecycleDays();
+
+    @Key("SINK_MAXCOMPUTE_RECORD_PACK_FLUSH_TIMEOUT")
+    @DefaultValue("-1")
+    Long getMaxComputeRecordPackFlushTimeout();
+
+}

src/main/java/com/gotocompany/depot/maxcompute/MaxComputeSink.java

@@ -0,0 +1,41 @@
+package com.gotocompany.depot.maxcompute;
+
+import com.gotocompany.depot.Sink;
+import com.gotocompany.depot.SinkResponse;
+import com.gotocompany.depot.error.ErrorInfo;
+import com.gotocompany.depot.error.ErrorType;
+import com.gotocompany.depot.exception.SinkException;
+import com.gotocompany.depot.maxcompute.client.MaxComputeClient;
+import com.gotocompany.depot.maxcompute.converter.record.RecordConverter;
+import com.gotocompany.depot.maxcompute.model.RecordWrappers;
+import com.gotocompany.depot.message.Message;
+import lombok.RequiredArgsConstructor;
+
+import java.io.IOException;
+import java.util.List;
+
+@RequiredArgsConstructor
+public class MaxComputeSink implements Sink {
+
+    private final MaxComputeClient maxComputeClient;
+    private final RecordConverter recordConverter;
+
+    @Override
+    public SinkResponse pushToSink(List<Message> messages) throws SinkException {
+        SinkResponse sinkResponse = new SinkResponse();
+        RecordWrappers recordWrappers = recordConverter.convert(messages);
+        recordWrappers.getInvalidRecords()
+                .forEach(invalidRecord -> sinkResponse.getErrors().put(invalidRecord.getIndex(), invalidRecord.getErrorInfo()));
+        try {
+            maxComputeClient.insert(recordWrappers.getValidRecords());
+        } catch (Exception e) {
+            recordWrappers.getValidRecords()
+                    .forEach(validRecord -> sinkResponse.getErrors().put(validRecord.getIndex(), new ErrorInfo(e, ErrorType.DEFAULT_ERROR)));
+        }
+        return sinkResponse;
+    }
+
+    @Override
+    public void close() throws IOException {}
+
+}

src/main/java/com/gotocompany/depot/maxcompute/client/MaxComputeClient.java

@@ -0,0 +1,78 @@
+package com.gotocompany.depot.maxcompute.client;
+
+import com.aliyun.odps.Odps;
+import com.aliyun.odps.OdpsException;
+import com.aliyun.odps.TableSchema;
+import com.aliyun.odps.account.Account;
+import com.aliyun.odps.account.AliyunAccount;
+import com.aliyun.odps.tunnel.TableTunnel;
+import com.gotocompany.depot.config.MaxComputeSinkConfig;
+import com.gotocompany.depot.maxcompute.client.insert.InsertManager;
+import com.gotocompany.depot.maxcompute.client.insert.NonPartitionedInsertManager;
+import com.gotocompany.depot.maxcompute.client.insert.PartitionedInsertManager;
+import com.gotocompany.depot.maxcompute.model.RecordWrapper;
+import com.gotocompany.depot.maxcompute.schema.partition.PartitioningStrategy;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class MaxComputeClient {
+
+    private final Odps odps;
+    private final MaxComputeSinkConfig maxComputeSinkConfig;
+    private final TableTunnel tableTunnel;
+    private final PartitioningStrategy partitioningStrategy;
+    private final InsertManager insertManager;
+
+    public MaxComputeClient(MaxComputeSinkConfig maxComputeSinkConfig,
+                            PartitioningStrategy partitioningStrategy) {
+        this.maxComputeSinkConfig = maxComputeSinkConfig;
+        this.partitioningStrategy = partitioningStrategy;
+        this.odps = initializeOdps();
+        this.tableTunnel = new TableTunnel(odps);
+        this.insertManager = initializeInsertManager();
+    }
+
+    public void upsertTable(TableSchema tableSchema) throws OdpsException {
+        String tableName = maxComputeSinkConfig.getMaxComputeTableName();
+        if (!this.odps.tables().exists(tableName)) {
+            this.odps.tables().create(odps.getDefaultProject(), tableName, tableSchema, "",
+                    false, maxComputeSinkConfig.getMaxComputeTableLifecycleDays(),
+                    null, null);
+        }
+    }
+
+    public void insert(List<RecordWrapper> recordWrappers) {
+        try {
+            insertManager.insert(recordWrappers);
+        } catch (Exception e) {
+            throw new RuntimeException("Failed to insert records into MaxCompute", e);
+        }
+    }
+
+    private Odps initializeOdps() {
+        Account account = new AliyunAccount(maxComputeSinkConfig.getMaxComputeAccessId(), maxComputeSinkConfig.getMaxComputeAccessKey());
+        Odps odpsClient = new Odps(account);
+        odpsClient.setDefaultProject(maxComputeSinkConfig.getMaxComputeProjectId());
+        odpsClient.setEndpoint(maxComputeSinkConfig.getMaxComputeOdpsUrl());
+        odpsClient.setCurrentSchema(maxComputeSinkConfig.getMaxComputeSchema());
+        odpsClient.setGlobalSettings(getGlobalSettings());
+        return odpsClient;
+    }
+
+    private InsertManager initializeInsertManager() {
+        if (maxComputeSinkConfig.isTablePartitioningEnabled()) {
+            return new PartitionedInsertManager(partitioningStrategy, tableTunnel, maxComputeSinkConfig);
+        }
+        return new NonPartitionedInsertManager(tableTunnel, maxComputeSinkConfig);
+    }
+
+    private Map<String, String> getGlobalSettings() {
+        Map<String, String> globalSettings = new HashMap<>();
+        globalSettings.put("setproject odps.schema.evolution.enable", "true");
+        globalSettings.put("odps.namespace.schema", "true");
+        return globalSettings;
+    }
+
+}

src/main/java/com/gotocompany/depot/maxcompute/client/insert/InsertManager.java

@@ -0,0 +1,11 @@
+package com.gotocompany.depot.maxcompute.client.insert;
+
+import com.aliyun.odps.tunnel.TunnelException;
+import com.gotocompany.depot.maxcompute.model.RecordWrapper;
+
+import java.io.IOException;
+import java.util.List;
+
+public interface InsertManager {
+    void insert(List<RecordWrapper> recordWrappers) throws TunnelException, IOException;
+}

src/main/java/com/gotocompany/depot/maxcompute/client/insert/NonPartitionedInsertManager.java

@@ -0,0 +1,38 @@
+package com.gotocompany.depot.maxcompute.client.insert;
+
+import com.aliyun.odps.tunnel.TableTunnel;
+import com.aliyun.odps.tunnel.TunnelException;
+import com.gotocompany.depot.config.MaxComputeSinkConfig;
+import com.gotocompany.depot.maxcompute.model.RecordWrapper;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.util.List;
+
+@RequiredArgsConstructor
+@Slf4j
+public class NonPartitionedInsertManager implements InsertManager {
+
+    private final TableTunnel tableTunnel;
+    private final MaxComputeSinkConfig maxComputeSinkConfig;
+
+    @Override
+    public void insert(List<RecordWrapper> recordWrappers) throws TunnelException, IOException {
+        TableTunnel.StreamUploadSession streamUploadSession = getStreamUploadSession();
+        TableTunnel.StreamRecordPack recordPack = streamUploadSession.newRecordPack();
+        for (RecordWrapper recordWrapper : recordWrappers) {
+            recordPack.append(recordWrapper.getRecord());
+        }
+        TableTunnel.FlushResult flushResult = recordPack.flush(
+                new TableTunnel.FlushOption()
+                        .timeout(maxComputeSinkConfig.getMaxComputeRecordPackFlushTimeout()));
+        log.info("Flushed {} records", flushResult.getRecordCount());
+    }
+
+    private TableTunnel.StreamUploadSession getStreamUploadSession() throws TunnelException {
+        return tableTunnel.buildStreamUploadSession(maxComputeSinkConfig.getMaxComputeProjectId(),
+                        maxComputeSinkConfig.getMaxComputeTableName())
+                .build();
+    }
+}

src/main/java/com/gotocompany/depot/maxcompute/client/insert/PartitionedInsertManager.java

@@ -0,0 +1,49 @@
+package com.gotocompany.depot.maxcompute.client.insert;
+
+import com.aliyun.odps.PartitionSpec;
+import com.aliyun.odps.tunnel.TableTunnel;
+import com.aliyun.odps.tunnel.TunnelException;
+import com.gotocompany.depot.config.MaxComputeSinkConfig;
+import com.gotocompany.depot.maxcompute.model.RecordWrapper;
+import com.gotocompany.depot.maxcompute.schema.partition.PartitioningStrategy;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+@RequiredArgsConstructor
+@Slf4j
+public class PartitionedInsertManager implements InsertManager {
+
+    private final PartitioningStrategy partitioningStrategy;
+    private final TableTunnel tableTunnel;
+    private final MaxComputeSinkConfig maxComputeSinkConfig;
+
+    @Override
+    public void insert(List<RecordWrapper> recordWrappers) throws TunnelException, IOException {
+        Map<PartitionSpec, List<RecordWrapper>> partitionSpecRecordWrapperMap = recordWrappers.stream()
+                .collect(Collectors.groupingBy(partitioningStrategy::getPartitionSpec));
+        for (Map.Entry<PartitionSpec, List<RecordWrapper>> entry : partitionSpecRecordWrapperMap.entrySet()) {
+            TableTunnel.StreamUploadSession streamUploadSession = getStreamUploadSession(entry.getKey());
+            TableTunnel.StreamRecordPack recordPack = streamUploadSession.newRecordPack();
+            for (RecordWrapper recordWrapper : entry.getValue()) {
+                recordPack.append(recordWrapper.getRecord());
+            }
+            TableTunnel.FlushResult flushResult = recordPack.flush(
+                    new TableTunnel.FlushOption()
+                            .timeout(maxComputeSinkConfig.getMaxComputeRecordPackFlushTimeout()));
+            log.info("Flushed {} records to partition {}", flushResult.getRecordCount(), entry.getKey());
+        }
+    }
+
+    private TableTunnel.StreamUploadSession getStreamUploadSession(PartitionSpec partitionSpec) throws TunnelException {
+        return tableTunnel.buildStreamUploadSession(maxComputeSinkConfig.getMaxComputeProjectId(),
+                        maxComputeSinkConfig.getMaxComputeTableName())
+                .setPartitionSpec(partitionSpec)
+                .build();
+    }
+
+}

src/main/java/com/gotocompany/depot/maxcompute/converter/ConverterOrchestrator.java

@@ -0,0 +1,72 @@
+package com.gotocompany.depot.maxcompute.converter;
+
+import com.aliyun.odps.type.TypeInfo;
+import com.google.protobuf.Descriptors;
+import com.gotocompany.depot.maxcompute.converter.payload.DurationPayloadConverter;
+import com.gotocompany.depot.maxcompute.converter.payload.MessagePayloadConverter;
+import com.gotocompany.depot.maxcompute.converter.payload.PayloadConverter;
+import com.gotocompany.depot.maxcompute.converter.payload.PrimitivePayloadConverter;
+import com.gotocompany.depot.maxcompute.converter.payload.StructPayloadConverter;
+import com.gotocompany.depot.maxcompute.converter.payload.TimestampPayloadConverter;
+import com.gotocompany.depot.maxcompute.converter.type.DurationTypeInfoConverter;
+import com.gotocompany.depot.maxcompute.converter.type.MessageTypeInfoConverter;
+import com.gotocompany.depot.maxcompute.converter.type.PrimitiveTypeInfoConverter;
+import com.gotocompany.depot.maxcompute.converter.type.StructTypeInfoConverter;
+import com.gotocompany.depot.maxcompute.converter.type.TimestampTypeInfoConverter;
+import com.gotocompany.depot.maxcompute.converter.type.TypeInfoConverter;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class ConverterOrchestrator {
+
+    private final List<TypeInfoConverter> typeInfoConverters;
+    private final List<PayloadConverter> payloadConverters;
+    private final Map<String, TypeInfo> typeInfoCache;
+
+    public ConverterOrchestrator() {
+        typeInfoConverters = new ArrayList<>();
+        payloadConverters = new ArrayList<>();
+        typeInfoCache = new ConcurrentHashMap<>();
+        initializeConverters();
+    }
+
+    public TypeInfo convert(Descriptors.FieldDescriptor fieldDescriptor) {
+        return typeInfoCache.computeIfAbsent(fieldDescriptor.getFullName(), key -> typeInfoConverters.stream()
+                .filter(converter -> converter.canConvert(fieldDescriptor))
+                .findFirst()
+                .map(converter -> converter.convert(fieldDescriptor))
+                .orElseThrow(() -> new IllegalArgumentException("Unsupported type: " + fieldDescriptor.getType())));
+    }
+
+    public Object convert(Descriptors.FieldDescriptor fieldDescriptor, Object object) {
+        return payloadConverters.stream()
+                .filter(converter -> converter.canConvert(fieldDescriptor))
+                .findFirst()
+                .map(converter -> converter.convert(fieldDescriptor, object))
+                .orElseThrow(() -> new IllegalArgumentException("Unsupported type: " + fieldDescriptor.getType()));
+    }
+
+    private void initializeConverters() {
+        PrimitiveTypeInfoConverter primitiveTypeInfoConverter = new PrimitiveTypeInfoConverter();
+        DurationTypeInfoConverter durationTypeInfoConverter = new DurationTypeInfoConverter();
+        StructTypeInfoConverter structTypeInfoConverter = new StructTypeInfoConverter();
+        TimestampTypeInfoConverter timestampTypeInfoConverter = new TimestampTypeInfoConverter();
+        MessageTypeInfoConverter messageTypeInfoConverter = new MessageTypeInfoConverter(typeInfoConverters);
+
+        typeInfoConverters.add(primitiveTypeInfoConverter);
+        typeInfoConverters.add(durationTypeInfoConverter);
+        typeInfoConverters.add(structTypeInfoConverter);
+        typeInfoConverters.add(timestampTypeInfoConverter);
+        typeInfoConverters.add(messageTypeInfoConverter);
+
+        payloadConverters.add(new PrimitivePayloadConverter(primitiveTypeInfoConverter));
+        payloadConverters.add(new DurationPayloadConverter(durationTypeInfoConverter));
+        payloadConverters.add(new StructPayloadConverter(structTypeInfoConverter));
+        payloadConverters.add(new TimestampPayloadConverter(timestampTypeInfoConverter));
+        payloadConverters.add(new MessagePayloadConverter(messageTypeInfoConverter, payloadConverters));
+    }
+
+}