kafka升级或者异步写入kafka失败，丢失消息

[mujianping]

[mujianping]

这块代码应该使用fileNodeKey对应的应该是sourceKey，否则在KafkaCallBack中写入的都是fileKey，在flush的时候进行rollback， for (String key : failedOffsetMap.keySet()) { WorkingFileNode wfn = ((LogSource) source).getCollectingFileNodeMap().get(key); LOGGER.info("rollback !wfn is " + wfn); if (wfn != null) { wfn.seek(failedOffsetMap.get(key)); } } 这个wfn始终是null，会丢消息