[core] Load RocksDB library in advance and catch exceptions

Author: LsomeYeah

paimon-core/src/main/java/org/apache/paimon/lookup/rocksdb/RocksDBStateFactory.java

@@ -29,6 +29,8 @@
 import org.rocksdb.RocksDB;
 import org.rocksdb.RocksDBException;
 import org.rocksdb.TtlDB;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nullable;
 
@@ -39,6 +41,8 @@
 /** Factory to create state. */
 public class RocksDBStateFactory implements StateFactory {
 
+    private static final Logger LOG = LoggerFactory.getLogger(RocksDBStateFactory.class);
+
     public static final String MERGE_OPERATOR_NAME = "stringappendtest";
 
     private final Options options;
@@ -50,6 +54,13 @@ public class RocksDBStateFactory implements StateFactory {
     public RocksDBStateFactory(
             String path, org.apache.paimon.options.Options conf, @Nullable Duration ttlSecs)
             throws IOException {
+        try {
+            RocksDB.loadLibrary();
+        } catch (Throwable e) {
+            LOG.error("Fail to load RocksDB library.", e);
+            throw new IOException("Fail to load RocksDB library.", e);
+        }
+
         DBOptions dbOptions =
                 RocksDBOptions.createDBOptions(
                         new DBOptions()

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/RocksDBListStateTest.java

@@ -33,12 +33,17 @@
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+import org.rocksdb.RocksDB;
 
+import java.io.IOException;
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.List;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Test for {@link RocksDBListState}. */
 public class RocksDBListStateTest {
@@ -71,6 +76,53 @@ void test() throws Exception {
         factory.close();
     }
 
+    @Test
+    void testRocksDBLibraryLoadSuccess() throws Exception {
+        // Test that RocksDBStateFactory can be created successfully when RocksDB library loads
+        // properly
+        RocksDBStateFactory factory =
+                new RocksDBStateFactory(tempDir.toString(), new Options(), null);
+
+        // Verify that the factory is created and can be used
+        assertThat(factory).isNotNull();
+        assertThat(factory.db()).isNotNull();
+        assertThat(factory.path()).isEqualTo(tempDir.toString());
+
+        factory.close();
+    }
+
+    @Test
+    void testRocksDBLibraryLoadFailure() {
+        // Test that IOException is thrown when RocksDB library fails to load
+        try (MockedStatic<RocksDB> mockedRocksDB = Mockito.mockStatic(RocksDB.class)) {
+            mockedRocksDB
+                    .when(RocksDB::loadLibrary)
+                    .thenThrow(new RuntimeException("Failed to load RocksDB library"));
+
+            assertThatThrownBy(
+                            () -> new RocksDBStateFactory(tempDir.toString(), new Options(), null))
+                    .isInstanceOf(IOException.class)
+                    .hasMessage("Fail to load RocksDB library.")
+                    .hasCauseInstanceOf(RuntimeException.class);
+        }
+    }
+
+    @Test
+    void testRocksDBLibraryLoadWithDifferentExceptionTypes() {
+        // Test that IOException is thrown for different types of exceptions during library loading
+        try (MockedStatic<RocksDB> mockedRocksDB = Mockito.mockStatic(RocksDB.class)) {
+            mockedRocksDB
+                    .when(RocksDB::loadLibrary)
+                    .thenThrow(new UnsatisfiedLinkError("Native library not found"));
+
+            assertThatThrownBy(
+                            () -> new RocksDBStateFactory(tempDir.toString(), new Options(), null))
+                    .isInstanceOf(IOException.class)
+                    .hasMessage("Fail to load RocksDB library.")
+                    .hasCauseInstanceOf(UnsatisfiedLinkError.class);
+        }
+    }
+
     public GenericRow row(String value) {
         return GenericRow.of(bs(value));
     }