[FLINK-26203] Basic table factory for Pulsar connector #56

tisonkun · 2023-07-24T06:03:16Z

This patch consists work from multiple contributors. I'll respect all their credits before merging.

Significant changes

(Please check any boxes [x] if the answer is "yes". You can first publish the PR and check them afterwards, for
convenience.)

Dependencies have been added or upgraded
Public API has been changed (Public API is any class annotated with @Public(Evolving))
Serializers have been changed
New feature has been introduced
- If yes, how is this documented? (not applicable / docs / JavaDocs / not documented)

Signed-off-by: tison <wander4096@gmail.com>

tisonkun · 2023-07-24T06:04:12Z

...onnector-pulsar/src/test/java/org/apache/flink/connector/pulsar/table/PulsarTableITCase.java

+
+    @ParameterizedTest
+    @ValueSource(strings = {JSON_FORMAT, AVRO_FORMAT, CSV_FORMAT})
+    void pulsarSourceSinkWithKeyAndPartialValue(String format) throws Exception {


This test still doesn't pass. But maybe we don't support this use case at all.

…other staff works well Signed-off-by: tison <wander4096@gmail.com>

Signed-off-by: tison <wander4096@gmail.com>

tisonkun · 2023-07-24T09:26:36Z

cc @syhily @cbornet I'll appreciate it if you can help with review and suggest on code or features.

I'm pending to add you two and @imaffe to the Co-authored-by columns.

tisonkun · 2023-07-24T10:21:27Z

flink-connector-pulsar/src/test/resources/log4j2-test.properties

@@ -18,7 +18,7 @@

 # Set root logger level to OFF to not flood build logs
 # set manually to INFO for debugging purposes
-rootLogger.level=OFF
+rootLogger.level=INFO


To-be reverted.

syhily · 2023-07-25T05:43:00Z

This is a huge PR. I may have time this weekend to glance at it.

MartijnVisser · 2023-07-27T09:15:19Z

I'll respect all their credits before merging.

@tisonkun This seems to indicate that this contains code that's been contributed by people outside of the ASF, is that true? Because we need to take copyright considerations, licenses etc into account, before that code can be merged into an ASF project.

tisonkun · 2023-07-27T09:24:18Z

outside of the ASF

@MartijnVisser those patches are licensed with ALv2 also, and with the copyright of StreamNative Inc. where we agree to contribute it to the upstream.

tisonkun · 2023-07-27T10:02:08Z

Update JIRA number, this should cover several subtasks under FLINK-26203.

MartijnVisser · 2023-07-27T14:29:49Z

Because we need to take copyright considerations, licenses etc into account, before that code can be merged into an ASF project.

@tisonkun and I talked about this offline and checked it, there's no problem to accept these contributions

Signed-off-by: tison <wander4096@gmail.com>

tisonkun · 2023-07-28T06:55:26Z

Support keybytes for now to minimize code change. Following work -

Add SQL client E2E test
Add upsert table factory
Revisit key serialization support. I know that users can have biz data in keys, but I don't know if it's common. @leonardBang @PatrickRen does Kafka supports mapping message key to Flink SQL columns?

tisonkun · 2023-07-28T06:55:57Z

This patch itself is ready for review and technically mergable. Feel free to leave a comment.

cc @nlu90.

nlu90 · 2023-08-22T16:20:22Z

...nnector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java

+     * <p>By default all {@link org.apache.flink.connector.pulsar.common.config.PulsarOptions} are
+     * included in the validateExcept() method./p>
+     */
+    public static final ConfigOption<String> ADMIN_URL =


we should look into removing the admin url

Yeah. I saw your issue here - https://issues.apache.org/jira/browse/FLINK-32938.

This is an inherited issue that we should handle it separately.

leonardBang

Thanks @tisonkun for the great work, I'm not familiar with Pulsar, I try to review this PR and left some comments.

...nnector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableFactory.java

...nnector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java

leonardBang · 2023-08-23T13:30:38Z

...nnector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java

+    private PulsarTableOptions() {}
+
+    public static final ConfigOption<List<String>> TOPICS =
+            ConfigOptions.key("topics")


we use topic and topic-pattern in kafka connector, I wonder could we unified this concept?

No. Support topic-pattern can be an advanced feature. I don't think it's a blocker of this patch.

We can file a JIRA ticket to support "topic-pattern" in the SQL connector. It's already supported by the DataStream connector.

leonardBang · 2023-08-23T13:33:17Z

...nnector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java

+            ConfigOptions.key("source.subscription-type")
+                    .enumType(SubscriptionType.class)
+                    .defaultValue(SubscriptionType.Exclusive)
+                    .withDescription(
+                            Description.builder()
+                                    .text(
+                                            "The [subscription type](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-subscriptions) that is supported by the [Pulsar DataStream source connector](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/pulsar/#pulsar-source). Currently, only %s and %s subscription types are supported.",
+                                            code("Exclusive"), code("Shared"))
+                                    .build());
+
+    /**
+     * Exactly same as {@link
+     * org.apache.flink.connector.pulsar.source.PulsarSourceOptions#PULSAR_SUBSCRIPTION_NAME}.
+     * Copied because we want to have a default value for it.
+     */
+    public static final ConfigOption<String> SOURCE_SUBSCRIPTION_NAME =
+            ConfigOptions.key("source.subscription-name")


I'm a little confused about the two configurations from users' perspective, why subscription-type has a default value and the later one does not?

They are totally different concepts. You may think of Flink deployment type can be default to application mode, but there is no reason for a default deployment name.

... and we actually fallback to a default subscription name in code:

.getOptional(SOURCE_SUBSCRIPTION_NAME) .orElse(DEFAULT_SUBSCRIPTION_NAME_PREFIX + randomAlphabetic(5)));

