[
https://issues.apache.org/jira/browse/FLINK-29647?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17628074#comment-17628074
]
Xintong Song commented on FLINK-29647:
--------------------------------------
[~jackin853],
As [~zhuzh] mentioned previously, you can implement your own serializer for
your classes.
To be specific, you would need to implement the {{TypeSerializer}} interface,
defining how your {{Connection}} class should be serialized/deserialized, and
replacing the `KryoSerializer` with the one you implemented in the following
code block.
{code:java}
public MySqlTwoPhaseCommitSink(){
super(new KryoSerializer<>(Connection.class,new ExecutionConfig()),
VoidSerializer.INSTANCE);
}
{code}
See [this
documentation|https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/fault-tolerance/serialization/custom_serialization/]
for more details.
> report stackoverflow when using kryo
> ------------------------------------
>
> Key: FLINK-29647
> URL: https://issues.apache.org/jira/browse/FLINK-29647
> Project: Flink
> Issue Type: Bug
> Components: API / Type Serialization System
> Affects Versions: 1.13.2
> Environment: flink 1.13.2 version （kryo 2.24 version）
> Reporter: Gao Fei
> Priority: Major
> Labels: KryoSerializer
>
> When using kryo to report stackoverflow, the error is as follows:
> {code:java}
> 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)
> {code}
> I am using two-phase commit to write data to mysql, the following is part of
> the mysql sink code:
> {code:java}
> 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);
> }
> }{code}
> I also found similar problem reports:
> https://github.com/EsotericSoftware/kryo/issues/341
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
