You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "dawidwys (via GitHub)" <gi...@apache.org> on 2023/01/31 11:42:25 UTC
[GitHub] [flink] dawidwys opened a new pull request, #21808: [FLINK-24456][Connectors / Kafka,Table SQL / Ecosystem] Support bounded offset in the Kafka table connector
dawidwys opened a new pull request, #21808:
URL: https://github.com/apache/flink/pull/21808
## What is the purpose of the change
This is a continuation of #17994 which I took over because it has been stale for quite some time
## Verifying this change
This change added tests:
* KafkaDynamicTableFactoryTest
* KafkaTableITCase#testKafkaSourceSinkWithBoundedSpecificOffsets
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (**yes** / no)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (yes / **no** / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (**yes** / no)
- If yes, how is the feature documented? (not applicable / **docs** / **JavaDocs** / not documented)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21808: [FLINK-24456][Connectors / Kafka,Table SQL / Ecosystem] Support bounded offset in the Kafka table connector
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #21808:
URL: https://github.com/apache/flink/pull/21808#discussion_r1091895779
##########
flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java:
##########
@@ -179,6 +180,66 @@ public void testKafkaSourceSink() throws Exception {
deleteTestTopic(topic);
}
+ @Test
+ public void testKafkaSourceSinkWithBoundedSpecificOffsets() throws Exception {
+ // we always use a different topic name for each parameterized topic,
+ // in order to make sure the topic can be created.
+ final String topic = "bounded_" + format + "_" + UUID.randomUUID();
+ createTestTopic(topic, 1, 1);
+
+ // ---------- Produce an event time stream into Kafka -------------------
+ String groupId = getStandardProps().getProperty("group.id");
+ String bootstraps = getBootstrapServers();
+
+ final String createTable =
+ String.format(
+ "CREATE TABLE kafka (\n"
+ + " `user_id` INT,\n"
+ + " `item_id` INT,\n"
+ + " `behavior` STRING\n"
+ + ") WITH (\n"
+ + " 'connector' = '%s',\n"
+ + " 'topic' = '%s',\n"
+ + " 'properties.bootstrap.servers' = '%s',\n"
+ + " 'properties.group.id' = '%s',\n"
+ + " 'scan.startup.mode' = 'earliest-offset',\n"
+ + " 'scan.bounded.mode' = 'specific-offsets',\n"
+ + " 'scan.bounded.specific-offsets' = 'partition:0,offset:2',\n"
Review Comment:
From what I understand (`SpecifiedOffsetsInitializer#getPartitionOffsets`) it will still be bounded, but partitions not specified won't be consumed. The **beginning** offset for that partition should be used.
I'll extend docs on that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] MartijnVisser commented on pull request #21808: [FLINK-24456][Connectors / Kafka,Table SQL / Ecosystem] Support bounded offset in the Kafka table connector
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on PR #21808:
URL: https://github.com/apache/flink/pull/21808#issuecomment-1410641924
Don't forget Spotless ;)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on pull request #21808: [FLINK-24456][Connectors / Kafka,Table SQL / Ecosystem] Support bounded offset in the Kafka table connector
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on PR #21808:
URL: https://github.com/apache/flink/pull/21808#issuecomment-1410942280
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] twalthr commented on a diff in pull request #21808: [FLINK-24456][Connectors / Kafka,Table SQL / Ecosystem] Support bounded offset in the Kafka table connector
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #21808:
URL: https://github.com/apache/flink/pull/21808#discussion_r1091852964
##########
docs/content.zh/docs/connectors/table/kafka.md:
##########
@@ -485,6 +510,22 @@ ROW<`version` INT, `behavior` STRING>
如果使用了 `specific-offsets`，必须使用另外一个配置项 `scan.startup.specific-offsets` 来为每个 partition 指定起始偏移量，
例如，选项值 `partition:0,offset:42;partition:1,offset:300` 表示 partition `0` 从偏移量 `42` 开始，partition `1` 从偏移量 `300` 开始。
+### Bounded Ending Position
+
+The config option `scan.bounded.mode` specifies the bounded mode for Kafka consumer. The valid enumerations are:
+<ul>
+<li><span markdown="span">`group-offsets`</span>: bounded from committed offsets in ZK / Kafka brokers of a specific consumer group.</li>
+<li><span markdown="span">`latest-offset`</span>: bounded from the latest offset.</li>
+<li><span markdown="span">`timestamp`</span>: bounded from user-supplied timestamp for each partition.</li>
+<li><span markdown="span">`specific-offsets`</span>: bounded from user-supplied specific offsets for each partition.</li>
+</ul>
+
+If config option value `scan.bounded.mode` is not set the default is an unbounded data stream.
+
+If `timestamp` is specified, another config option `scan.bounded.timestamp-millis` is required to specify a specific bounded timestamp in milliseconds since January 1, 1970 00:00:00.000 GMT.
Review Comment:
> Prior to 1972, this time was called Greenwich Mean Time (GMT)
nevermind
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #21808: [FLINK-24456][Connectors / Kafka,Table SQL / Ecosystem] Support bounded offset in the Kafka table connector
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #21808:
URL: https://github.com/apache/flink/pull/21808#issuecomment-1410214989
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "57a4dc95d11fd20db3a19ecb6202c85d966b55dc",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "57a4dc95d11fd20db3a19ecb6202c85d966b55dc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 57a4dc95d11fd20db3a19ecb6202c85d966b55dc UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21808: [FLINK-24456][Connectors / Kafka,Table SQL / Ecosystem] Support bounded offset in the Kafka table connector
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #21808:
URL: https://github.com/apache/flink/pull/21808#discussion_r1091890518
##########
flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java:
##########
@@ -179,6 +180,66 @@ public void testKafkaSourceSink() throws Exception {
deleteTestTopic(topic);
}
+ @Test
+ public void testKafkaSourceSinkWithBoundedSpecificOffsets() throws Exception {
Review Comment:
Sure, I can.
So far I did not want to add too many IT tests, as this PR does not introduce a new feature, but only exposes what has been implemented already for DataStream. I assumed, but I haven't checked, this should've been covered by the DataStream tests.
I left that single test as a smoke test for the entire stack, but I can add one more, which can serve as an example.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21808: [FLINK-24456][Connectors / Kafka,Table SQL / Ecosystem] Support bounded offset in the Kafka table connector
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #21808:
URL: https://github.com/apache/flink/pull/21808#discussion_r1092006595
##########
docs/content.zh/docs/connectors/table/kafka.md:
##########
@@ -309,6 +309,31 @@ CREATE TABLE KafkaTable (
<td>Long</td>
<td>在使用 <code>'timestamp'</code> 启动模式时指定启动的时间戳（单位毫秒）。</td>
</tr>
+ <tr>
+ <td><h5>scan.bounded.mode</h5></td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">unbounded</td>
+ <td>String</td>
Review Comment:
It is enum, actually. It's just that these docs are not autogenerated and I guess it's been copied over from `scan.startup.mode` which has not been updated as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on pull request #21808: [FLINK-24456][Connectors / Kafka,Table SQL / Ecosystem] Support bounded offset in the Kafka table connector
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on PR #21808:
URL: https://github.com/apache/flink/pull/21808#issuecomment-1411180102
Passed in my private azure: https://dev.azure.com/wysakowiczdawid/Flink/_build/results?buildId=1550&view=results
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys closed pull request #21808: [FLINK-24456][Connectors / Kafka,Table SQL / Ecosystem] Support bounded offset in the Kafka table connector
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys closed pull request #21808: [FLINK-24456][Connectors / Kafka,Table SQL / Ecosystem] Support bounded offset in the Kafka table connector
URL: https://github.com/apache/flink/pull/21808
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #21808: [FLINK-24456][Connectors / Kafka,Table SQL / Ecosystem] Support bounded offset in the Kafka table connector
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #21808:
URL: https://github.com/apache/flink/pull/21808#discussion_r1092125818
##########
flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java:
##########
@@ -240,6 +246,76 @@ public void testKafkaSourceSinkWithBoundedSpecificOffsets() throws Exception {
deleteTestTopic(topic);
}
+ @Test
+ public void testKafkaSourceSinkWithBoundedTimestamp() throws Exception {
+ // we always use a different topic name for each parameterized topic,
+ // in order to make sure the topic can be created.
+ final String topic = "bounded_" + format + "_" + UUID.randomUUID();
+ createTestTopic(topic, 1, 1);
+
+ // ---------- Produce an event time stream into Kafka -------------------
+ String groupId = getStandardProps().getProperty("group.id");
+ String bootstraps = getBootstrapServers();
+
+ final String createTable =
+ String.format(
+ "CREATE TABLE kafka (\n"
+ + " `user_id` INT,\n"
+ + " `item_id` INT,\n"
+ + " `behavior` STRING,\n"
+ + " `event_time` TIMESTAMP(3) METADATA FROM 'timestamp'"
Review Comment:
Makes sense! Updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] twalthr commented on a diff in pull request #21808: [FLINK-24456][Connectors / Kafka,Table SQL / Ecosystem] Support bounded offset in the Kafka table connector
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #21808:
URL: https://github.com/apache/flink/pull/21808#discussion_r1091834613
##########
docs/content.zh/docs/connectors/table/kafka.md:
##########
@@ -309,6 +309,31 @@ CREATE TABLE KafkaTable (
<td>Long</td>
<td>在使用 <code>'timestamp'</code> 启动模式时指定启动的时间戳（单位毫秒）。</td>
</tr>
+ <tr>
+ <td><h5>scan.bounded.mode</h5></td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">unbounded</td>
+ <td>String</td>
Review Comment:
I guess this is not an enum because we still don't support `-` in names?
##########
docs/content.zh/docs/connectors/table/kafka.md:
##########
@@ -485,6 +510,22 @@ ROW<`version` INT, `behavior` STRING>
如果使用了 `specific-offsets`，必须使用另外一个配置项 `scan.startup.specific-offsets` 来为每个 partition 指定起始偏移量，
例如，选项值 `partition:0,offset:42;partition:1,offset:300` 表示 partition `0` 从偏移量 `42` 开始，partition `1` 从偏移量 `300` 开始。
+### Bounded Ending Position
+
+The config option `scan.bounded.mode` specifies the bounded mode for Kafka consumer. The valid enumerations are:
+<ul>
+<li><span markdown="span">`group-offsets`</span>: bounded from committed offsets in ZK / Kafka brokers of a specific consumer group.</li>
+<li><span markdown="span">`latest-offset`</span>: bounded from the latest offset.</li>
+<li><span markdown="span">`timestamp`</span>: bounded from user-supplied timestamp for each partition.</li>
+<li><span markdown="span">`specific-offsets`</span>: bounded from user-supplied specific offsets for each partition.</li>
+</ul>
+
+If config option value `scan.bounded.mode` is not set the default is an unbounded data stream.
Review Comment:
`unbounded data stream` -> `unbounded table`
##########
flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/config/BoundedMode.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors.kafka.config;
+
+import org.apache.flink.annotation.Internal;
+
+/** End modes for the Kafka Consumer. */
+@Internal
+public enum BoundedMode {
+
+ /** Do not end consuming */
+ UNBOUNDED,
+
+ /** End from committed offsets in ZK / Kafka brokers of a specific consumer group (default). */
+ GROUP_OFFSETS,
+
+ /** End from the latest offset. */
Review Comment:
What does that mean in reality? Is the latest offset or group offsets fetched at the beginning of the execution? Maybe we can improve the docs a bit.
##########
flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java:
##########
@@ -179,6 +180,66 @@ public void testKafkaSourceSink() throws Exception {
deleteTestTopic(topic);
}
+ @Test
+ public void testKafkaSourceSinkWithBoundedSpecificOffsets() throws Exception {
+ // we always use a different topic name for each parameterized topic,
+ // in order to make sure the topic can be created.
+ final String topic = "bounded_" + format + "_" + UUID.randomUUID();
+ createTestTopic(topic, 1, 1);
+
+ // ---------- Produce an event time stream into Kafka -------------------
+ String groupId = getStandardProps().getProperty("group.id");
+ String bootstraps = getBootstrapServers();
+
+ final String createTable =
+ String.format(
+ "CREATE TABLE kafka (\n"
+ + " `user_id` INT,\n"
+ + " `item_id` INT,\n"
+ + " `behavior` STRING\n"
+ + ") WITH (\n"
+ + " 'connector' = '%s',\n"
+ + " 'topic' = '%s',\n"
+ + " 'properties.bootstrap.servers' = '%s',\n"
+ + " 'properties.group.id' = '%s',\n"
+ + " 'scan.startup.mode' = 'earliest-offset',\n"
+ + " 'scan.bounded.mode' = 'specific-offsets',\n"
+ + " 'scan.bounded.specific-offsets' = 'partition:0,offset:2',\n"
Review Comment:
Another item for the documentation and potentially tests: What happens if I don't specify an end offset for all partitions? Will the source still be bounded?
##########
docs/content.zh/docs/connectors/table/kafka.md:
##########
@@ -485,6 +510,22 @@ ROW<`version` INT, `behavior` STRING>
如果使用了 `specific-offsets`，必须使用另外一个配置项 `scan.startup.specific-offsets` 来为每个 partition 指定起始偏移量，
例如，选项值 `partition:0,offset:42;partition:1,offset:300` 表示 partition `0` 从偏移量 `42` 开始，partition `1` 从偏移量 `300` 开始。
+### Bounded Ending Position
+
+The config option `scan.bounded.mode` specifies the bounded mode for Kafka consumer. The valid enumerations are:
+<ul>
+<li><span markdown="span">`group-offsets`</span>: bounded from committed offsets in ZK / Kafka brokers of a specific consumer group.</li>
+<li><span markdown="span">`latest-offset`</span>: bounded from the latest offset.</li>
+<li><span markdown="span">`timestamp`</span>: bounded from user-supplied timestamp for each partition.</li>
+<li><span markdown="span">`specific-offsets`</span>: bounded from user-supplied specific offsets for each partition.</li>
+</ul>
+
+If config option value `scan.bounded.mode` is not set the default is an unbounded data stream.
+
+If `timestamp` is specified, another config option `scan.bounded.timestamp-millis` is required to specify a specific bounded timestamp in milliseconds since January 1, 1970 00:00:00.000 GMT.
Review Comment:
nit: `GMT` -> `UTC`
##########
flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java:
##########
@@ -179,6 +180,66 @@ public void testKafkaSourceSink() throws Exception {
deleteTestTopic(topic);
}
+ @Test
+ public void testKafkaSourceSinkWithBoundedSpecificOffsets() throws Exception {
Review Comment:
Can we also add a test for timestamps?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] twalthr commented on a diff in pull request #21808: [FLINK-24456][Connectors / Kafka,Table SQL / Ecosystem] Support bounded offset in the Kafka table connector
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #21808:
URL: https://github.com/apache/flink/pull/21808#discussion_r1092114325
##########
flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java:
##########
@@ -240,6 +246,76 @@ public void testKafkaSourceSinkWithBoundedSpecificOffsets() throws Exception {
deleteTestTopic(topic);
}
+ @Test
+ public void testKafkaSourceSinkWithBoundedTimestamp() throws Exception {
+ // we always use a different topic name for each parameterized topic,
+ // in order to make sure the topic can be created.
+ final String topic = "bounded_" + format + "_" + UUID.randomUUID();
+ createTestTopic(topic, 1, 1);
+
+ // ---------- Produce an event time stream into Kafka -------------------
+ String groupId = getStandardProps().getProperty("group.id");
+ String bootstraps = getBootstrapServers();
+
+ final String createTable =
+ String.format(
+ "CREATE TABLE kafka (\n"
+ + " `user_id` INT,\n"
+ + " `item_id` INT,\n"
+ + " `behavior` STRING,\n"
+ + " `event_time` TIMESTAMP(3) METADATA FROM 'timestamp'"
Review Comment:
Why don't you use `TIMESTAMP_LTZ(3)` here? This would avoid a later `LocalDateTime.ofInstant(Instant.ofEpochMilli(0L), ZoneId.systemDefault())`, no?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org