[cdc-pipeline-connector][starrocks] Introduce starrocks cdc pipeline DataSink #2765

banmoy · 2023-11-28T02:04:07Z

This close #2645. Add Implementation of DataSink for StarRocks which supports

StarRocks primary key table + at-least-once for idempotent writing
schema changes including CreateTableEvent, AddColumnEvent and DropColumnEvent. Other types of schema change will be supported in the following pr.

lvyanquan

Thanks for contribute this connector of high code quality. Left some comments.

lvyanquan · 2023-11-28T06:55:29Z

flink-cdc-connect/flink-cdc-pipeline-connectors/pom.xml

+            <groupId>com.ververica</groupId>
+            <artifactId>flink-cdc-composer</artifactId>
+            <version>${revision}</version>
+        </dependency>


Connector would not depend on flink-cdc-composer, we can remove FactoryDiscoveryUtils from StarRocksDataSinkFactoryTest to get rid of it.

lvyanquan · 2023-11-28T06:57:40Z

...src/test/java/com/ververical/cdc/connectors/starrocks/sink/StarRocksDataSinkFactoryTest.java

+                                .put("password", "")
+                                .build());
+        DataSink dataSink =
+                sinkFactory.createDataSink(


Met NullPointerException when running this test.

lvyanquan · 2023-11-28T07:34:02Z

...rc/main/java/com/ververica/cdc/connectors/starrocks/sink/EventRecordSerializationSchema.java

+    public StarRocksRowData serialize(Event record) {
+        if (record instanceof SchemaChangeEvent) {
+            applySchemaChangeEvent((SchemaChangeEvent) record);
+            return null;


Just a worry, will this lead to NoPointException?

Thanks for the reminding. Connector will deal with the null value.

lvyanquan · 2023-11-28T07:50:29Z

...ctor-starrocks/src/main/java/com/ververica/cdc/connectors/starrocks/sink/StarRocksUtils.java

+                        record ->
+                                DATE_FORMATTER.format(
+                                        Date.valueOf(
+                                                LocalDate.ofEpochDay(record.getInt(fieldPos))));


Is this because of different ways of handling time that this method was rewritten instead of using createFieldGetter in RecordData？

Yes, we need to convert the data to a string so that it can be sent to starrocks in json format

lvyanquan · 2023-12-03T03:22:34Z

@banmoy Sorry for bother you. Met the follow error while running test:

Caused by: java.lang.NoClassDefFoundError: com/starrocks/shade/org/apache/commons/compress/utils/Lists
        at com.starrocks.connector.flink.catalog.StarRocksCatalog.executeSingleColumnStatement(StarRocksCatalog.java:456)
        at com.starrocks.connector.flink.catalog.StarRocksCatalog.databaseExists(StarRocksCatalog.java:107)
        at com.ververica.cdc.connectors.starrocks.sink.StarRocksMetadataApplier.applyCreateTable(StarRocksMetadataApplier.java:92)
        at com.ververica.cdc.connectors.starrocks.sink.StarRocksMetadataApplier.applySchemaChange(StarRocksMetadataApplier.java:71)
        at com.ververica.cdc.runtime.operators.schema.coordinator.SchemaRegistryRequestHandler.applySchemaChange(SchemaRegistryRequestHandler.java:86)
        at com.ververica.cdc.runtime.operators.schema.coordinator.SchemaRegistryRequestHandler.flushSuccess(SchemaRegistryRequestHandler.java:145)
        at com.ververica.cdc.runtime.operators.schema.coordinator.SchemaRegistry.handleEventFromOperator(SchemaRegistry.java:123)
        at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder.handleEventFromOperator(OperatorCoordinatorHolder.java:204)
        at org.apache.flink.runtime.scheduler.DefaultOperatorCoordinatorHandler.deliverOperatorEventToCoordinator(DefaultOperatorCoordinatorHandler.java:121)
        ... 33 more
Caused by: java.lang.ClassNotFoundException: com.starrocks.shade.org.apache.commons.compress.utils.Lists
        at java.base/java.net.URLClassLoader.findClass(URLClassLoader.java:476)
        at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:594)
        at org.apache.flink.util.FlinkUserCodeClassLoader.loadClassWithoutExceptionHandling(FlinkUserCodeClassLoader.java:67)
        at org.apache.flink.util.ChildFirstClassLoader.loadClassWithoutExceptionHandling(ChildFirstClassLoader.java:74)
        at org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:51)
        at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:527)
        ... 42 more

