[core] Make CatalogContext implements Serializable (#4936)

paimon-common/src/main/java/org/apache/paimon/catalog/CatalogContext.java

@@ -21,12 +21,15 @@
 import org.apache.paimon.annotation.Public;
 import org.apache.paimon.fs.FileIOLoader;
 import org.apache.paimon.fs.Path;
+import org.apache.paimon.hadoop.SerializableConfiguration;
 import org.apache.paimon.options.Options;
 
 import org.apache.hadoop.conf.Configuration;
 
 import javax.annotation.Nullable;
 
+import java.io.Serializable;
+
 import static org.apache.paimon.options.CatalogOptions.WAREHOUSE;
 import static org.apache.paimon.utils.HadoopUtils.getHadoopConfiguration;
 import static org.apache.paimon.utils.Preconditions.checkNotNull;
@@ -37,10 +40,12 @@
  * @since 0.4.0
  */
 @Public
-public class CatalogContext {
+public class CatalogContext implements Serializable {
+
+    private static final long serialVersionUID = 1L;
 
     private final Options options;
-    private final Configuration hadoopConf;
+    private final SerializableConfiguration hadoopConf;
     @Nullable private final FileIOLoader preferIOLoader;
     @Nullable private final FileIOLoader fallbackIOLoader;
 
@@ -50,7 +55,9 @@ private CatalogContext(
             @Nullable FileIOLoader preferIOLoader,
             @Nullable FileIOLoader fallbackIOLoader) {
         this.options = checkNotNull(options);
-        this.hadoopConf = hadoopConf == null ? getHadoopConfiguration(options) : hadoopConf;
+        this.hadoopConf =
+                new SerializableConfiguration(
+                        hadoopConf == null ? getHadoopConfiguration(options) : hadoopConf);
         this.preferIOLoader = preferIOLoader;
         this.fallbackIOLoader = fallbackIOLoader;
     }
@@ -92,7 +99,7 @@ public Options options() {
 
     /** Return hadoop {@link Configuration}. */
     public Configuration hadoopConf() {
-        return hadoopConf;
+        return hadoopConf.get();
     }
 
     @Nullable

paimon-core/src/test/java/org/apache/paimon/catalog/CatalogFactoryTest.java

@@ -22,6 +22,8 @@
 import org.apache.paimon.fs.local.LocalFileIO;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.table.CatalogTableType;
+import org.apache.paimon.utils.HadoopUtilsITCase.TestFileIOLoader;
+import org.apache.paimon.utils.InstantiationUtil;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdfs.HdfsConfiguration;
@@ -84,4 +86,15 @@ public void testContextDefaultHadoopConf(@TempDir java.nio.file.Path path) {
         assertThat(conf.get("fs.defaultFS")).isEqualTo(defaultFS);
         assertThat(conf.get("dfs.replication")).isEqualTo(replication);
     }
+
+    @Test
+    public void testContextSerializable() throws IOException, ClassNotFoundException {
+        Configuration conf = new Configuration(false);
+        conf.set("my_key", "my_value");
+        CatalogContext context =
+                CatalogContext.create(
+                        new Options(), conf, new TestFileIOLoader(), new TestFileIOLoader());
+        context = InstantiationUtil.clone(context);
+        assertThat(context.hadoopConf().get("my_key")).isEqualTo(conf.get("my_key"));
+    }
 }