Rename EventRecordSerializationSchema

Author: banmoy

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/com/ververica/cdc/connectors/starrocks/sink/EventRecordSerializer.java renamed to flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/com/ververica/cdc/connectors/starrocks/sink/EventRecordSerializationSchema.java

@@ -18,7 +18,7 @@
 
 import com.starrocks.connector.flink.table.data.DefaultStarRocksRowData;
 import com.starrocks.connector.flink.table.data.StarRocksRowData;
-import com.starrocks.connector.flink.table.sink.v2.RecordSerializer;
+import com.starrocks.connector.flink.table.sink.v2.RecordSerializationSchema;
 import com.starrocks.connector.flink.tools.JsonWrapper;
 import com.ververica.cdc.common.data.RecordData;
 import com.ververica.cdc.common.event.CreateTableEvent;
@@ -38,7 +38,7 @@
 import static com.ververica.cdc.connectors.starrocks.sink.StarRocksUtils.createFieldGetter;
 
 /** Serializer for the input {@link Event}. It will serialize a row to a json string. */
-public class EventRecordSerializer implements RecordSerializer<Event> {
+public class EventRecordSerializationSchema implements RecordSerializationSchema<Event> {
 
     private static final long serialVersionUID = 1L;
 

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/com/ververica/cdc/connectors/starrocks/sink/StarRocksDataSink.java

@@ -54,7 +54,7 @@ public StarRocksDataSink(
     @Override
     public EventSinkProvider getEventSinkProvider() {
         StarRocksSink<Event> starRocksSink =
-                SinkFunctionFactory.createSink(sinkOptions, new EventRecordSerializer());
+                SinkFunctionFactory.createSink(sinkOptions, new EventRecordSerializationSchema());
         return FlinkSinkProvider.of(starRocksSink);
     }
 

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/com/ververica/cdc/connectors/starrocks/sink/StarRocksMetadataApplier.java

@@ -83,7 +83,14 @@ private void applyCreateTable(CreateTableEvent createTableEvent) {
         if (!catalog.databaseExists(starRocksTable.getDatabaseName())) {
             catalog.createDatabase(starRocksTable.getDatabaseName(), true);
         }
-        catalog.createTable(starRocksTable, true);
+
+        try {
+            catalog.createTable(starRocksTable, true);
+            LOG.info("Successful to create table, event: {}", createTableEvent);
+        } catch (StarRocksCatalogException e) {
+            LOG.error("Failed to create table, event: {}", createTableEvent.tableId(), e);
+            throw new RuntimeException("Failed to create table, event: " + createTableEvent, e);
+        }
     }
 
     private void applyAddColumn(AddColumnEvent addColumnEvent) {
@@ -119,7 +126,10 @@ private void applyAddColumn(AddColumnEvent addColumnEvent) {
             alterException = e;
         }
 
-        // Check the actual table
+        // Check whether the columns have been actually added to the table.
+        // This is useful for duplicate schema change after failover. Adding
+        // same columns will fail on StarRocks side, but it should be successful
+        // on CDC side
         StarRocksTable table = null;
         try {
             table = catalog.getTable(tableId.getSchemaName(), tableId.getTableName()).orElse(null);
@@ -184,7 +194,10 @@ private void applyDropColumn(DropColumnEvent dropColumnEvent) {
             alterException = e;
         }
 
-        // Check the actual table
+        // Check whether the columns have been actually dropped from the table.
+        // This is useful for duplicate schema change after failover. Drop
+        // non-existed columns will fail on StarRocks side, but it should be
+        // successful on CDC side
         StarRocksTable table = null;
         try {
             table = catalog.getTable(tableId.getSchemaName(), tableId.getTableName()).orElse(null);

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/com/ververical/cdc/connectors/starrocks/sink/EventRecordSerializerTest.java renamed to flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/com/ververical/cdc/connectors/starrocks/sink/EventRecordSerializationSchemaTest.java

@@ -40,7 +40,7 @@
 import com.ververica.cdc.common.types.TimestampType;
 import com.ververica.cdc.common.types.VarCharType;
 import com.ververica.cdc.common.utils.SchemaUtils;
-import com.ververica.cdc.connectors.starrocks.sink.EventRecordSerializer;
+import com.ververica.cdc.connectors.starrocks.sink.EventRecordSerializationSchema;
 import com.ververica.cdc.runtime.typeutils.BinaryRecordDataGenerator;
 import org.junit.After;
 import org.junit.Before;
@@ -56,15 +56,15 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
 
-/** Tests for {@link EventRecordSerializer}. */
-public class EventRecordSerializerTest {
+/** Tests for {@link EventRecordSerializationSchema}. */
+public class EventRecordSerializationSchemaTest {
 
-    private EventRecordSerializer serializer;
+    private EventRecordSerializationSchema serializer;
     private ObjectMapper objectMapper;
 
     @Before
     public void setup() {
-        this.serializer = new EventRecordSerializer();
+        this.serializer = new EventRecordSerializationSchema();
         this.serializer.open();
         this.objectMapper = new ObjectMapper();
     }