canal ddl语句重新从之前执行过的时间再次执行一遍失败

需要在哪里过滤掉已经消费过的执行情况

2017-12-22 14:12:47,365 (main) [INFO - com.trcloud.hamal.stream.core.operator.impl.canal.CanalSource.parseColumns(CanalSource.java:238)] parse ddl sql : ALTER TABLE account_user_info_stream_test ADD COLUMN test001 varchar(255) NOT NULL AFTER modifiedon 2017-12-22 14:12:47,375 (main) [INFO - com.trcloud.hamal.stream.core.operator.impl.canal.CanalSource.parseColumns(CanalSource.java:336)] ddl result map : {"test001":{"curName":"test001","key":false,"nulled":false,"oldName":"","operPimay":false,"operType":"ADD","type":"varchar(255)"}} 2017-12-22 14:12:47,390 (main) [INFO - com.trcloud.hamal.stream.core.operator.impl.canal.Canal2SqlOperator.transform(Canal2SqlOperator.java:77)] sqlDecorators: [{"eventType":"ALTER","sql":"ALTER TABLE test.account_user_info_stream_test ADD COLUMN test001 varchar(255) NOT NULL DEFAULT '' ;","storm_time":"2017-12-22 14:11:23","tableName":"account_user_info_stream_test"}] 2017-12-22 14:12:47,432 (main) [WARN - com.trcloud.hamal.stream.core.operator.impl.sink.GPSinkOperator.process(GPSinkOperator.java:92)] batch exec error, try exec one by one 2017-12-22 14:12:47,433 (main) [INFO - com.trcloud.hamal.stream.core.operator.impl.sink.GPSinkOperator.processAutoCommit(GPSinkOperator.java:141)] current sql is: ALTER TABLE test.account_user_info_stream_test ADD COLUMN test001 varchar(255) NOT NULL DEFAULT '' ; 2017-12-22 14:12:47,465 (main) [ERROR - com.trcloud.hamal.stream.core.operator.impl.sink.GPSinkOperator.processAutoCommit(GPSinkOperator.java:168)] exec sql one by one failed: org.postgresql.util.PSQLException: ERROR: column "test001" of relation "account_user_info_stream_test" already exists at org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2182) at org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:1911) at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:173) at org.postgresql.jdbc.PgStatement.execute(PgStatement.java:622) at org.postgresql.jdbc.PgStatement.executeWithFlags(PgStatement.java:458) at org.postgresql.jdbc.PgStatement.executeUpdate(PgStatement.java:406) at com.zaxxer.hikari.pool.ProxyStatement.executeUpdate(ProxyStatement.java:120) at com.zaxxer.hikari.pool.HikariProxyStatement.executeUpdate(HikariProxyStatement.java) at com.trcloud.hamal.stream.core.operator.impl.sink.GPSinkOperator.processAutoCommit(GPSinkOperator.java:142) at com.trcloud.hamal.stream.core.operator.impl.sink.GPSinkOperator.process(GPSinkOperator.java:95) at com.trcloud.hamal.stream.core.operator.impl.canal.CanalOperatorChain.start(CanalOperatorChain.java:49) at com.trcloud.hamal.stream.core.StreamJob.start(StreamJob.java:121) at com.trcloud.hamal.stream.core.JobRunner.main(JobRunner.java:46) 2017-12-22 14:12:47,468 (main) [ERROR - com.trcloud.hamal.stream.core.operator.impl.sink.GPSinkOperator.process(GPSinkOperator.java:97)] exec sql one by one failed: 200001: sink sql failed! 2017-12-22 14:12:49,165 (main) [ERROR - com.trcloud.hamal.stream.core.operator.impl.canal.CanalOperatorChain.start(CanalOperatorChain.java:51)] 200001: sink sql failed!

原提问者GitHub用户liuwenfeng554