banmoy · 2023-12-04T02:16:10Z

@banmoy Sorry for bother you. Met the follow error while running test:

Caused by: java.lang.NoClassDefFoundError: com/starrocks/shade/org/apache/commons/compress/utils/Lists
        at com.starrocks.connector.flink.catalog.StarRocksCatalog.executeSingleColumnStatement(StarRocksCatalog.java:456)
        at com.starrocks.connector.flink.catalog.StarRocksCatalog.databaseExists(StarRocksCatalog.java:107)
        at com.ververica.cdc.connectors.starrocks.sink.StarRocksMetadataApplier.applyCreateTable(StarRocksMetadataApplier.java:92)
        at com.ververica.cdc.connectors.starrocks.sink.StarRocksMetadataApplier.applySchemaChange(StarRocksMetadataApplier.java:71)
        at com.ververica.cdc.runtime.operators.schema.coordinator.SchemaRegistryRequestHandler.applySchemaChange(SchemaRegistryRequestHandler.java:86)
        at com.ververica.cdc.runtime.operators.schema.coordinator.SchemaRegistryRequestHandler.flushSuccess(SchemaRegistryRequestHandler.java:145)
        at com.ververica.cdc.runtime.operators.schema.coordinator.SchemaRegistry.handleEventFromOperator(SchemaRegistry.java:123)
        at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder.handleEventFromOperator(OperatorCoordinatorHolder.java:204)
        at org.apache.flink.runtime.scheduler.DefaultOperatorCoordinatorHandler.deliverOperatorEventToCoordinator(DefaultOperatorCoordinatorHandler.java:121)
        ... 33 more
Caused by: java.lang.ClassNotFoundException: com.starrocks.shade.org.apache.commons.compress.utils.Lists
        at java.base/java.net.URLClassLoader.findClass(URLClassLoader.java:476)
        at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:594)
        at org.apache.flink.util.FlinkUserCodeClassLoader.loadClassWithoutExceptionHandling(FlinkUserCodeClassLoader.java:67)
        at org.apache.flink.util.ChildFirstClassLoader.loadClassWithoutExceptionHandling(ChildFirstClassLoader.java:74)
        at org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:51)
        at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:527)
        ... 42 more

@lvyanquan Thanks for the reminding. Have fixed it and the test passed.

Signed-off-by: PengFei Li <lpengfei2016@gmail.com>

lvyanquan

LGTM, +1.

leonardBang

Thanks @banmoy for the great work and @lvyanquan for the continuous review, LGTM.

This closes #2645.

leonardBang · 2023-12-05T07:20:46Z

Resolved by 33b7c85

github-actions bot added the values-pipeline-connector label Nov 28, 2023

hirolee88 mentioned this pull request Nov 28, 2023

Update mysql-postgres-tutorial.md #2764

Merged

banmoy force-pushed the starrocks_pipeline branch 2 times, most recently from 528bfda to cd19d56 Compare November 28, 2023 03:51

lvyanquan reviewed Nov 28, 2023

View reviewed changes

leonardBang assigned banmoy Nov 28, 2023

banmoy force-pushed the starrocks_pipeline branch 4 times, most recently from 1e9b23d to f4689be Compare December 2, 2023 13:23

github-actions bot added the build label Dec 2, 2023

banmoy force-pushed the starrocks_pipeline branch from 3110d27 to ca52990 Compare December 3, 2023 16:34

banmoy added 16 commits December 4, 2023 21:03

Add framework

bf5d243

Implement StarRocksMetadataApplier

