解析Flink UpsertKafka源码

我们由UpsertKafkaDynamicTableFactory来观察这个connector的实现

首先我们查看UpsertKafkaDynamicTableFactory的uml类图关系

从上面我们可以看到DynamicTable一般都需要实现具体的DynamicTableSourceFactory和DynamicTableSinkFactory

笔者这次主要写的是upsert_sert所以我们这次主要看DynamicTableSinkFactory的源码。

@Override
public DynamicTableSink createDynamicTableSink(Context context) {
    // TableFactoryHelper的作用很简洁，用于帮助DynamicTableFactory进行option的格式验证
    FactoryUtil.TableFactoryHelper helper =
            FactoryUtil.createTableFactoryHelper(
                    this, autoCompleteSchemaRegistrySubject(context));

  	// 获取所有的配置选项
    final ReadableConfig tableOptions = helper.getOptions();

    // 定义所有的序列化的key，value format
    EncodingFormat<SerializationSchema<RowData>> keyEncodingFormat =
            helper.discoverEncodingFormat(SerializationFormatFactory.class, KEY_FORMAT);
    EncodingFormat<SerializationSchema<RowData>> valueEncodingFormat =
            helper.discoverEncodingFormat(SerializationFormatFactory.class, VALUE_FORMAT);

    // 校验除了kafak自身参数以外的所有参数
    helper.validateExcept(PROPERTIES_PREFIX);
  	// 这里我们先去下面，看完解析之后，我们在回到这个位置
    validateSink(
            tableOptions,
            keyEncodingFormat,
            valueEncodingFormat,
            context.getPrimaryKeyIndexes());

    Tuple2<int[], int[]> keyValueProjections =
            createKeyValueProjections(context.getCatalogTable());
    // 获取key前缀
    final String keyPrefix = tableOptions.getOptional(KEY_FIELDS_PREFIX).orElse(null);
    final Properties properties = getKafkaProperties(context.getCatalogTable().getOptions());

    Integer parallelism = tableOptions.get(SINK_PARALLELISM);
    
    //设置batch参数
    int batchSize = tableOptions.get(SINK_BUFFER_FLUSH_MAX_ROWS);
    Duration batchInterval = tableOptions.get(SINK_BUFFER_FLUSH_INTERVAL);
    SinkBufferFlushMode flushMode =
            new SinkBufferFlushMode(batchSize, batchInterval.toMillis());

    // use {@link org.apache.kafka.clients.producer.internals.DefaultPartitioner}.
    // it will use hash partition if key is set else in round-robin behaviour.
    return new KafkaDynamicSink(
            context.getPhysicalRowDataType(),
            context.getPhysicalRowDataType(),
            keyEncodingFormat,
            new EncodingFormatWrapper(valueEncodingFormat),
            keyValueProjections.f0,
            keyValueProjections.f1,
            keyPrefix,
            tableOptions.get(TOPIC).get(0),
            properties,
            null,
            DeliveryGuarantee.AT_LEAST_ONCE,
            true,
            flushMode,
            parallelism,
            tableOptions.get(TRANSACTIONAL_ID_PREFIX));
}

validateSink

private static void validateSink(
        ReadableConfig tableOptions,
        Format keyFormat,
        Format valueFormat,
        int[] primaryKeyIndexes) {
    // 校验器去校验topic
    validateTopic(tableOptions);
    // 校验对应的key value格式
    validateFormat(keyFormat, valueFormat, tableOptions);
    // upsert是必须要key的
    validatePKConstraints(primaryKeyIndexes);
  	// 是否开启buffer sink模式
    validateSinkBufferFlush(tableOptions);
}