[Improve] Use new Factory API to create KafkaSource #5635

ruanwenjun · 2023-10-16T07:25:22Z

Purpose of this pull request

KafkaSource use new Factory API to create, after this change the CatalogTable can be send to post task.

Does this PR introduce any user-facing change?

How was this patch tested?

Use e2e case to assert the catalog generate by KafkaSource

Check list

If any new Jar binary package adding in your PR, please add License Notice according
New License Guide
If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
If you are contributing the connector code, please check that the following files are updated:
1. Update change log that in connector document. For more details you can refer to connector-v2
2. Update plugin-mapping.properties and add new connector information in it
3. Update the pom file of seatunnel-dist
Update the release-note.

hailin0 · 2023-10-19T02:35:27Z

...-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSource.java

@@ -105,127 +67,36 @@ public String getPluginName() {

    @Override
    public void prepare(Config config) throws PrepareFailException {


remove this method ？

#5625

https://github.com/apache/seatunnel/pull/5647/files#diff-693a7a9c81078360435112eee7dca72a70a267a56778d1cfa44aad93033e89abR42

hailin0 · 2023-10-19T02:35:36Z

...-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSource.java

-        }
-
-        setDeserialization(config);
+        kafkaSourceConfig = new KafkaSourceConfig(ReadonlyConfig.fromConfig(config));
    }

    @Override
    public SeaTunnelRowType getProducedType() {


remove this method ？

#5625

This method can not be removed for now. I need to do some extra work on the translation layer in the next PR before I can properly remove it. For this PR, it is better to keep it.

hailin0 · 2023-10-19T02:39:51Z

seatunnel-api/src/main/java/org/apache/seatunnel/api/configuration/ReadonlyConfig.java

@@ -66,6 +73,10 @@ public <T> T get(Option<T> option) {
        return getOptional(option).orElseGet(option::defaultValue);
    }

+    public <T> T get(Option<T> option, T defaultValue) {


Why not use the following:

Define default value in Option

getOptional(Option).orElse(defaultValue)

Yes, it's better way, I removed this method.

Hisoka-X · 2023-10-19T03:21:53Z

.../src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java

+        CheckResult result = readonlyConfig.checkAllExists(TOPIC, BOOTSTRAP_SERVERS);
+        if (!result.isSuccess()) {
+            throw new KafkaConnectorException(
+                    SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
+                    String.format(
+                            "PluginName: %s, PluginType: %s, Message: %s",
+                            CONNECTOR_IDENTITY, PluginType.SOURCE, result.getMsg()));
+        }


I think checkAllExists is unnecessary. As ReadonlyConfig, if it implement TableSourceFactory, we will check whether all configs are legal. Please refer

seatunnel/seatunnel-api/src/main/java/org/apache/seatunnel/api/table/factory/FactoryUtil.java

Line 116 in aefc135

ConfigValidator.of(context.getOptions()).validate(factory.optionRule());

Hisoka-X · 2023-10-19T03:23:43Z

...v2-e2e/connector-kafka-e2e/src/test/resources/textFormatIT/kafka_source_text_to_console.conf

-        c_bytes = bytes
-        c_date = date
-        c_timestamp = timestamp
+      columns = [


Could you create new conf not replace old one? I think we should make sure both two config way worked fine for now.

Agree with you

Hisoka-X · 2023-10-21T02:11:08Z

.../src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaSourceConfig.java

+                    Collections.emptyMap(),
+                    Collections.emptyList(),


It should be fine after #5681 merged.

hailin0

LGTM

ruanwenjun changed the title ~~KafkaSource use Factory to create source~~ Use new Factory API to create KafkaSource Oct 16, 2023

ruanwenjun force-pushed the dev_wenjun_KafakSourceUseFactory branch 2 times, most recently from 9fbc1d0 to 5b12a51 Compare October 16, 2023 09:42

ruanwenjun added the improve label Oct 16, 2023

ruanwenjun force-pushed the dev_wenjun_KafakSourceUseFactory branch from 5b12a51 to 990ae3b Compare October 16, 2023 11:17

ruanwenjun changed the title ~~Use new Factory API to create KafkaSource~~ [Improve] Use new Factory API to create KafkaSource Oct 16, 2023

ruanwenjun requested review from hailin0 and Hisoka-X October 17, 2023 12:08

hailin0 reviewed Oct 19, 2023

View reviewed changes

ruanwenjun force-pushed the dev_wenjun_KafakSourceUseFactory branch from 990ae3b to 9b4b165 Compare October 19, 2023 03:01

Hisoka-X reviewed Oct 19, 2023

View reviewed changes

ruanwenjun force-pushed the dev_wenjun_KafakSourceUseFactory branch 2 times, most recently from 410dd3c to 5d7344c Compare October 19, 2023 13:21

KafkaSource use Factory to create source

3834b8a

ruanwenjun force-pushed the dev_wenjun_KafakSourceUseFactory branch from 5d7344c to 3834b8a Compare October 20, 2023 05:38

Hisoka-X approved these changes Oct 21, 2023

View reviewed changes

github-actions bot added approved reviewed labels Oct 21, 2023

hailin0 approved these changes Oct 21, 2023

View reviewed changes

ruanwenjun merged commit 1c6176e into apache:dev Oct 21, 2023

ruanwenjun deleted the dev_wenjun_KafakSourceUseFactory branch October 21, 2023 02:22

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Improve] Use new Factory API to create KafkaSource #5635

[Improve] Use new Factory API to create KafkaSource #5635

ruanwenjun commented Oct 16, 2023

hailin0 Oct 19, 2023

hailin0 Oct 19, 2023

hailin0 Oct 19, 2023

Hisoka-X Oct 19, 2023

hailin0 Oct 19, 2023

ruanwenjun Oct 19, 2023

Hisoka-X Oct 19, 2023

Hisoka-X Oct 19, 2023

hailin0 Oct 19, 2023

ruanwenjun Oct 19, 2023

Hisoka-X Oct 21, 2023 •

edited

Loading

hailin0 left a comment

		@@ -105,127 +67,36 @@ public String getPluginName() {

		@Override
		public void prepare(Config config) throws PrepareFailException {

[Improve] Use new Factory API to create KafkaSource #5635

[Improve] Use new Factory API to create KafkaSource #5635

Conversation

ruanwenjun commented Oct 16, 2023

Purpose of this pull request

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

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

Choose a reason for hiding this comment

Hisoka-X Oct 21, 2023 • edited Loading

Choose a reason for hiding this comment

hailin0 left a comment

Choose a reason for hiding this comment

Hisoka-X Oct 21, 2023 •

edited

Loading