feat(java): support auto generic cast for fury deserialization

java/fury-benchmark/src/main/java/io/fury/benchmark/state/JsonbState.java

@@ -18,7 +18,6 @@
 
 import com.alibaba.fastjson2.JSONB;
 import com.alibaba.fastjson2.JSONFactory;
-import com.alibaba.fastjson2.JSONObject;
 import com.alibaba.fastjson2.JSONReader;
 import com.alibaba.fastjson2.JSONWriter;
 import com.alibaba.fastjson2.reader.ObjectReaderProvider;
@@ -239,12 +238,5 @@ public static void main(String[] args) {
     state.setup();
     state.bufferType = BufferType.directBuffer;
     state.setup();
-
-    ;
-    JSONObject json = new JSONObject();
-    json.put("k", 1);
-    Fury fury = Fury.builder().requireClassRegistration(false).build();
-    byte[] bytes = fury.serialize(json);
-    System.out.println(fury.deserialize(bytes));
   }
 }

java/fury-core/src/main/java/io/fury/Fury.java

@@ -70,6 +70,7 @@
  * @author chaokunyang
  */
 @NotThreadSafe
+@SuppressWarnings("unchecked")
 public final class Fury {
   private static final Logger LOG = LoggerFactory.getLogger(Fury.class);
 
@@ -617,7 +618,33 @@ public long readLong(MemoryBuffer buffer) {
     return LongSerializer.readLong(buffer, longEncoding);
   }
 
-  /** Deserialize <code>obj</code> from a byte array. */
+  /**
+   * Deserialize <code>obj</code> from a byte array. Note declared result type must be a class or
+   * superclass/interface of the serialized object, otherwise a {@link ClassCastException} will be
+   * thrown.
+   *
+   * <p>For example, if you serialized an object of type `Integer`, then your declared deserialized
+   * type is `String`, the deserialization will just throw a {@link ClassCastException}.
+   *
+   * <pre>{@code
+   * byte[] bytes = fury.serialize(1);
+   * String o = fury.deserialize(bytes); // throw ClassCastException.
+   * }</pre>
+   *
+   * @param bytes serialized data.
+   * @param <T> result type of the data.
+   * @return deserialized object.
+   */
+  public <T> T deserializeTyped(byte[] bytes) {
+    return (T) deserialize(MemoryUtils.wrap(bytes), null);
+  }
+
+  /**
+   * Deserialize <code>obj</code> from a byte array.
+   *
+   * @param bytes serialized data.
+   * @return deserialized object.
+   */
   public Object deserialize(byte[] bytes) {
     return deserialize(MemoryUtils.wrap(bytes), null);
   }

java/fury-core/src/main/java/io/fury/ThreadLocalFury.java

@@ -69,6 +69,10 @@ public MemoryBuffer serialize(MemoryBuffer buffer, Object obj) {
     return bindingThreadLocal.get().get().serialize(buffer, obj);
   }
 
+  public <T> T deserializeTyped(byte[] bytes) {
+    return bindingThreadLocal.get().get().deserializeTyped(bytes);
+  }
+
   public Object deserialize(byte[] bytes) {
     return bindingThreadLocal.get().get().deserialize(bytes);
   }

java/fury-core/src/main/java/io/fury/ThreadSafeFury.java

@@ -39,6 +39,8 @@ public interface ThreadSafeFury {
 
   MemoryBuffer serialize(MemoryBuffer buffer, Object obj);
 
+  <T> T deserializeTyped(byte[] bytes);
+
   Object deserialize(byte[] bytes);
 
   Object deserialize(long address, int size);

java/fury-core/src/main/java/io/fury/pool/ThreadPoolFury.java

@@ -71,6 +71,16 @@ public MemoryBuffer serialize(MemoryBuffer buffer, Object obj) {
     }
   }
 
+  public <T> T deserializeTyped(byte[] bytes) {
+    Fury fury = null;
+    try {
+      fury = furyPooledObjectFactory.getFury();
+      return fury.deserializeTyped(bytes);
+    } finally {
+      furyPooledObjectFactory.returnFury(fury);
+    }
+  }
+
   public Object deserialize(byte[] bytes) {
     Fury fury = null;
     try {

java/fury-core/src/test/java/io/fury/FuryTest.java

@@ -535,4 +535,18 @@ public void testJavaOutputStream() throws IOException {
       assertEquals(newObj, beanA);
     }
   }
+
+  @Test
+  public void testDeserializationAutotypeCast() {
+    Fury fury = getJavaFury();
+    byte[] bytes = fury.serialize(1);
+    Integer i = fury.deserializeTyped(bytes);
+    Assert.assertEquals(i, 1);
+    Assert.assertThrows(
+        ClassCastException.class,
+        () -> {
+          String str = fury.deserializeTyped(bytes);
+          System.out.println(str);
+        });
+  }
 }

java/fury-core/src/test/java/io/fury/ThreadSafeFuryTest.java

@@ -86,6 +86,7 @@ public void testSerialize() throws Exception {
               try {
                 fury.setClassLoader(beanA.getClass().getClassLoader());
                 assertEquals(fury.deserialize(fury.serialize(beanA)), beanA);
+                assertEquals(fury.deserializeTyped(fury.serialize(beanA)), beanA);
               } catch (Exception e) {
                 hasException = true;
                 e.printStackTrace();