dear:
   请教两个问题
1)  用下面的代码消费kafka 发生序列化异常时,会发生JOB反复重试,重启后也是这样,
改用FlinkKafkaConsumer010类的话,有相关的解决方法,参照https://stackoverflow.com/questions/51301549/how-to-handle-exception-while-parsing-json-in-flink/51302225
不知道,用Kafka类的话,如何解决
.connect(
    new Kafka()
      .version("0.10")
      .topic("test-input")
2)  对于timestamp类型字段,用JDBCAppendTableSink 
把DataStream<Row>写入到mysql时,会发下面的错误LocalTimeStamp到Timestamp的转型错误
    kafka消息是avro格式,字段类型设置为timestamp(3),我是把System.currentTimeMillis()写入到kafka中的
    jdbc参数类型设置为Types.SQL_TIMESTAMP
thanks

回复