devbox
算法编织者2
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Flink kafka source定义并行度_the kafka table with 'json' format doesn't support

作者：算法编织者2 | 2024-01-29 18:08:27

赞

踩

the kafka table with 'json' format doesn't support defining
  1. /*
  2.  * Licensed to the Apache Software Foundation (ASF) under one
  3.  * or more contributor license agreements.  See the NOTICE file
  4.  * distributed with this work for additional information
  5.  * regarding copyright ownership.  The ASF licenses this file
  6.  * to you under the Apache License, Version 2.0 (the
  7.  * "License"); you may not use this file except in compliance
  8.  * with the License.  You may obtain a copy of the License at
  9.  *
  10.  *     http://www.apache.org/licenses/LICENSE-2.0
  11.  *
  12.  * Unless required by applicable law or agreed to in writing, software
  13.  * distributed under the License is distributed on an "AS IS" BASIS,
  14.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15.  * See the License for the specific language governing permissions and
  16.  * limitations under the License.
  17.  */
  18.  
  19. package org.apache.flink.streaming.connectors.kafka.table;
  20.  
  21. import org.apache.flink.annotation.Internal;
  22. import org.apache.flink.api.common.serialization.DeserializationSchema;
  23. import org.apache.flink.api.common.serialization.SerializationSchema;
  24. import org.apache.flink.configuration.ConfigOption;
  25. import org.apache.flink.configuration.ConfigOptions;
  26. import org.apache.flink.configuration.Configuration;
  27. import org.apache.flink.configuration.ReadableConfig;
  28. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase;
  29. import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
  30. import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
  31. import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
  32. import org.apache.flink.table.api.ValidationException;
  33. import org.apache.flink.table.catalog.CatalogTable;
  34. import org.apache.flink.table.catalog.ObjectIdentifier;
  35. import org.apache.flink.table.connector.format.DecodingFormat;
  36. import org.apache.flink.table.connector.format.EncodingFormat;
  37. import org.apache.flink.table.connector.format.Format;
  38. import org.apache.flink.table.connector.sink.DynamicTableSink;
  39. import org.apache.flink.table.connector.source.DynamicTableSource;
  40. import org.apache.flink.table.data.RowData;
  41. import org.apache.flink.table.factories.DeserializationFormatFactory;
  42. import org.apache.flink.table.factories.DynamicTableSinkFactory;
  43. import org.apache.flink.table.factories.DynamicTableSourceFactory;
  44. import org.apache.flink.table.factories.FactoryUtil;
  45. import org.apache.flink.table.factories.FactoryUtil.TableFactoryHelper;
  46. import org.apache.flink.table.factories.SerializationFormatFactory;
  47. import org.apache.flink.table.types.DataType;
  48. import org.apache.flink.types.RowKind;
  49.  
  50. import javax.annotation.Nullable;
  51.  
  52. import java.time.Duration;
  53. import java.util.HashSet;
  54. import java.util.List;
  55. import java.util.Map;
  56. import java.util.Optional;
  57. import java.util.Properties;
  58. import java.util.Set;
  59. import java.util.regex.Pattern;
  60.  
  61. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.KEY_FIELDS;
  62. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.KEY_FIELDS_PREFIX;
  63. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.KEY_FORMAT;
  64. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.PROPERTIES_PREFIX;
  65. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.PROPS_BOOTSTRAP_SERVERS;
  66. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.PROPS_GROUP_ID;
  67. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.SCAN_STARTUP_MODE;
  68. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.SCAN_STARTUP_SPECIFIC_OFFSETS;
  69. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.SCAN_STARTUP_TIMESTAMP_MILLIS;
  70. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.SCAN_TOPIC_PARTITION_DISCOVERY;
  71. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.SINK_PARTITIONER;
  72. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.SINK_SEMANTIC;
  73. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.StartupOptions;
  74. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.TOPIC;
  75. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.TOPIC_PATTERN;
  76. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.VALUE_FIELDS_INCLUDE;
  77. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.VALUE_FORMAT;
  78. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.createKeyFormatProjection;
  79. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.createValueFormatProjection;
  80. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.getFlinkKafkaPartitioner;
  81. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.getKafkaProperties;
  82. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.getSinkSemantic;
  83. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.getStartupOptions;
  84. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.validateTableSinkOptions;
  85. import static org.apache.flink.streaming.connectors.kafka.table.KafkaOptions.validateTableSourceOptions;
  86. import static org.apache.flink.table.factories.FactoryUtil.SINK_PARALLELISM;
  87.  
  88.  
  89. @Internal
  90. public class KafkaDynamicTableFactory
  91.         implements DynamicTableSourceFactory, DynamicTableSinkFactory {
  92.     public static final ConfigOption<Integer> SOURCE_PARALLELISM =
  93.             ConfigOptions.key("source.parallelism")
  94.                     .intType()
  95.                     .noDefaultValue()
  96.                     .withDescription(
  97.                             "Defines a custom parallelism for the source. "
  98.                                     + "By default, if this option is not defined, the planner will derive the parallelism "
  99.                                     + "for each statement individually by also considering the global configuration.");
  100.  
  101.     public static final String IDENTIFIER = "kafka";
  102.  
  103.     @Override
  104.     public String factoryIdentifier() {
  105.         return IDENTIFIER;
  106.     }
  107.  
  108.     @Override
  109.     public Set<ConfigOption<?>> requiredOptions() {
  110.         final Set<ConfigOption<?>> options = new HashSet<>();
  111.         options.add(PROPS_BOOTSTRAP_SERVERS);
  112.         return options;
  113.     }
  114.  
  115.     @Override
  116.     public Set<ConfigOption<?>> optionalOptions() {
  117.         final Set<ConfigOption<?>> options = new HashSet<>();
  118.         options.add(FactoryUtil.FORMAT);
  119.         options.add(KEY_FORMAT);
  120.         options.add(KEY_FIELDS);
  121.         options.add(KEY_FIELDS_PREFIX);
  122.         options.add(VALUE_FORMAT);
  123.         options.add(VALUE_FIELDS_INCLUDE);
  124.         options.add(TOPIC);
  125.         options.add(TOPIC_PATTERN);
  126.         options.add(PROPS_GROUP_ID);
  127.         options.add(SCAN_STARTUP_MODE);
  128.         options.add(SCAN_STARTUP_SPECIFIC_OFFSETS);
  129.         options.add(SCAN_TOPIC_PARTITION_DISCOVERY);
  130.         options.add(SCAN_STARTUP_TIMESTAMP_MILLIS);
  131.         options.add(SINK_PARTITIONER);
  132.         options.add(SINK_SEMANTIC);
  133.         options.add(SINK_PARALLELISM);
  134.         options.add(SOURCE_PARALLELISM);
  135.         return options;
  136.     }
  137.  
  138.     @Override
  139.     public DynamicTableSource createDynamicTableSource(Context context) {
  140.         final TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
  141.  
  142.         final ReadableConfig tableOptions = helper.getOptions();
  143.  
  144.         final Optional<DecodingFormat<DeserializationSchema<RowData>>> keyDecodingFormat =
  145.                 getKeyDecodingFormat(helper);
  146.  
  147.         final DecodingFormat<DeserializationSchema<RowData>> valueDecodingFormat =
  148.                 getValueDecodingFormat(helper);
  149.  
  150.         helper.validateExcept(PROPERTIES_PREFIX);
  151.  
  152.         validateTableSourceOptions(tableOptions);
  153.  
  154.         validatePKConstraints(
  155.                 context.getObjectIdentifier(), context.getCatalogTable(), valueDecodingFormat);
  156.  
  157.         final StartupOptions startupOptions = getStartupOptions(tableOptions);
  158.  
  159.         final Properties properties = getKafkaProperties(context.getCatalogTable().getOptions());
  160.  
  161.         // add topic-partition discovery
  162.         properties.setProperty(
  163.                 FlinkKafkaConsumerBase.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS,
  164.                 String.valueOf(
  165.                         tableOptions
  166.                                 .getOptional(SCAN_TOPIC_PARTITION_DISCOVERY)
  167.                                 .map(Duration::toMillis)
  168.                                 .orElse(FlinkKafkaConsumerBase.PARTITION_DISCOVERY_DISABLED)));
  169.  
  170.         final DataType physicalDataType =
  171.                 context.getCatalogTable().getSchema().toPhysicalRowDataType();
  172.  
  173.         final int[] keyProjection = createKeyFormatProjection(tableOptions, physicalDataType);
  174.  
  175.         final int[] valueProjection = createValueFormatProjection(tableOptions, physicalDataType);
  176.  
  177.         final String keyPrefix = tableOptions.getOptional(KEY_FIELDS_PREFIX).orElse(null);
  178.  
  179.         return createKafkaTableSource(
  180.                 physicalDataType,
  181.                 keyDecodingFormat.orElse(null),
  182.                 valueDecodingFormat,
  183.                 keyProjection,
  184.                 valueProjection,
  185.                 keyPrefix,
  186.                 KafkaOptions.getSourceTopics(tableOptions),
  187.                 KafkaOptions.getSourceTopicPattern(tableOptions),
  188.                 properties,
  189.                 startupOptions.startupMode,
  190.                 startupOptions.specificOffsets,
  191.                 startupOptions.startupTimestampMillis);
  192.     }
  193.  
  194.     @Override
  195.     public DynamicTableSink createDynamicTableSink(Context context) {
  196.         final TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
  197.  
  198.         final ReadableConfig tableOptions = helper.getOptions();
  199.  
  200.         final Optional<EncodingFormat<SerializationSchema<RowData>>> keyEncodingFormat =
  201.                 getKeyEncodingFormat(helper);
  202.  
  203.         final EncodingFormat<SerializationSchema<RowData>> valueEncodingFormat =
  204.                 getValueEncodingFormat(helper);
  205.  
  206.         helper.validateExcept(PROPERTIES_PREFIX);
  207.  
  208.         validateTableSinkOptions(tableOptions);
  209.  
  210.         validatePKConstraints(
  211.                 context.getObjectIdentifier(), context.getCatalogTable(), valueEncodingFormat);
  212.  
  213.         final DataType physicalDataType =
  214.                 context.getCatalogTable().getSchema().toPhysicalRowDataType();
  215.  
  216.         final int[] keyProjection = createKeyFormatProjection(tableOptions, physicalDataType);
  217.  
  218.         final int[] valueProjection = createValueFormatProjection(tableOptions, physicalDataType);
  219.  
  220.         final String keyPrefix = tableOptions.getOptional(KEY_FIELDS_PREFIX).orElse(null);
  221.  
  222.         final Integer parallelism = tableOptions.getOptional(SINK_PARALLELISM).orElse(null);
  223.  
  224.         return createKafkaTableSink(
  225.                 physicalDataType,
  226.                 keyEncodingFormat.orElse(null),
  227.                 valueEncodingFormat,
  228.                 keyProjection,
  229.                 valueProjection,
  230.                 keyPrefix,
  231.                 tableOptions.get(TOPIC).get(0),
  232.                 getKafkaProperties(context.getCatalogTable().getOptions()),
  233.                 getFlinkKafkaPartitioner(tableOptions, context.getClassLoader()).orElse(null),
  234.                 getSinkSemantic(tableOptions),
  235.                 parallelism);
  236.     }
  237.  
  238.     // --------------------------------------------------------------------------------------------
  239.  
  240.     private static Optional<DecodingFormat<DeserializationSchema<RowData>>> getKeyDecodingFormat(
  241.             TableFactoryHelper helper) {
  242.         final Optional<DecodingFormat<DeserializationSchema<RowData>>> keyDecodingFormat =
  243.                 helper.discoverOptionalDecodingFormat(
  244.                         DeserializationFormatFactory.class, KEY_FORMAT);
  245.         keyDecodingFormat.ifPresent(
  246.                 format -> {
  247.                     if (!format.getChangelogMode().containsOnly(RowKind.INSERT)) {
  248.                         throw new ValidationException(
  249.                                 String.format(
  250.                                         "A key format should only deal with INSERT-only records. "
  251.                                                 + "But %s has a changelog mode of %s.",
  252.                                         helper.getOptions().get(KEY_FORMAT),
  253.                                         format.getChangelogMode()));
  254.                     }
  255.                 });
  256.         return keyDecodingFormat;
  257.     }
  258.  
  259.     private static Optional<EncodingFormat<SerializationSchema<RowData>>> getKeyEncodingFormat(
  260.             TableFactoryHelper helper) {
  261.         final Optional<EncodingFormat<SerializationSchema<RowData>>> keyEncodingFormat =
  262.                 helper.discoverOptionalEncodingFormat(SerializationFormatFactory.class, KEY_FORMAT);
  263.         keyEncodingFormat.ifPresent(
  264.                 format -> {
  265.                     if (!format.getChangelogMode().containsOnly(RowKind.INSERT)) {
  266.                         throw new ValidationException(
  267.                                 String.format(
  268.                                         "A key format should only deal with INSERT-only records. "
  269.                                                 + "But %s has a changelog mode of %s.",
  270.                                         helper.getOptions().get(KEY_FORMAT),
  271.                                         format.getChangelogMode()));
  272.                     }
  273.                 });
  274.         return keyEncodingFormat;
  275.     }
  276.  
  277.     private static DecodingFormat<DeserializationSchema<RowData>> getValueDecodingFormat(
  278.             TableFactoryHelper helper) {
  279.         return helper.discoverOptionalDecodingFormat(
  280.                 DeserializationFormatFactory.class, FactoryUtil.FORMAT)
  281.                 .orElseGet(
  282.                         () ->
  283.                                 helper.discoverDecodingFormat(
  284.                                         DeserializationFormatFactory.class, VALUE_FORMAT));
  285.     }
  286.  
  287.     private static EncodingFormat<SerializationSchema<RowData>> getValueEncodingFormat(
  288.             TableFactoryHelper helper) {
  289.         return helper.discoverOptionalEncodingFormat(
  290.                 SerializationFormatFactory.class, FactoryUtil.FORMAT)
  291.                 .orElseGet(
  292.                         () ->
  293.                                 helper.discoverEncodingFormat(
  294.                                         SerializationFormatFactory.class, VALUE_FORMAT));
  295.     }
  296.  
  297.     private static void validatePKConstraints(
  298.             ObjectIdentifier tableName, CatalogTable catalogTable, Format format) {
  299.         if (catalogTable.getSchema().getPrimaryKey().isPresent()
  300.                 && format.getChangelogMode().containsOnly(RowKind.INSERT)) {
  301.             Configuration options = Configuration.fromMap(catalogTable.getOptions());
  302.             String formatName =
  303.                     options.getOptional(FactoryUtil.FORMAT).orElse(options.get(VALUE_FORMAT));
  304.             throw new ValidationException(
  305.                     String.format(
  306.                             "The Kafka table '%s' with '%s' format doesn't support defining PRIMARY KEY constraint"
  307.                                     + " on the table, because it can't guarantee the semantic of primary key.",
  308.                             tableName.asSummaryString(), formatName));
  309.         }
  310.     }
  311.  
  312.     // --------------------------------------------------------------------------------------------
  313.  
  314.     protected KafkaDynamicSource createKafkaTableSource(
  315.             DataType physicalDataType,
  316.             @Nullable DecodingFormat<DeserializationSchema<RowData>> keyDecodingFormat,
  317.             DecodingFormat<DeserializationSchema<RowData>> valueDecodingFormat,
  318.             int[] keyProjection,
  319.             int[] valueProjection,
  320.             @Nullable String keyPrefix,
  321.             @Nullable List<String> topics,
  322.             @Nullable Pattern topicPattern,
  323.             Properties properties,
  324.             StartupMode startupMode,
  325.             Map<KafkaTopicPartition, Long> specificStartupOffsets,
  326.             long startupTimestampMillis) {
  327.         return new KafkaDynamicSource(
  328.                 physicalDataType,
  329.                 keyDecodingFormat,
  330.                 valueDecodingFormat,
  331.                 keyProjection,
  332.                 valueProjection,
  333.                 keyPrefix,
  334.                 topics,
  335.                 topicPattern,
  336.                 properties,
  337.                 startupMode,
  338.                 specificStartupOffsets,
  339.                 startupTimestampMillis,
  340.                 false);
  341.     }
  342.  
  343.     protected KafkaDynamicSink createKafkaTableSink(
  344.             DataType physicalDataType,
  345.             @Nullable EncodingFormat<SerializationSchema<RowData>> keyEncodingFormat,
  346.             EncodingFormat<SerializationSchema<RowData>> valueEncodingFormat,
  347.             int[] keyProjection,
  348.             int[] valueProjection,
  349.             @Nullable String keyPrefix,
  350.             String topic,
  351.             Properties properties,
  352.             FlinkKafkaPartitioner<RowData> partitioner,
  353.             KafkaSinkSemantic semantic,
  354.             Integer parallelism) {
  355.         return new KafkaDynamicSink(
  356.                 physicalDataType,
  357.                 keyEncodingFormat,
  358.                 valueEncodingFormat,
  359.                 keyProjection,
  360.                 valueProjection,
  361.                 keyPrefix,
  362.                 topic,
  363.                 properties,
  364.                 partitioner,
  365.                 semantic,
  366.                 false,
  367.                 parallelism);
  368.     }
  369. }
  370.  
  371. //
  372. /*
  373.  * Licensed to the Apache Software Foundation (ASF) under one
  374.  * or more contributor license agreements.  See the NOTICE file
  375.  * distributed with this work for additional information
  376.  * regarding copyright ownership.  The ASF licenses this file
  377.  * to you under the Apache License, Version 2.0 (the
  378.  * "License"); you may not use this file except in compliance
  379.  * with the License.  You may obtain a copy of the License at
  380.  *
  381.  *     http://www.apache.org/licenses/LICENSE-2.0
  382.  *
  383.  * Unless required by applicable law or agreed to in writing, software
  384.  * distributed under the License is distributed on an "AS IS" BASIS,
  385.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  386.  * See the License for the specific language governing permissions and
  387.  * limitations under the License.
  388.  */
  389.  
  390. package org.apache.flink.table.planner.plan.nodes.common
  391.  
  392. import org.apache.flink.api.common.eventtime.WatermarkStrategy
  393. import org.apache.flink.api.common.io.InputFormat
  394. import org.apache.flink.api.dag.Transformation
  395. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
  396. import org.apache.flink.table.connector.source.{DataStreamScanProvider, InputFormatProvider, ScanTableSource, SourceFunctionProvider, SourceProvider}
  397. import org.apache.flink.table.data.RowData
  398. import org.apache.flink.table.planner.calcite.FlinkTypeFactory
  399. import org.apache.flink.table.planner.plan.schema.TableSourceTable
  400. import org.apache.flink.table.runtime.connector.source.ScanRuntimeProviderContext
  401. import org.apache.flink.table.runtime.typeutils.InternalTypeInfo
  402.  
  403. import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
  404. import org.apache.calcite.rel.RelWriter
  405. import org.apache.calcite.rel.`type`.RelDataType
  406. import org.apache.calcite.rel.core.TableScan
  407.  
  408. import scala.collection.JavaConverters._
  409.  
  410. /**
  411.   * Base physical RelNode to read data from an external source defined by a [[ScanTableSource]].
  412.   */
  413. abstract class CommonPhysicalTableSourceScan(
  414.                                               cluster: RelOptCluster,
  415.                                               traitSet: RelTraitSet,
  416.                                               relOptTable: TableSourceTable)
  417.   extends TableScan(cluster, traitSet, relOptTable) {
  418.  
  419.   // cache table source transformation.
  420.   protected var sourceTransform: Transformation[_] = _
  421.  
  422.   protected val tableSourceTable: TableSourceTable = relOptTable.unwrap(classOf[TableSourceTable])
  423.  
  424.   protected[flink] val tableSource: ScanTableSource =
  425.     tableSourceTable.tableSource.asInstanceOf[ScanTableSource]
  426.  
  427.   override def deriveRowType(): RelDataType = {
  428.     // TableScan row type should always keep same with its
  429.     // interval RelOptTable's row type.
  430.     relOptTable.getRowType
  431.   }
  432.  
  433.   override def explainTerms(pw: RelWriter): RelWriter = {
  434.     super.explainTerms(pw)
  435.       .item("fields", getRowType.getFieldNames.asScala.mkString(", "))
  436.   }
  437.  
  438.   protected def createSourceTransformation(
  439.                                             env: StreamExecutionEnvironment,
  440.                                             name: String): Transformation[RowData] = {
  441.     val runtimeProvider = tableSource.getScanRuntimeProvider(ScanRuntimeProviderContext.INSTANCE)
  442.     val outRowType = FlinkTypeFactory.toLogicalRowType(tableSourceTable.getRowType)
  443.     val outTypeInfo = InternalTypeInfo.of(outRowType)
  444.  
  445.     val transformation = runtimeProvider match {
  446.       case provider: SourceFunctionProvider =>
  447.         val sourceFunction = provider.createSourceFunction()
  448.         env
  449.           .addSource(sourceFunction, name, outTypeInfo)
  450.           .getTransformation
  451.       case provider: InputFormatProvider =>
  452.         val inputFormat = provider.createInputFormat()
  453.         createInputFormatTransformation(env, inputFormat, name, outTypeInfo)
  454.       case provider: SourceProvider =>
  455.         // TODO: Push down watermark strategy to source scan
  456.         val strategy: WatermarkStrategy[RowData] = WatermarkStrategy.noWatermarks()
  457.         env.fromSource(provider.createSource(), strategy, name).getTransformation
  458.       case provider: DataStreamScanProvider =>
  459.         provider.produceDataStream(env).getTransformation
  460.     }
  461.     val parallelism = tableSourceTable.catalogTable.toProperties.get("source.parallelism")
  462.     if(parallelism != null){
  463.       transformation.setParallelism(parallelism.toInt)
  464.     }
  465.     transformation
  466.   }
  467.  
  468.   /**
  469.     * Creates a [[Transformation]] based on the given [[InputFormat]].
  470.     * The implementation is different for streaming mode and batch mode.
  471.     */
  472.   protected def createInputFormatTransformation(
  473.                                                  env: StreamExecutionEnvironment,
  474.                                                  inputFormat: InputFormat[RowData, _],
  475.                                                  name: String,
  476.                                                  outTypeInfo: InternalTypeInfo[RowData]): Transformation[RowData]
  477. }

需要覆盖的类有2个：KafkaDynamicTableFactory，CommonPhysicalTableSourceScan

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/article/detail/44088
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号