From 90b7732224997b3cf56e445a2f98ebb1bfd9ac4d Mon Sep 17 00:00:00 2001
From: chenxu <chenxu@dmetasoul.com>
Date: Tue, 10 Oct 2023 18:46:26 +0800
Subject: [PATCH] update fury to 0.4.0

Signed-off-by: chenxu <chenxu@dmetasoul.com>
---
 lakesoul-flink/pom.xml                                   | 2 +-
 .../lakesoul/types/BinarySourceRecordSerializer.java     | 9 ++++++---
 2 files changed, 7 insertions(+), 4 deletions(-)

diff --git a/lakesoul-flink/pom.xml b/lakesoul-flink/pom.xml
index f9dbc9e0d..e9a42ba9c 100644
--- a/lakesoul-flink/pom.xml
+++ b/lakesoul-flink/pom.xml
@@ -250,7 +250,7 @@ SPDX-License-Identifier: Apache-2.0
         <dependency>
             <groupId>org.furyio</groupId>
             <artifactId>fury-core</artifactId>
-            <version>0.1.0</version>
+            <version>0.4.0</version>
         </dependency>
 
         <dependency>
diff --git a/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/types/BinarySourceRecordSerializer.java b/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/types/BinarySourceRecordSerializer.java
index 7352c60af..235782cdf 100644
--- a/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/types/BinarySourceRecordSerializer.java
+++ b/lakesoul-flink/src/main/java/org/apache/flink/lakesoul/types/BinarySourceRecordSerializer.java
@@ -9,8 +9,8 @@
 import com.esotericsoftware.kryo.io.Input;
 import com.esotericsoftware.kryo.io.Output;
 import io.fury.Fury;
-import io.fury.Language;
 import io.fury.ThreadLocalFury;
+import io.fury.config.Language;
 import org.apache.flink.core.memory.MemorySegment;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.data.binary.BinaryFormat;
@@ -78,10 +78,13 @@ public BinarySourceRecordSerializer() {
     }
 
     @Override public void write(Kryo kryo, Output output, BinarySourceRecord object) {
-        fury.getCurrentFury().serializeJavaObject(output, object);
+        fury.execute(f -> {
+            f.serializeJavaObject(output, object);
+            return 0;
+        });
     }
 
     @Override public BinarySourceRecord read(Kryo kryo, Input input, Class<BinarySourceRecord> type) {
-        return fury.getCurrentFury().deserializeJavaObject(input, BinarySourceRecord.class);
+        return fury.execute(f -> f.deserializeJavaObject(input, BinarySourceRecord.class));
     }
 }