[jira] [Comment Edited] (FLINK-13340) Add more Kafka topic option of flink-connector-kafka

["Timo Walther (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-13340?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16897160#comment-16897160 ] 

Timo Walther edited comment on FLINK-13340 at 7/31/19 1:08 PM:
---------------------------------------------------------------

Hi [~dubin555], your proposed changes sound reasonable to me. I will assign this issue to you. Regarding the API, I would prefer `topics(String...)` with var args. For the pattern version, I would vote for the string parameter. But we can also discuss that in the PR later.


was (Author: twalthr):
Hi [~dubin555], your proposed changes sound reasonable to me. I will assign this issue to you.

> Add more Kafka topic option of flink-connector-kafka
> ----------------------------------------------------
>
>                 Key: FLINK-13340
>                 URL: https://issues.apache.org/jira/browse/FLINK-13340
>             Project: Flink
>          Issue Type: Improvement
>          Components: Connectors / Kafka, Table SQL / API
>    Affects Versions: 1.8.1
>            Reporter: DuBin
>            Assignee: DuBin
>            Priority: Major
>              Labels: features
>   Original Estimate: 48h
>  Remaining Estimate: 48h
>
> Currently, only 'topic' option implemented in the Kafka Connector Descriptor, we can only use it like :
> {code:java}
>     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)
>       ){code}
> but we cannot consume multiple topics or a topic regex pattern. 
> Here is my thoughts:
> {code:java}
>           .topic("test-flink-1") 
>           //.topics("test-flink-1,test-flink-2") or topics(List<String> topics)
>           //.subscriptionPattern("test-flink-.*") or subscriptionPattern(Pattern pattern)
> {code}
> I already implement the code on my local env with help of the FlinkKafkaConsumer, and it works.



--
This message was sent by Atlassian JIRA
(v7.6.14#76016)