leonardBang · 2023-08-23T13:39:14Z

...nnector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableOptions.java

+                                            code("earliest"),
+                                            code("latest"),
+                                            code("ledgerId:entryId:partitionId"),


The earliest and latest worth a public configuration for users from my side, these startup modes are similar to specific-offset in kafka connector. I don't think the configuration in Kafka connector is good one, I just want to align them for decreasing the users' understanding cost.

I don't understand your suggestion. Earliest and lastest can be regarded as specific positions that the Pulsar broker can understand. They are barely positions that don't need a separate config option.

...pulsar/src/main/java/org/apache/flink/connector/pulsar/table/PulsarTableValidationUtils.java

...in/java/org/apache/flink/connector/pulsar/table/source/PulsarTableDeserializationSchema.java

pom.xml

...ector-pulsar/src/main/java/org/apache/flink/connector/pulsar/table/sink/PulsarTableSink.java

Signed-off-by: tison <wander4096@gmail.com>

tisonkun · 2023-08-24T13:49:49Z

@leonardBang All comments addressed. Also pushed a commit for including upsert mode code. Please give another look.

Signed-off-by: tison <wander4096@gmail.com>

leonardBang

Thanks @tisonkun for the update and great work, LGTM.
Btw, the task to for sql jar and task for docs can start as well next.

Signed-off-by: tison <wander4096@gmail.com> Co-authored-by: Christophe Bornet <cbornet@hotmail.com> Co-authored-by: Yufei Zhang <affeisme@gmail.com> Co-authored-by: Yufan Sheng <yufan@streamnative.io>

tisonkun · 2023-08-31T02:54:19Z

@leonardBang Thank you! Merged.

sql jar and task for docs can start as well next

Yep. Let me create the related JIRA tickets..

tisonkun added 5 commits July 24, 2023 12:10

[FLINK-XXXXX] Basic table factory for Pulsar connector

1cb15ea

Signed-off-by: tison <wander4096@gmail.com>

Add and test options and utils

2de2ecc

Signed-off-by: tison <wander4096@gmail.com>

Add table source sink

93af846

Signed-off-by: tison <wander4096@gmail.com>

