[FLINK-37735][serializer] Replace class.newInstance with constructor.newInstance

Author: X-czh

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java

@@ -31,6 +31,7 @@
 import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
+import java.lang.reflect.Constructor;
 import java.lang.reflect.Field;
 import java.lang.reflect.Modifier;
 import java.util.Arrays;
@@ -61,6 +62,11 @@ public final class PojoSerializer<T> extends TypeSerializer<T> {
     /** The POJO type class. */
     private final Class<T> clazz;
 
+    /**
+     * The cached constructor, which is not serializable and kept as a separate transient member.
+     */
+    private transient Constructor<T> constructor;
+
     /**
      * Fields of the POJO and their serializers.
      *
@@ -212,14 +218,26 @@ public T createInstance() {
             return null;
         }
         try {
-            T t = clazz.newInstance();
+            T t = instantiateRaw();
             initializeFields(t);
             return t;
         } catch (Exception e) {
             throw new RuntimeException("Cannot instantiate class.", e);
         }
     }
 
+    private T instantiateRaw() {
+        try {
+            if (constructor == null) {
+                constructor = clazz.getDeclaredConstructor();
+                constructor.setAccessible(true);
+            }
+            return constructor.newInstance();
+        } catch (Exception e) {
+            throw new RuntimeException("Cannot instantiate class.", e);
+        }
+    }
+
     protected void initializeFields(T t) {
         for (int i = 0; i < numFields; i++) {
             if (fields[i] != null) {
@@ -256,7 +274,7 @@ public T copy(T from) {
         } else if (actualType == clazz) {
             T target;
             try {
-                target = (T) from.getClass().newInstance();
+                target = instantiateRaw();
             } catch (Throwable t) {
                 throw new RuntimeException("Cannot instantiate class.", t);
             }

flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializer.java

@@ -27,12 +27,18 @@
 import org.apache.flink.types.NullFieldException;
 
 import java.io.IOException;
+import java.lang.reflect.Constructor;
 
 @Internal
 public class TupleSerializer<T extends Tuple> extends TupleSerializerBase<T> {
 
     private static final long serialVersionUID = 1L;
 
+    /**
+     * The cached constructor, which is not serializable and kept as a separate transient member.
+     */
+    private transient Constructor<T> constructor;
+
     public TupleSerializer(Class<T> tupleClass, TypeSerializer<?>[] fieldSerializers) {
         super(tupleClass, fieldSerializers);
     }
@@ -61,7 +67,7 @@ public TupleSerializer<T> duplicate() {
     @Override
     public T createInstance() {
         try {
-            T t = tupleClass.newInstance();
+            T t = instantiateRaw();
 
             for (int i = 0; i < arity; i++) {
                 t.setField(fieldSerializers[i].createInstance(), i);
@@ -77,7 +83,7 @@ public T createInstance() {
     public T createInstance(Object[] fields) {
 
         try {
-            T t = tupleClass.newInstance();
+            T t = instantiateRaw();
 
             for (int i = 0; i < arity; i++) {
                 t.setField(fields[i], i);
@@ -163,7 +169,11 @@ public TypeSerializerSnapshot<T> snapshotConfiguration() {
 
     private T instantiateRaw() {
         try {
-            return tupleClass.newInstance();
+            if (constructor == null) {
+                constructor = tupleClass.getDeclaredConstructor();
+                constructor.setAccessible(true);
+            }
+            return constructor.newInstance();
         } catch (Exception e) {
             throw new RuntimeException("Cannot instantiate tuple.", e);
         }