[FLINK-39063][connect/fluss] Fix some spelling mistakes in fluss pipeline connector (#4271)

Author: ThorneANN

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussEventSerializationSchema.java

@@ -145,14 +145,14 @@ private TableSchemaInfo(
 
     static Map<Integer, Integer> sanityCheckAndGenerateIndexMapping(
             com.alibaba.fluss.metadata.Schema inferredFlussSchema,
-            com.alibaba.fluss.metadata.Schema currentFlussnewSchema) {
+            com.alibaba.fluss.metadata.Schema currentFlussNewSchema) {
         List<String> inferredSchemaColumnNames = inferredFlussSchema.getColumnNames();
         Map<String, Integer> reverseIndex = new HashMap<>();
         for (int i = 0; i < inferredSchemaColumnNames.size(); i++) {
             reverseIndex.put(inferredSchemaColumnNames.get(i), i);
         }
 
-        List<String> currentSchemaColumnNames = currentFlussnewSchema.getColumnNames();
+        List<String> currentSchemaColumnNames = currentFlussNewSchema.getColumnNames();
         Map<Integer, Integer> indexMapping = new HashMap<>();
         for (int newSchemaIndex = 0;
                 newSchemaIndex < currentSchemaColumnNames.size();
@@ -167,7 +167,7 @@ static Map<Integer, Integer> sanityCheckAndGenerateIndexMapping(
                 // changes.
                 // In the future, meta applier will be used to handle column changes.
                 DataType oldDataType = inferredFlussSchema.getRowType().getTypeAt(oldSchemaIndex);
-                DataType newDataType = currentFlussnewSchema.getRowType().getTypeAt(newSchemaIndex);
+                DataType newDataType = currentFlussNewSchema.getRowType().getTypeAt(newSchemaIndex);
                 if (!oldDataType.copy(false).equals(newDataType.copy(false))) {
                     throw new IllegalArgumentException(
                             "The data type of column "

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetaDataApplier.java

@@ -149,7 +149,7 @@ private void sanityCheck(TableDescriptor inferredFlussTable, TableInfo currentTa
                         .collect(Collectors.toList());
         if (!inferredPrimaryKeyColumnNames.equals(currentPrimaryKeyColumnNames)) {
             throw new ValidationException(
-                    "The table schema inffered by Flink CDC is not matched with current Fluss table schema. "
+                    "The table schema inferred by Flink CDC is not matched with current Fluss table schema. "
                             + "\n New Fluss table's primary keys : "
                             + inferredPrimaryKeyColumnNames
                             + "\n Current Fluss's primary keys: "
@@ -160,7 +160,7 @@ private void sanityCheck(TableDescriptor inferredFlussTable, TableInfo currentTa
         List<String> currentBucketKeys = currentTableInfo.getBucketKeys();
         if (!inferredBucketKeys.equals(currentBucketKeys)) {
             throw new ValidationException(
-                    "The table schema inffered by Flink CDC is not matched with current Fluss table schema. "
+                    "The table schema inferred by Flink CDC is not matched with current Fluss table schema. "
                             + "\n New Fluss table's bucket keys : "
                             + inferredBucketKeys
                             + "\n Current Fluss's bucket keys: "
@@ -171,7 +171,7 @@ private void sanityCheck(TableDescriptor inferredFlussTable, TableInfo currentTa
         List<String> currentPartitionKeys = currentTableInfo.getPartitionKeys();
         if (!inferredPartitionKeys.equals(currentPartitionKeys)) {
             throw new ValidationException(
-                    "The table schema inffered by Flink CDC is not matched with current Fluss table schema. "
+                    "The table schema inferred by Flink CDC is not matched with current Fluss table schema. "
                             + "\n New Fluss table's partition keys : "
                             + inferredPartitionKeys
                             + "\n Current Fluss's partition keys: "

…sink/v2/metrics/WarppedFlussCounter.java …sink/v2/metrics/WrappedFlussCounter.javaflink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/WarppedFlussCounter.java renamed to flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/WrappedFlussCounter.java flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/WarppedFlussCounter.java renamed to flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/WrappedFlussCounter.java

@@ -24,11 +24,11 @@
  * additional information regarding copyright ownership. */
 
 /** An implementation of Flink's {@link Counter} which wraps Fluss's Counter. */
-public class WarppedFlussCounter implements Counter {
+public class WrappedFlussCounter implements Counter {
 
     private final com.alibaba.fluss.metrics.Counter flussCounter;
 
-    public WarppedFlussCounter(com.alibaba.fluss.metrics.Counter flussCounter) {
+    public WrappedFlussCounter(com.alibaba.fluss.metrics.Counter flussCounter) {
         this.flussCounter = flussCounter;
     }
 

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/WrapperFlussMetricRegistry.java

@@ -116,7 +116,7 @@ private Map<String, String> getVariables(AbstractMetricGroup group) {
     private void registerMetric(MetricGroup metricGroup, Metric metric, String metricName) {
         switch (metric.getMetricType()) {
             case COUNTER:
-                metricGroup.counter(metricName, new WarppedFlussCounter((Counter) metric));
+                metricGroup.counter(metricName, new WrappedFlussCounter((Counter) metric));
                 break;
             case METER:
                 metricGroup.meter(metricName, new WrapperFlussMeter((Meter) metric));

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussMetadataApplierTest.java

@@ -511,7 +511,7 @@ void testRecreateTableWithDifferentSchema() throws Exception {
                                     applier.applySchemaChange(
                                             new CreateTableEvent(tableId, differentSchema)))
                     .hasMessageContaining(
-                            "The table schema inffered by Flink CDC is not matched with current Fluss table schema");
+                            "The table schema inferred by Flink CDC is not matched with current Fluss table schema");
         }
 
         // recreate table with schema1 again