3fbbfd7

Signed-off-by: PengFei Li <lpengfei2016@gmail.com>

Implement StarRocksDataSink

a64b5b2

Signed-off-by: PengFei Li <lpengfei2016@gmail.com>

Remove StarRocksCatalog

31784e6

Signed-off-by: PengFei Li <lpengfei2016@gmail.com>

Add EventRecordSerializerTest

be4456c

Add StarRocksMetadataApplierTest

b027fde

set log off

b9fc733

Add snapshot repo

0a39a7e

Rename EventRecordSerializationSchema

619962e

Fix StarRocksDataSinkFactoryTest.testCreateDataSink

a1cd241

Add context for EventRecordSerializationSchema#serialize

5b2b945

minor fix

464d4b8

Signed-off-by: PengFei Li <lpengfei2016@gmail.com>

Fix jdbc driver load

c057ae6

Force to use at-least-once

23a2a90

Rename TableConfig to TableCreateConfig

aa7e4f3

Bump connector version to release 1.2.9

916828f

banmoy added 3 commits December 4, 2023 21:03

Remove snapshot repo from pom

b834bc5

add common-compress dependency

f18bbe8

Fix compile failure

4db2da4

banmoy force-pushed the starrocks_pipeline branch from ca52990 to 4db2da4 Compare December 4, 2023 13:11

Support TIMESTAMP_WITH_LOCAL_TIME_ZONE

7688367

banmoy force-pushed the starrocks_pipeline branch from 49420ba to 7688367 Compare December 4, 2023 15:04

lvyanquan approved these changes Dec 4, 2023

View reviewed changes

leonardBang added this to the V3.0.0 milestone Dec 5, 2023

leonardBang approved these changes Dec 5, 2023

View reviewed changes

leonardBang pushed a commit that referenced this pull request Dec 5, 2023

[pipeline-connector][starrocks] Add starrocks pipeline connector (#2765)

33b7c85

This closes #2645.

leonardBang closed this Dec 5, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[cdc-pipeline-connector][starrocks] Introduce starrocks cdc pipeline DataSink #2765

[cdc-pipeline-connector][starrocks] Introduce starrocks cdc pipeline DataSink #2765

banmoy commented Nov 28, 2023 •

edited

lvyanquan left a comment

lvyanquan Nov 28, 2023

lvyanquan Nov 28, 2023

banmoy Nov 28, 2023

lvyanquan Nov 28, 2023 •

edited

banmoy Nov 28, 2023

lvyanquan Nov 28, 2023

banmoy Nov 28, 2023

lvyanquan commented Dec 3, 2023

banmoy commented Dec 4, 2023

lvyanquan left a comment •

edited

leonardBang left a comment

leonardBang commented Dec 5, 2023

[cdc-pipeline-connector][starrocks] Introduce starrocks cdc pipeline DataSink #2765

[cdc-pipeline-connector][starrocks] Introduce starrocks cdc pipeline DataSink #2765

Conversation

banmoy commented Nov 28, 2023 • edited

lvyanquan left a comment

Choose a reason for hiding this comment

lvyanquan Nov 28, 2023

Choose a reason for hiding this comment

lvyanquan Nov 28, 2023

Choose a reason for hiding this comment

banmoy Nov 28, 2023

Choose a reason for hiding this comment

lvyanquan Nov 28, 2023 • edited

Choose a reason for hiding this comment

banmoy Nov 28, 2023

Choose a reason for hiding this comment

lvyanquan Nov 28, 2023

Choose a reason for hiding this comment

banmoy Nov 28, 2023

Choose a reason for hiding this comment

lvyanquan commented Dec 3, 2023

banmoy commented Dec 4, 2023

lvyanquan left a comment • edited

Choose a reason for hiding this comment

leonardBang left a comment

Choose a reason for hiding this comment

leonardBang commented Dec 5, 2023

banmoy commented Nov 28, 2023 •

edited

lvyanquan Nov 28, 2023 •

edited

lvyanquan left a comment •

edited