content: package com.alibaba.hologres.connector.flink.example;
  
  import com.alibaba.hologres.client.Put.MutationType;
  import com.alibaba.hologres.client.exception.HoloClientException;
  import com.alibaba.hologres.client.model.Record;
  import com.alibaba.hologres.connector.flink.api.HologresRecordConverter;
  import com.alibaba.hologres.connector.flink.api.HologresTableSchema;
  import com.alibaba.hologres.connector.flink.config.HologresConfigs;
  import com.alibaba.hologres.connector.flink.config.HologresConnectionParam;
  import com.alibaba.hologres.connector.flink.example.SourceItem.EventType;
  import com.alibaba.hologres.connector.flink.jdbc.HologresJDBCWriter;
  import com.alibaba.hologres.connector.flink.sink.v1.AbstractHologresOutputFormat;
  import com.alibaba.hologres.connector.flink.utils.SchemaUtil;
  import org.apache.flink.api.common.typeinfo.Types;
  import org.apache.flink.configuration.Configuration;
  import org.apache.flink.streaming.api.datastream.DataStreamSource;
  import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  import org.apache.flink.streaming.api.functions.sink.OutputFormatSinkFunction;
  import org.apache.flink.table.api.DataTypes;
  import org.apache.flink.table.api.TableSchema;
  
  import java.io.IOException;
  import java.math.BigDecimal;
  import java.sql.Timestamp;
  
  /**
   * A Flink data streak example sinking data to Hologres.
   */
  public class FlinkDataStreamUserDefineItemToHoloExample {
      /**
       * Hologres DDL. create table sink_table(user_id bigint, user_name text, price decimal(38,
       * 2),sale_timestamp timestamptz);
       *
       * @param args
       * @throws Exception
       */
      public static void main(String[] args) throws Exception {
          SettingHelper settingHelper = new SettingHelper(args, "setting.properties");
          String endpoint = settingHelper.getEndpoint();
          String username = settingHelper.getUsername();
          String password = settingHelper.getPassword();
          String database = settingHelper.getDatabase();
          String tableName = "sink_table";
  
          StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  
          DataStreamSource<SourceItem> source =
                  env.fromElements(
                          new SourceItem(
                                  123L,
                                  "Adam",
                                  new BigDecimal("123.11"),
                                  new Timestamp(System.currentTimeMillis())),
                          new SourceItem(
                                  234,
                                  "Bob",
                                  new BigDecimal("000.11"),
                                  new Timestamp(System.currentTimeMillis())));
  
          TableSchema schema =
                  TableSchema.builder()
                          .field("user_id", DataTypes.BIGINT())
                          .field("user_name", DataTypes.STRING())
                          .field("price", DataTypes.DECIMAL(38, 2))
                          .field("sale_timestamp", Types.SQL_TIMESTAMP)
                          .build();
  
          Configuration configuration = new Configuration();
          configuration.set(HologresConfigs.ENDPOINT, endpoint);
          configuration.set(HologresConfigs.DATABASE, database);
          configuration.set(HologresConfigs.USERNAME, username);
          configuration.set(HologresConfigs.PASSWORD, password);
          configuration.set(HologresConfigs.TABLE, tableName);
          HologresConnectionParam hologresConnectionParam =
                  new HologresConnectionParam(configuration);
  
          source.addSink(
                  new OutputFormatSinkFunction<SourceItem>(
                          new AbstractHologresOutputFormat<SourceItem>(
                                  hologresConnectionParam,
                                  new HologresJDBCWriter<>(
                                          hologresConnectionParam,
                                          schema.getFieldNames(),
                                          SchemaUtil.getLogicalTypes(schema),
                                          new RecordConverter(hologresConnectionParam))) {
                              @Override
                              public long writeData(SourceItem data) throws HoloClientException, IOException {
                                  if (data.eventType == EventType.DELETE) {
                                      return hologresIOClient.writeDeleteRecord(data);
                                  } else {
                                      return hologresIOClient.writeAddRecord(data);
                                  }
                              }
                          }));
          env.execute("Insert");
      }
  
      /**
       * 将用户POJO数据转换至Hologres Record的实现.
       */
      public static class RecordConverter implements HologresRecordConverter<SourceItem, Record> {
          private HologresConnectionParam hologresConnectionParam;
          private HologresTableSchema tableSchema;
  
          public RecordConverter(HologresConnectionParam hologresConnectionParam) {
              this.hologresConnectionParam = hologresConnectionParam;
          }
  
          @Override
          public Record convertFrom(SourceItem record) {
              if (tableSchema == null) {
                  this.tableSchema =
                          HologresTableSchema.get(hologresConnectionParam.getJdbcOptions());
              }
              Record result = new Record(tableSchema.get());
              result.setObject(0, record.userId);
              result.setObject(1, record.userName);
              result.setObject(2, record.price);
              result.setObject(3, record.saleTimestamp);
              /* 在DataStream作业中，用户需要使用自定义的OutputFormatSinkFunction以及RecordConverter，如果要支持消息的回撤，需要在此处对convert结果设置MutationType。 需要hologres-connector 1.3.2及以上版本 */
              if (record.eventType == EventType.DELETE) {
                  result.setType(MutationType.DELETE);
              }
              return result;
          }
  
          @Override
          public SourceItem convertTo(Record record) {
              throw new UnsupportedOperationException("No need to implement");
          }
  
          @Override
          public Record convertToPrimaryKey(SourceItem record) {
              throw new UnsupportedOperationException("No need to implement");
          }
      }
  }
  