[FLINK-29647] report stackoverflow when using kryo

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Workaround
• Affects Version/s: 1.13.2
• Component/s: API / Type Serialization System
• Labels: KryoSerializer

Environment

flink 1.13.2 version （kryo 2.24 version）

Language

Java

• Assignee: Unassigned
• Reporter: Gao Fei
• Created: 15/Oct/22 05:56
• Updated: 03/Nov/22 02:17
• Resolved: 27/Oct/22 07:13

Description

When using kryo to report stackoverflow, the error is as follows:

java.lang.StackOverflowError at com.esotericsoftware.kryo.Generics.getConcreteClass(Generics.java:43) at com.esotericsoftware.kryo.Generics.getConcreteClass(Generics.java:44) at com.esotericsoftware.kryo.Generics.getConcreteClass(Generics.java:44) at com.esotericsoftware.kryo.Generics.getConcreteClass(Generics.java:44) at com.esotericsoftware.kryo.Generics.getConcreteClass(Generics.java:44) at com.esotericsoftware.kryo.Generics.getConcreteClass(Generics.java:44) at com.esotericsoftware.kryo.Generics.getConcreteClass(Generics.java:44) at com.esotericsoftware.kryo.Generics.getConcreteClass(Generics.java:44) at 
com.esotericsoftware.kryo.Generics.getConcreteClass(Generics.java:44)

 I am using two-phase commit to write data to mysql, the following is part of the mysql sink code:

public class MySqlTwoPhaseCommitSink extends TwoPhaseCommitSinkFunction<Tuple2<String,Integer>, Connection,Void> {
    private static final Logger log = LoggerFactory.getLogger(MySqlTwoPhaseCommitSink.class);
    public MySqlTwoPhaseCommitSink(){
        super(new KryoSerializer<>(Connection.class,new ExecutionConfig()), VoidSerializer.INSTANCE);
    }
    @Override
    public void invoke(Connection connection, Tuple2<String,Integer> tp, Context context) throws Exception {
        log.info("start invoke...");
        //TODO
        //omit here
    }
    @Override
    public Connection beginTransaction() throws Exception {
        log.info("start beginTransaction.......");
        String url = "jdbc:mysql://localhost:3306/bigdata?useUnicode=true&characterEncoding=UTF-8";
        Connection connection = DBConnectUtil.getConnection(url, "root", "123456");
        return connection;
    }
    @Override
    public void preCommit(Connection connection) throws Exception {
        log.info("start preCommit...");
    }
    @Override
    public void commit(Connection connection) {
        log.info("start commit...");
        DBConnectUtil.commit(connection);
    }
    @Override
    public void abort(Connection connection) {
        log.info("start abort rollback...");
        DBConnectUtil.rollback(connection);
    }
}

I also found similar problem reports: https://github.com/EsotericSoftware/kryo/issues/341