[Java] implement Externalizable serialization

java/fury-core/src/main/java/io/fury/resolver/ClassResolver.java

@@ -30,6 +30,7 @@
 import io.fury.memory.MemoryBuffer;
 import io.fury.serializer.ArraySerializers;
 import io.fury.serializer.BufferSerializers;
+import io.fury.serializer.ExternalizableSerializer;
 import io.fury.serializer.JavaSerializer;
 import io.fury.serializer.JdkProxySerializer;
 import io.fury.serializer.LocaleSerializer;
@@ -559,6 +560,8 @@ public Class<? extends Serializer> getSerializerClass(Class<?> cls) {
         return TimeSerializers.ZoneIdSerializer.class;
       } else if (TimeZone.class.isAssignableFrom(cls)) {
         return TimeSerializers.TimeZoneSerializer.class;
+      } else if (Externalizable.class.isAssignableFrom(cls)) {
+        return ExternalizableSerializer.class;
       }
       if (requireJavaSerialization(cls)) {
         return getJavaSerializer(cls);

java/fury-core/src/main/java/io/fury/serializer/ExternalizableSerializer.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2023 The Fury authors
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.fury.serializer;
+
+import io.fury.Fury;
+import io.fury.io.FuryObjectInput;
+import io.fury.io.FuryObjectOutput;
+import io.fury.memory.MemoryBuffer;
+import io.fury.util.Platform;
+import io.fury.util.Utils;
+import java.io.Externalizable;
+import java.io.IOException;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+
+/**
+ * Serializer for class implements {@link Externalizable}.
+ *
+ * @author chaokunyang
+ */
+public class ExternalizableSerializer<T extends Externalizable> extends Serializer<T> {
+  private Constructor<T> constructor;
+  private final FuryObjectInput objectInput;
+  private final FuryObjectOutput objectOutput;
+
+  public ExternalizableSerializer(Fury fury, Class<T> cls) {
+    super(fury, cls);
+    try {
+      constructor = cls.getConstructor();
+    } catch (NoSuchMethodException e) {
+      Utils.ignore(e);
+    }
+
+    objectInput = new FuryObjectInput(fury, null);
+    objectOutput = new FuryObjectOutput(fury, null);
+  }
+
+  @Override
+  public void write(MemoryBuffer buffer, T value) {
+    objectOutput.setBuffer(buffer);
+    try {
+      value.writeExternal(objectOutput);
+    } catch (IOException e) {
+      Platform.throwException(e);
+    }
+  }
+
+  @Override
+  public void crossLanguageWrite(MemoryBuffer buffer, T value) {
+    throw new UnsupportedOperationException("Externalizable can only be used in java");
+  }
+
+  @Override
+  public T read(MemoryBuffer buffer) {
+    T t;
+    if (constructor != null) {
+      try {
+        t = constructor.newInstance();
+      } catch (InstantiationException | IllegalAccessException | InvocationTargetException e) {
+        throw new RuntimeException(e);
+      }
+    } else {
+      t = Platform.newInstance(type);
+    }
+    objectInput.setBuffer(buffer);
+    try {
+      t.readExternal(objectInput);
+    } catch (IOException | ClassNotFoundException e) {
+      Platform.throwException(e);
+    }
+    return t;
+  }
+}

java/fury-core/src/test/java/io/fury/serializer/ExternalizableSerializerTest.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2023 The Fury authors
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.fury.serializer;
+
+import static org.testng.Assert.assertEquals;
+
+import com.google.common.base.Preconditions;
+import io.fury.Fury;
+import io.fury.Language;
+import java.io.Externalizable;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import lombok.Data;
+import org.testng.annotations.Test;
+
+public class ExternalizableSerializerTest {
+
+  @Data
+  public static class A implements Externalizable {
+    private int x;
+    private int y;
+    private byte[] bytes;
+
+    @Override
+    public void writeExternal(ObjectOutput out) throws IOException {
+      out.writeInt(x);
+      out.writeInt(y);
+      out.writeInt(bytes.length);
+      out.write(bytes);
+    }
+
+    @Override
+    public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+      this.x = in.readInt();
+      this.y = in.readInt();
+      int len = in.readInt();
+      byte[] arr = new byte[len];
+      Preconditions.checkArgument(in.read(arr) == len);
+      this.bytes = arr;
+    }
+  }
+
+  @Test
+  public void testExternalizable() {
+    A a = new A();
+    a.x = 1;
+    a.y = 1;
+    a.bytes = "bytes".getBytes();
+
+    Fury fury =
+        Fury.builder()
+            .withLanguage(Language.JAVA)
+            .withReferenceTracking(false)
+            .disableSecureMode()
+            .build();
+    assertEquals(a, fury.deserialize(fury.serialize(a)));
+  }
+}