[FLINK-13340] Add more Kafka topic option of flink-connector-kafka

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Duplicate
• Affects Version/s: 1.8.1
• Component/s: Connectors / Kafka, Table SQL / API
• Labels: features pull-request-available
• Assignee: DuBin
• Reporter: DuBin
• Created: 20/Jul/19 01:53
• Updated: 19/Feb/21 07:25
• Resolved: 23/Sep/20 02:54

Description

Currently, only 'topic' option implemented in the Kafka Connector Descriptor, we can only use it like :

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val tableEnv = StreamTableEnvironment.create(env)

    tableEnv
      .connect(
        new Kafka()
          .version("0.11")
          .topic("test-flink-1")
          .startFromEarliest()
          .property("zookeeper.connect", "localhost:2181")
          .property("bootstrap.servers", "localhost:9092"))
      .withFormat(
        new Json()
          .deriveSchema()
      )
      .withSchema(
        new Schema()
          .field("name", Types.STRING)
          .field("age", Types.STRING)
      )

but we cannot consume multiple topics or a topic regex pattern. 

Here is my thoughts:

          .topic("test-flink-1") 
          //.topics("test-flink-1,test-flink-2") or topics(List<String> topics)
          //.subscriptionPattern("test-flink-.*") or subscriptionPattern(Pattern pattern)

I already implement the code on my local env with help of the FlinkKafkaConsumer, and it works.

Issue Links

• duplicates: FLINK-13701 KafkaTableSourceSink doesn't support topic pattern
• is duplicated by: FLINK-18449 Make topic discovery and partition discovery configurable for FlinkKafkaConsumer in Table API
• links to: GitHub Pull Request #9356