[jira] [Commented] (FLINK-13340) Add more Kafka topic option of flink-connector-kafka
[
https://issues.apache.org/jira/browse/FLINK-13340?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16903496#comment-16903496
]
DuBin commented on FLINK-13340:
---
hi [~twalthr], can you please help review this PR?
> 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, pull-request-available
> Original Estimate: 48h
> Time Spent: 10m
> Remaining Estimate: 47h 50m
>
> 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
> 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)
[jira] [Commented] (FLINK-13340) Add more Kafka topic option of flink-connector-kafka
[
https://issues.apache.org/jira/browse/FLINK-13340?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16897160#comment-16897160
]
Timo Walther commented on FLINK-13340:
--
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
>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
> 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)
[jira] [Commented] (FLINK-13340) Add more Kafka topic option of flink-connector-kafka
[
https://issues.apache.org/jira/browse/FLINK-13340?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16897155#comment-16897155
]
Robert Metzger commented on FLINK-13340:
Thanks a lot for following the process! I hope a committer working on that part
of Flink will soon take a look!
> 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
>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
> 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)
[jira] [Commented] (FLINK-13340) Add more Kafka topic option of flink-connector-kafka
[
https://issues.apache.org/jira/browse/FLINK-13340?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16889283#comment-16889283
]
DuBin commented on FLINK-13340:
---
BTW, I follow the new 'Code Contribution Process', the issue need a committer
to review. If the review passed, can you committer assign this issue to me?
Thanks!
> 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
>Affects Versions: 1.8.1
>Reporter: 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
> 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)
