The metrics package had inconsistent naming conventions and a typo in the class name:

  - Typo: WarppedFlussCounter (double 'p') instead of proper spelling
  - Inconsistency: Mix of Wrapped and Wrapper prefixes across metric wrapper classes

  Files Modified

  1. Renamed Metric Wrapper Classes

  WrapperFlussCounter.java (formerly WarppedFlussCounter.java)
  - Fixed typo: Warpped → Wrapper
  - Updated class name: WarppedFlussCounter → WrapperFlussCounter
  - Updated constructor name to match

  WrapperFlussGauge.java (formerly WrappedFlussGauge.java)
  - Unified naming: WrappedFlussGauge → WrapperFlussGauge
  - Updated class name and constructor to use Wrapper prefix

  2. Updated References

  WrapperFlussMetricRegistry.java
  - Line 119: Updated new WrappedFlussCounter(...) → new WrapperFlussCounter(...)
  - Line 125: Updated new WrappedFlussGauge<>(...) → new WrapperFlussGauge<>(...)

  3. Fixed Test Expectations

  FlussMetadataApplierTest.java
  - Line 514: Fixed expected error message spelling from "inffered" → "inferred"
  - This aligned test expectations with the actual error messages in production code

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/WrapperFlussCounter.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/WrapperFlussCounter.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/WrapperFlussCounter.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 WrapperFlussCounter implements Counter {
 
     private final com.alibaba.fluss.metrics.Counter flussCounter;
 
-    public WarppedFlussCounter(com.alibaba.fluss.metrics.Counter flussCounter) {
+    public WrapperFlussCounter(com.alibaba.fluss.metrics.Counter flussCounter) {
         this.flussCounter = flussCounter;
     }
 

…s/sink/v2/metrics/WrappedFlussGauge.java …s/sink/v2/metrics/WrapperFlussGauge.javaflink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/v2/metrics/WrappedFlussGauge.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/WrapperFlussGauge.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/WrappedFlussGauge.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/WrapperFlussGauge.java

@@ -24,11 +24,11 @@
  * additional information regarding copyright ownership. */
 
 /** An implementation of Flink's {@link Gauge} which wraps Fluss's Gauge. */
-public class WrappedFlussGauge<T> implements Gauge<T> {
+public class WrapperFlussGauge<T> implements Gauge<T> {
 
     private final com.alibaba.fluss.metrics.Gauge<T> flussGauge;
 
-    public WrappedFlussGauge(com.alibaba.fluss.metrics.Gauge<T> flussGauge) {
+    public WrapperFlussGauge(com.alibaba.fluss.metrics.Gauge<T> flussGauge) {
         this.flussGauge = flussGauge;
     }
 

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,13 +116,13 @@ 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 WrapperFlussCounter((Counter) metric));
                 break;
             case METER:
                 metricGroup.meter(metricName, new WrapperFlussMeter((Meter) metric));
                 break;
             case GAUGE:
-                metricGroup.gauge(metricName, new WrappedFlussGauge<>((Gauge<?>) metric));
+                metricGroup.gauge(metricName, new WrapperFlussGauge<>((Gauge<?>) metric));
                 break;
             case HISTOGRAM:
                 metricGroup.histogram(metricName, new WrapperFlussHistogram((Histogram) 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