[typo] Fix many typos and optimize some code

Author: zcoo

fluss-client/src/main/java/org/apache/fluss/client/lookup/LookupType.java

@@ -24,5 +24,5 @@
 public enum LookupType {
     LOOKUP,
     LOOKUP_WITH_INSERT_IF_NOT_EXISTS,
-    PREFIX_LOOKUP;
+    PREFIX_LOOKUP
 }

fluss-common/src/test/java/org/apache/fluss/row/BinaryArrayTest.java

@@ -287,7 +287,7 @@ public void testToArrayWithNullThrowsException() {
         writer.writeInt(2, 30);
         writer.complete();
 
-        assertThatThrownBy(() -> array.toIntArray())
+        assertThatThrownBy(array::toIntArray)
                 .isInstanceOf(RuntimeException.class)
                 .hasMessageContaining("Primitive array must not contain a null value");
     }

fluss-common/src/test/java/org/apache/fluss/row/GenericArrayTest.java

@@ -185,7 +185,7 @@ public void testNullInObjectArray() {
         assertThat(array.isNullAt(1)).isTrue();
         assertThat(array.isNullAt(2)).isFalse();
 
-        assertThatThrownBy(() -> array.toIntArray())
+        assertThatThrownBy(array::toIntArray)
                 .isInstanceOf(RuntimeException.class)
                 .hasMessageContaining("Primitive array must not contain a null value");
     }

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TableBucketWriteResult.java

@@ -53,7 +53,7 @@ public class TableBucketWriteResult<WriteResult> implements Serializable {
     private final long maxTimestamp;
 
     // the total number of write results in one round of tiering,
-    // used for downstream commiter operator to determine when all write results
+    // used for downstream committer operator to determine when all write results
     // for the round of tiering is finished
     private final int numberOfWriteResults;
 

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java

@@ -492,7 +492,7 @@ private List<TieringSplit> populateNumberOfTieringSplits(List<TieringSplit> tier
 
     @Override
     public TieringSourceEnumeratorState snapshotState(long checkpointId) throws Exception {
-        // do nothing, the downstream lake commiter will snapshot the state to Fluss Cluster
+        // do nothing, the downstream lake committer will snapshot the state to Fluss Cluster
         return new TieringSourceEnumeratorState();
     }
 

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java

@@ -205,7 +205,7 @@ void beforeEach() throws Exception {
                     DEFAULT_DB, new CatalogDatabaseImpl(Collections.emptyMap(), null), true);
         } catch (CatalogException e) {
             // the auto partitioned manager may create the db zk node
-            // in an another thread, so if exception is NodeExistsException, just ignore
+            // in another thread, so if exception is NodeExistsException, just ignore
             if (!ExceptionUtils.findThrowableWithMessage(e, "KeeperException$NodeExistsException")
                     .isPresent()) {
                 throw e;
@@ -275,7 +275,7 @@ void testCreateTable() throws Exception {
                                 this.tableInDefaultDb, CATALOG_NAME));
         // should be ok since we set ignoreIfNotExists = true
         catalog.dropTable(this.tableInDefaultDb, true);
-        // create table from an non-exist db
+        // create table from a non-exist db
         ObjectPath nonExistDbPath = ObjectPath.fromString("non.exist");
 
         // remove bucket-key
@@ -300,7 +300,7 @@ void testCreateTable() throws Exception {
                         resolvedSchema);
         catalog.createTable(this.tableInDefaultDb, table2, false);
         tableCreated = catalog.getTable(this.tableInDefaultDb);
-        // need to over write the option
+        // need to overwrite the option
         addedOptions.put(BUCKET_KEY.key(), "third");
 
         expectedTable = addOptions(table2, addedOptions);
@@ -492,7 +492,7 @@ void testCreateAndDropMaterializedTable() throws Exception {
         // should be ok since we set ignoreIfNotExists = true
         catalog.dropTable(mt1, true);
 
-        // create table from an non-exist db
+        // create table from a non-exist db
         ObjectPath nonExistDbPath = ObjectPath.fromString("non.exist");
 
         // remove bucket-key
@@ -658,10 +658,10 @@ void testDatabase() throws Exception {
     @Test
     void testOperatePartitions() throws Exception {
         catalog.createDatabase("db1", new CatalogDatabaseImpl(Collections.emptyMap(), null), false);
-        assertThatThrownBy(() -> catalog.listPartitions(new ObjectPath("db1", "unkown_table")))
+        assertThatThrownBy(() -> catalog.listPartitions(new ObjectPath("db1", "unknown_table")))
                 .isInstanceOf(TableNotExistException.class)
                 .hasMessage(
-                        "Table (or view) db1.unkown_table does not exist in Catalog test-catalog.");
+                        "Table (or view) db1.unknown_table does not exist in Catalog test-catalog.");
 
         // create a none partitioned table.
         CatalogTable table = this.newCatalogTable(Collections.emptyMap());
@@ -843,7 +843,7 @@ void testConnectionFailureHandling() {
                         Collections::emptyMap);
 
         // Test open() throws proper exception
-        assertThatThrownBy(() -> badCatalog.open())
+        assertThatThrownBy(badCatalog::open)
                 .isInstanceOf(IllegalConfigurationException.class)
                 .hasMessageContaining("No resolvable bootstrap urls");
     }

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java

@@ -739,7 +739,7 @@ void testTableReachMaxTieringDuration() throws Throwable {
         try (FlussMockSplitEnumeratorContext<TieringSplit> context =
                         new FlussMockSplitEnumeratorContext<>(numSubtasks);
                 TieringSourceEnumerator enumerator =
-                        createTieringSourceEnumerator(flussConf, context); ) {
+                        createTieringSourceEnumerator(flussConf, context)) {
             enumerator.start();
 
             // Register all readers

fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergWriteResult.java

@@ -25,7 +25,7 @@
 
 import java.io.Serializable;
 
-/** The write result of Iceberg lake writer to pass to commiter to commit. */
+/** The write result of Iceberg lake writer to pass to committer to commit. */
 public class IcebergWriteResult implements Serializable {
 
     private static final long serialVersionUID = 1L;

fluss-lake/fluss-lake-lance/src/main/java/org/apache/fluss/lake/lance/tiering/LanceWriteResult.java

@@ -22,7 +22,7 @@
 import java.io.Serializable;
 import java.util.List;
 
-/** The write result of Lance lake writer to pass to commiter to commit. */
+/** The write result of Lance lake writer to pass to committer to commit. */
 public class LanceWriteResult implements Serializable {
     private static final long serialVersionUID = 1L;
 

fluss-lake/fluss-lake-lance/src/main/java/org/apache/fluss/lake/lance/tiering/LanceWriteResultSerializer.java

@@ -37,7 +37,7 @@ public int getVersion() {
     @Override
     public byte[] serialize(LanceWriteResult lanceWriteResult) throws IOException {
         try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
-                ObjectOutputStream oos = new ObjectOutputStream(baos); ) {
+                ObjectOutputStream oos = new ObjectOutputStream(baos)) {
             oos.writeObject(lanceWriteResult);
             return baos.toByteArray();
         }