[
https://issues.apache.org/jira/browse/FLINK-8190?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16276764#comment-16276764
]
ASF GitHub Bot commented on FLINK-8190:
---------------------------------------
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/5117#discussion_r154644907
--- Diff: docs/dev/connectors/kafka.md ---
@@ -309,6 +311,57 @@ prior to Flink 1.3.x, partition discovery cannot be
enabled on the restore run.
with an exception. In this case, in order to use partition discovery,
please first take a savepoint in Flink 1.3.x and
then restore again from that.
+#### Topic discovery
+
+At a higher-level, the Flink Kafka Consumer is also capable of discovering
topics, based on pattern matching on the
+topic names using regular expressions. See the below for an example:
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+final StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+
+Properties properties = new Properties();
+properties.setProperty("bootstrap.servers", "localhost:9092");
+properties.setProperty("group.id", "test");
+
+FlinkKafkaConsumer011<String> myConsumer = new FlinkKafkaConsumer011<>(
+ java.util.regex.Pattern.compile("test-topic-[0-9]"),
+ new SimpleStringSchema(),
+ properties);
+
+DataStream<String> stream = env.addSource(myConsumer);
+...
+{% endhighlight %}
+</div>
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+val env = StreamExecutionEnvironment.getExecutionEnvironment()
+
+val properties = new Properties()
+properties.setProperty("bootstrap.servers", "localhost:9092")
+properties.setProperty("group.id", "test")
+
+val myConsumer = new FlinkKafkaConsumer08[String](
+ java.util.regex.Pattern.compile("test-topic-[0-9]"),
+ new SimpleStringSchema,
+ properties)
+
+val stream = env.addSource(myConsumer)
+...
+{% endhighlight %}
+</div>
+</div>
+
+In the above example, all topics with names that match the specified
regular expression
+(starting with `test-topic-` and ending with a single digit) will be
subscribed by the consumer
+when the job starts running.
+
+To allow the consumer to discover dynamically created topics after the job
started running,
+set a non-negative value for `flink.partition-discovery.interval-millis`.
This allows
+the the consumer to discover partitions of new topics with names that also
match the specified
--- End diff --
nit: double "the"
> Add extra FlinkKafkaConsumer constructors to expose pattern-based topic
> subscription functionality
> --------------------------------------------------------------------------------------------------
>
> Key: FLINK-8190
> URL: https://issues.apache.org/jira/browse/FLINK-8190
> Project: Flink
> Issue Type: New Feature
> Components: Kafka Connector
> Reporter: Tzu-Li (Gordon) Tai
> Assignee: Tzu-Li (Gordon) Tai
> Fix For: 1.4.0, 1.5.0
>
>
> The required internals for pattern-based topic discovery was implemented as
> part of FLINK-4022 (along with partition discovery). However, the
> functionality for pattern-based topic discovery was not yet exposed via any
> visible user API on the version-specific subclasses of
> {{FlinkKafkaConsumerBase}}.
> I propose to add two more constructors for this:
> {code}
> public FlinkKafkaConsumerXX(java.util.regex.Pattern subscriptionPattern,
> DeserializationSchema<T> schema, Properties props);
> public FlinkKafkaConsumerXX(java.util.regex.Pattern subscriptionPattern,
> KeyedDeserializationSchema<T> schema, Properties props);
> {code}
> This allows the consumer to pick up all matching topics on startup.
> To continuously pick up matching topics on the fly when they are created
> after the job has already started running, users should additionally set the
> {{KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS}} property, as they would do for
> partition discovery.
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
