content: /*
   * 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.connectors.kudu.table.dynamic;
  
  import org.apache.flink.configuration.ConfigOption;
  import org.apache.flink.configuration.ConfigOptions;
  import org.apache.flink.configuration.ReadableConfig;
  import org.apache.flink.connectors.kudu.connector.KuduTableInfo;
  import org.apache.flink.connectors.kudu.connector.reader.KuduReaderConfig;
  import org.apache.flink.connectors.kudu.connector.writer.KuduWriterConfig;
  import org.apache.flink.connectors.kudu.table.function.lookup.KuduLookupOptions;
  import org.apache.flink.connectors.kudu.table.utils.KuduTableUtils;
  import org.apache.flink.table.api.TableSchema;
  import org.apache.flink.table.connector.sink.DynamicTableSink;
  import org.apache.flink.table.connector.source.DynamicTableSource;
  import org.apache.flink.table.factories.DynamicTableSinkFactory;
  import org.apache.flink.table.factories.DynamicTableSourceFactory;
  import org.apache.flink.table.factories.FactoryUtil;
  import org.apache.kudu.shaded.com.google.common.collect.Sets;
  
  import java.util.Optional;
  import java.util.Set;
  
  /**
   * Factory for creating configured instances of {@link KuduDynamicTableSource}/{@link KuduDynamicTableSink} in
   * a stream environment.
   */
  public class KuduDynamicTableSourceSinkFactory implements DynamicTableSourceFactory, DynamicTableSinkFactory {
      public static final String IDENTIFIER = "kudu";
      public static final ConfigOption<String> KUDU_TABLE = ConfigOptions
              .key("kudu.table")
              .stringType()
              .noDefaultValue()
              .withDescription("kudu's table name");
  
      public static final ConfigOption<String> KUDU_MASTERS =
              ConfigOptions
                      .key("kudu.masters")
                      .stringType()
                      .noDefaultValue()
                      .withDescription("kudu's master server address");
  
  
      public static final ConfigOption<String> KUDU_HASH_COLS =
              ConfigOptions
                      .key("kudu.hash-columns")
                      .stringType()
                      .noDefaultValue()
                      .withDescription("kudu's hash columns");
  
      public static final ConfigOption<Integer> KUDU_REPLICAS =
              ConfigOptions
                      .key("kudu.replicas")
                      .intType()
                      .defaultValue(3)
                      .withDescription("kudu's replica nums");
  
      public static final ConfigOption<Integer> KUDU_MAX_BUFFER_SIZE =
              ConfigOptions
                      .key("kudu.max-buffer-size")
                      .intType()
                      .noDefaultValue()
                      .withDescription("kudu's max buffer size");
  
      public static final ConfigOption<Integer> KUDU_FLUSH_INTERVAL =
              ConfigOptions
                      .key("kudu.flush-interval")
                      .intType()
                      .noDefaultValue()
                      .withDescription("kudu's data flush interval");
  
      public static final ConfigOption<Long> KUDU_OPERATION_TIMEOUT =
              ConfigOptions
                      .key("kudu.operation-timeout")
                      .longType()
                      .noDefaultValue()
                      .withDescription("kudu's operation timeout");
  
      public static final ConfigOption<Boolean> KUDU_IGNORE_NOT_FOUND =
              ConfigOptions
                      .key("kudu.ignore-not-found")
                      .booleanType()
                      .noDefaultValue()
                      .withDescription("if true, ignore all not found rows");
  
      public static final ConfigOption<Boolean> KUDU_IGNORE_DUPLICATE =
              ConfigOptions
                      .key("kudu.ignore-not-found")
                      .booleanType()
                      .noDefaultValue()
                      .withDescription("if true, ignore all dulicate rows");
  
      /**
       * hash partition bucket nums
       */
      public static final ConfigOption<Integer> KUDU_HASH_PARTITION_NUMS =
              ConfigOptions
                      .key("kudu.hash-partition-nums")
                      .intType()
                      .defaultValue(KUDU_REPLICAS.defaultValue() * 2)
                      .withDescription("kudu's hash partition bucket nums, defaultValue is 2 * replica nums");
  
      public static final ConfigOption<String> KUDU_PRIMARY_KEY_COLS =
              ConfigOptions
                      .key("kudu.primary-key-columns")
                      .stringType()
                      .noDefaultValue()
                      .withDescription("kudu's primary key, primary key must be ordered");
  
  
      public static final ConfigOption<Integer> KUDU_SCAN_ROW_SIZE =
              ConfigOptions
                      .key("kudu.scan.row-size")
                      .intType()
                      .defaultValue(0)
                      .withDescription("kudu's scan row size");
  
      /**
       * lookup cache config
       */
      public static final ConfigOption<Long> KUDU_LOOKUP_CACHE_MAX_ROWS =
              ConfigOptions
                      .key("kudu.lookup.cache.max-rows")
                      .longType()
                      .defaultValue(-1L)
                      .withDescription("the max number of rows of lookup cache, over this value, the oldest rows will " +
                              "be eliminated. \"cache.max-rows\" and \"cache.ttl\" options must all be specified if any" +
                              " of them is " +
                              "specified. Cache is not enabled as default.");
  
      public static final ConfigOption<Long> KUDU_LOOKUP_CACHE_TTL =
              ConfigOptions
                      .key("kudu.lookup.cache.ttl")
                      .longType()
                      .defaultValue(-1L)
                      .withDescription("the cache time to live.");
  
      public static final ConfigOption<Integer> KUDU_LOOKUP_MAX_RETRIES =
              ConfigOptions
                      .key("kudu.lookup.max-retries")
                      .intType()
                      .defaultValue(3)
                      .withDescription("the max retry times if lookup database failed.");
  
      @Override
      public DynamicTableSink createDynamicTableSink(Context context) {
          ReadableConfig config = getReadableConfig(context);
          String masterAddresses = config.get(KUDU_MASTERS);
          String tableName = config.get(KUDU_TABLE);
          Optional<Long> operationTimeout = config.getOptional(KUDU_OPERATION_TIMEOUT);
          Optional<Integer> flushInterval = config.getOptional(KUDU_FLUSH_INTERVAL);
          Optional<Integer> bufferSize = config.getOptional(KUDU_MAX_BUFFER_SIZE);
          Optional<Boolean> ignoreNotFound = config.getOptional(KUDU_IGNORE_NOT_FOUND);
          Optional<Boolean> ignoreDuplicate = config.getOptional(KUDU_IGNORE_DUPLICATE);
          TableSchema schema = context.getCatalogTable().getSchema();
          TableSchema physicalSchema = KuduTableUtils.getSchemaWithSqlTimestamp(schema);
          KuduTableInfo tableInfo = KuduTableUtils.createTableInfo(tableName, schema,
                  context.getCatalogTable().toProperties());
  
          KuduWriterConfig.Builder configBuilder = KuduWriterConfig.Builder
                  .setMasters(masterAddresses);
          operationTimeout.ifPresent(configBuilder::setOperationTimeout);
          flushInterval.ifPresent(configBuilder::setFlushInterval);
          bufferSize.ifPresent(configBuilder::setMaxBufferSize);
          ignoreNotFound.ifPresent(configBuilder::setIgnoreNotFound);
          ignoreDuplicate.ifPresent(configBuilder::setIgnoreDuplicate);
          return new KuduDynamicTableSink(configBuilder, physicalSchema, tableInfo);
      }
  
      /**
       * get readableConfig
       *
       * @param context
       * @return {@link ReadableConfig}
       */
      private ReadableConfig getReadableConfig(Context context) {
          FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
          return helper.getOptions();
      }
  
  
      @Override
      public DynamicTableSource createDynamicTableSource(Context context) {
          ReadableConfig config = getReadableConfig(context);
          String masterAddresses = config.get(KUDU_MASTERS);
  
          int scanRowSize = config.get(KUDU_SCAN_ROW_SIZE);
          long kuduCacheMaxRows = config.get(KUDU_LOOKUP_CACHE_MAX_ROWS);
          long kuduCacheTtl = config.get(KUDU_LOOKUP_CACHE_TTL);
          int kuduMaxReties = config.get(KUDU_LOOKUP_MAX_RETRIES);
  
          // build kudu lookup options
          KuduLookupOptions kuduLookupOptions = KuduLookupOptions.Builder.options().withCacheMaxSize(kuduCacheMaxRows)
                  .withCacheExpireMs(kuduCacheTtl)
                  .withMaxRetryTimes(kuduMaxReties)
                  .build();
  
          TableSchema schema = context.getCatalogTable().getSchema();
          TableSchema physicalSchema = KuduTableUtils.getSchemaWithSqlTimestamp(schema);
          KuduTableInfo tableInfo = KuduTableUtils.createTableInfo(config.get(KUDU_TABLE), schema,
                  context.getCatalogTable().toProperties());
  
          KuduReaderConfig.Builder configBuilder = KuduReaderConfig.Builder
                  .setMasters(masterAddresses)
                  .setRowLimit(scanRowSize);
          return new KuduDynamicTableSource(configBuilder, tableInfo, physicalSchema, physicalSchema.getFieldNames(),
                  kuduLookupOptions);
      }
  
      @Override
      public String factoryIdentifier() {
          return IDENTIFIER;
      }
  
      @Override
      public Set<ConfigOption<?>> requiredOptions() {
          return Sets.newHashSet(KUDU_TABLE, KUDU_MASTERS);
      }
  
      @Override
      public Set<ConfigOption<?>> optionalOptions() {
          return Sets.newHashSet(KUDU_HASH_COLS, KUDU_HASH_PARTITION_NUMS,
                  KUDU_PRIMARY_KEY_COLS, KUDU_SCAN_ROW_SIZE, KUDU_REPLICAS,
                  KUDU_MAX_BUFFER_SIZE, KUDU_MAX_BUFFER_SIZE, KUDU_OPERATION_TIMEOUT,
                  KUDU_IGNORE_NOT_FOUND, KUDU_IGNORE_DUPLICATE,
                  //lookup
                  KUDU_LOOKUP_CACHE_MAX_ROWS, KUDU_LOOKUP_CACHE_TTL, KUDU_LOOKUP_MAX_RETRIES);
      }
  }
  