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，会丢消息