[Refactor] Rename StarRocksRecordSerializationSchema to RecordSeriali…

…zationSchema Signed-off-by: PengFei Li <[email protected]>
banmoy · Nov 28, 2023 · 92be62e · 92be62e
1 parent c88c133
commit 92be62e
Show file tree

Hide file tree

Showing 5 changed files with 9 additions and 9 deletions.
diff --git a/src/main/java/com/starrocks/connector/flink/table/sink/SinkFunctionFactory.java b/src/main/java/com/starrocks/connector/flink/table/sink/SinkFunctionFactory.java
@@ -25,7 +25,7 @@
 import com.starrocks.connector.flink.row.sink.StarRocksIRowTransformer;
 import com.starrocks.connector.flink.row.sink.StarRocksISerializer;
 import com.starrocks.connector.flink.row.sink.StarRocksSerializerFactory;
-import com.starrocks.connector.flink.table.sink.v2.StarRocksRecordSerializationSchema;
+import com.starrocks.connector.flink.table.sink.v2.RecordSerializationSchema;
 import com.starrocks.connector.flink.table.sink.v2.RowDataSerializationSchema;
 import com.starrocks.connector.flink.table.sink.v2.StarRocksSink;
 import com.starrocks.data.load.stream.properties.StreamLoadProperties;
@@ -152,12 +152,12 @@ public static StarRocksSink<RowData> createSink(
     }
 
     public static <T> StarRocksSink<T> createSink(
-            StarRocksSinkOptions sinkOptions, StarRocksRecordSerializationSchema<T> recordSerializer) {
+            StarRocksSinkOptions sinkOptions, RecordSerializationSchema<T> serializationSchema) {
         detectStarRocksFeature(sinkOptions);
         SinkVersion sinkVersion = getSinkVersion(sinkOptions);
         if (sinkVersion == SinkVersion.V2) {
             StreamLoadProperties streamLoadProperties = sinkOptions.getProperties(null);
-            return new StarRocksSink<>(sinkOptions, recordSerializer, streamLoadProperties);
+            return new StarRocksSink<>(sinkOptions, serializationSchema, streamLoadProperties);
         }
         throw new UnsupportedOperationException("New sink api don't support sink type " + sinkVersion.name());
     }

diff --git a/...2/StarRocksRecordSerializationSchema.java → ...le/sink/v2/RecordSerializationSchema.java b/...2/StarRocksRecordSerializationSchema.java → ...le/sink/v2/RecordSerializationSchema.java
@@ -30,7 +30,7 @@
  *
  * @param <T> the type of input record being serialized
  */
-public interface StarRocksRecordSerializationSchema<T> extends Serializable {
+public interface RecordSerializationSchema<T> extends Serializable {
 
     /** Open the serializer. */
     void open();

diff --git a/src/main/java/com/starrocks/connector/flink/table/sink/v2/RowDataSerializationSchema.java b/src/main/java/com/starrocks/connector/flink/table/sink/v2/RowDataSerializationSchema.java
@@ -30,7 +30,7 @@
 import com.starrocks.connector.flink.tools.JsonWrapper;
 
 /** Serializer for the {@link RowData} record. */
-public class RowDataSerializationSchema implements StarRocksRecordSerializationSchema<RowData> {
+public class RowDataSerializationSchema implements RecordSerializationSchema<RowData> {
 
     private static final long serialVersionUID = 1L;
 

diff --git a/src/main/java/com/starrocks/connector/flink/table/sink/v2/StarRocksSink.java b/src/main/java/com/starrocks/connector/flink/table/sink/v2/StarRocksSink.java
@@ -42,12 +42,12 @@ public class StarRocksSink<InputT>
     private static final Logger LOG = LoggerFactory.getLogger(StarRocksSink.class);
 
     private final StarRocksSinkOptions sinkOptions;
-    private final StarRocksRecordSerializationSchema<InputT> serializationSchema;
+    private final RecordSerializationSchema<InputT> serializationSchema;
     private final StreamLoadProperties streamLoadProperties;
 
     public StarRocksSink(
             StarRocksSinkOptions sinkOptions,
-            StarRocksRecordSerializationSchema<InputT> serializationSchema,
+            RecordSerializationSchema<InputT> serializationSchema,
             StreamLoadProperties streamLoadProperties) {
         this.sinkOptions = sinkOptions;
         this.serializationSchema = serializationSchema;

diff --git a/src/main/java/com/starrocks/connector/flink/table/sink/v2/StarRocksWriter.java b/src/main/java/com/starrocks/connector/flink/table/sink/v2/StarRocksWriter.java
@@ -53,15 +53,15 @@ public class StarRocksWriter<InputT>
     private static final Logger LOG = LoggerFactory.getLogger(StarRocksWriter.class);
 
     private final StarRocksSinkOptions sinkOptions;
-    private final StarRocksRecordSerializationSchema<InputT> serializationSchema;
+    private final RecordSerializationSchema<InputT> serializationSchema;
     private final StarRocksStreamLoadListener streamLoadListener;
     private final LabelGeneratorFactory labelGeneratorFactory;
     private final StreamLoadManagerV2 sinkManager;
     private long totalReceivedRows = 0;
 
     public StarRocksWriter(
             StarRocksSinkOptions sinkOptions,
-            StarRocksRecordSerializationSchema<InputT> serializationSchema,
+            RecordSerializationSchema<InputT> serializationSchema,
             StreamLoadProperties streamLoadProperties,
             Sink.InitContext initContext,
             Collection<StarRocksWriterState> recoveredState) throws Exception {