[typo] fix typos commited -> committed (#1457)

Author: xiaochen-zhou

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/tiering/committer/TieringCommitOperator.java

@@ -75,7 +75,7 @@
  * LakeCommitter#toCommittable(List)}, and then call method {@link LakeCommitter#commit(Object,
  * Map)} to commit to lake.
  *
- * <p>Finally, it will also commit the commited lake snapshot to Fluss cluster to make Fluss aware
+ * <p>Finally, it will also commit the committed lake snapshot to Fluss cluster to make Fluss aware
  * of the tiering progress.
  */
 public class TieringCommitOperator<WriteResult, Committable>
@@ -195,23 +195,23 @@ private Committable commitWriteResults(
 
             // to committable
             Committable committable = lakeCommitter.toCommittable(writeResults);
-            // before commit to lake, check fluss not missing any lake snapshot commited by fluss
+            // before commit to lake, check fluss not missing any lake snapshot committed by fluss
             checkFlussNotMissingLakeSnapshot(
                     tablePath,
                     lakeCommitter,
                     committable,
                     flussCurrentLakeSnapshot == null
                             ? null
                             : flussCurrentLakeSnapshot.getSnapshotId());
-            long commitedSnapshotId =
+            long committedSnapshotId =
                     lakeCommitter.commit(committable, toBucketOffsetsProperty(logOffsets));
             // commit to fluss
             Map<TableBucket, Long> logEndOffsets = new HashMap<>();
             for (TableBucketWriteResult<WriteResult> writeResult : committableWriteResults) {
                 logEndOffsets.put(writeResult.tableBucket(), writeResult.logEndOffset());
             }
             flussTableLakeSnapshotCommitter.commit(
-                    new FlussTableLakeSnapshot(tableId, commitedSnapshotId, logEndOffsets));
+                    new FlussTableLakeSnapshot(tableId, committedSnapshotId, logEndOffsets));
             return committable;
         }
     }

fluss-flink/fluss-flink-common/src/test/java/com/alibaba/fluss/flink/tiering/committer/TieringCommitOperatorTest.java

@@ -231,10 +231,10 @@ void testCommitMeetsEmptyWriteResult() throws Exception {
 
     @Test
     void testTableCommitWhenFlussMissingLakeSnapshot() throws Exception {
-        CommittedLakeSnapshot mockCommitedSnapshot =
+        CommittedLakeSnapshot mockCommittedSnapshot =
                 mockCommittedLakeSnapshot(Collections.singletonList(null), 2);
         TestingLakeTieringFactory.TestingLakeCommitter testingLakeCommitter =
-                new TestingLakeTieringFactory.TestingLakeCommitter(mockCommitedSnapshot);
+                new TestingLakeTieringFactory.TestingLakeCommitter(mockCommittedSnapshot);
         committerOperator =
                 new TieringCommitOperator<>(
                         parameters,
@@ -257,15 +257,15 @@ void testTableCommitWhenFlussMissingLakeSnapshot() throws Exception {
                 tablePath,
                 tableId,
                 2,
-                getExpectedLogEndOffsets(tableId, mockCommitedSnapshot, Collections.emptyMap()),
+                getExpectedLogEndOffsets(tableId, mockCommittedSnapshot, Collections.emptyMap()),
                 String.format(
                         "The current Fluss's lake snapshot %d is less than lake actual snapshot %d committed by Fluss for table: {tablePath=%s, tableId=%d},"
                                 + " missing snapshot: %s.",
                         null,
-                        mockCommitedSnapshot.getLakeSnapshotId(),
+                        mockCommittedSnapshot.getLakeSnapshotId(),
                         tablePath,
                         tableId,
-                        mockCommitedSnapshot));
+                        mockCommittedSnapshot));
 
         Map<TableBucket, Long> expectedLogEndOffsets = new HashMap<>();
         for (int bucket = 0; bucket < 3; bucket++) {
@@ -290,10 +290,10 @@ void testPartitionedTableCommitWhenFlussMissingLakeSnapshot() throws Exception {
         Map<String, Long> partitionIdByNames =
                 FLUSS_CLUSTER_EXTENSION.waitUntilPartitionAllReady(tablePath);
 
-        CommittedLakeSnapshot mockCommitedSnapshot =
+        CommittedLakeSnapshot mockCommittedSnapshot =
                 mockCommittedLakeSnapshot(Collections.singletonList(null), 3);
         TestingLakeTieringFactory.TestingLakeCommitter testingLakeCommitter =
-                new TestingLakeTieringFactory.TestingLakeCommitter(mockCommitedSnapshot);
+                new TestingLakeTieringFactory.TestingLakeCommitter(mockCommittedSnapshot);
         committerOperator =
                 new TieringCommitOperator<>(
                         parameters,
@@ -317,14 +317,14 @@ void testPartitionedTableCommitWhenFlussMissingLakeSnapshot() throws Exception {
                 tablePath,
                 tableId,
                 3,
-                getExpectedLogEndOffsets(tableId, mockCommitedSnapshot, Collections.emptyMap()),
+                getExpectedLogEndOffsets(tableId, mockCommittedSnapshot, Collections.emptyMap()),
                 String.format(
                         "The current Fluss's lake snapshot %d is less than lake actual snapshot %d committed by Fluss for table: {tablePath=%s, tableId=%d}, missing snapshot: %s.",
                         null,
-                        mockCommitedSnapshot.getLakeSnapshotId(),
+                        mockCommittedSnapshot.getLakeSnapshotId(),
                         tablePath,
                         tableId,
-                        mockCommitedSnapshot));
+                        mockCommittedSnapshot));
     }
 
     private CommittedLakeSnapshot mockCommittedLakeSnapshot(List<Long> partitions, int snapshotId) {

fluss-lake/fluss-lake-paimon/src/main/java/com/alibaba/fluss/lake/paimon/tiering/PaimonLakeCommitter.java

@@ -167,7 +167,7 @@ public CommittedLakeSnapshot getMissingLakeSnapshot(@Nullable Long latestLakeSna
 
     @Nullable
     private Snapshot getCommittedLatestSnapshotOfLake(String commitUser) throws IOException {
-        // get the latest snapshot commited by fluss or latest commited id
+        // get the latest snapshot committed by fluss or latest committed id
         SnapshotManager snapshotManager = fileStoreTable.snapshotManager();
         Long userCommittedSnapshotIdOrLatestCommitId =
                 fileStoreTable
@@ -182,7 +182,7 @@ private Snapshot getCommittedLatestSnapshotOfLake(String commitUser) throws IOEx
         Snapshot snapshot = snapshotManager.tryGetSnapshot(userCommittedSnapshotIdOrLatestCommitId);
 
         if (!snapshot.commitUser().equals(commitUser)) {
-            // the snapshot is still not commited by Fluss, return directly
+            // the snapshot is still not committed by Fluss, return directly
             return null;
         }
         return snapshot;