Add and test table factory

87155fe

Signed-off-by: tison <wander4096@gmail.com>

try avoid keybytes issue

26eced3

Signed-off-by: tison <wander4096@gmail.com>

boring-cyborg bot added the component=Connectors/Pulsar label Jul 24, 2023

tisonkun commented Jul 24, 2023

View reviewed changes

tisonkun added 2 commits July 24, 2023 14:58

temporarily disable pulsarSourceSinkWithKeyAndPartialValue to verify …

f01f817

…other staff works well Signed-off-by: tison <wander4096@gmail.com>

fix dependency convergence

61f1b0d

Signed-off-by: tison <wander4096@gmail.com>

tisonkun requested review from MartijnVisser, leonardBang and reswqa July 24, 2023 10:20

tisonkun commented Jul 24, 2023

View reviewed changes

tisonkun changed the title ~~[FLINK-XXXXX] Basic table factory for Pulsar connector~~ [FLINK-26203] Basic table factory for Pulsar connector Jul 27, 2023

fix ser key

3fd4bf0

Signed-off-by: tison <wander4096@gmail.com>

tisonkun marked this pull request as ready for review July 28, 2023 06:52

nlu90 approved these changes Aug 22, 2023

View reviewed changes

leonardBang reviewed Aug 23, 2023

View reviewed changes

Address comments

c6b19db

Signed-off-by: tison <wander4096@gmail.com>

tisonkun force-pushed the table-connector branch from d189462 to c6b19db Compare August 24, 2023 12:22

impl upsert mode

d62907c

Signed-off-by: tison <wander4096@gmail.com>

tisonkun force-pushed the table-connector branch from d2468c5 to d62907c Compare August 24, 2023 14:00

tisonkun requested a review from leonardBang August 24, 2023 14:00

leonardBang approved these changes Aug 30, 2023

View reviewed changes

tisonkun merged commit 75c79ad into apache:main Aug 31, 2023
3 checks passed

tisonkun deleted the table-connector branch August 31, 2023 02:50

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-26203] Basic table factory for Pulsar connector #56

[FLINK-26203] Basic table factory for Pulsar connector #56

tisonkun commented Jul 24, 2023

tisonkun Jul 24, 2023

tisonkun commented Jul 24, 2023

tisonkun Jul 24, 2023

syhily commented Jul 25, 2023

MartijnVisser commented Jul 27, 2023

tisonkun commented Jul 27, 2023 •

edited

tisonkun commented Jul 27, 2023

MartijnVisser commented Jul 27, 2023

tisonkun commented Jul 28, 2023

tisonkun commented Jul 28, 2023

nlu90 Aug 22, 2023

tisonkun Aug 23, 2023

leonardBang left a comment

leonardBang Aug 23, 2023

tisonkun Aug 24, 2023

tisonkun Aug 24, 2023

leonardBang Aug 23, 2023

tisonkun Aug 24, 2023

tisonkun Aug 24, 2023

leonardBang Aug 23, 2023

tisonkun Aug 24, 2023

tisonkun commented Aug 24, 2023

leonardBang left a comment

tisonkun commented Aug 31, 2023

[FLINK-26203] Basic table factory for Pulsar connector #56

[FLINK-26203] Basic table factory for Pulsar connector #56

Conversation

tisonkun commented Jul 24, 2023

Significant changes

Choose a reason for hiding this comment

tisonkun commented Jul 24, 2023

Choose a reason for hiding this comment

syhily commented Jul 25, 2023

MartijnVisser commented Jul 27, 2023

tisonkun commented Jul 27, 2023 • edited

tisonkun commented Jul 27, 2023

MartijnVisser commented Jul 27, 2023

tisonkun commented Jul 28, 2023

tisonkun commented Jul 28, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

leonardBang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tisonkun commented Aug 24, 2023

leonardBang left a comment

Choose a reason for hiding this comment

tisonkun commented Aug 31, 2023

tisonkun commented Jul 27, 2023 •

edited