autumn/JChargePointProtocol
1
0
Fork 0
mirror of https://gitee.com/san-bing/JChargePointProtocol synced 2026-05-04 01:49:58 +08:00
Code Issues Packages Projects Releases Wiki Activity

优化逻辑

Browse Source
This commit is contained in:
三丙
2025-06-25 23:32:32 +08:00
parent 47e1b5d9d0
commit 412ae4d6d4
1 changed files with 10 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
jcpp-protocol-api/src/main/java/sanbing/jcpp/protocol/forwarder/KafkaForwarder.java
View File

@@ -180,29 +180,22 @@ public class KafkaForwarder extends Forwarder {
log.debug("Kafka 消息转发完成, success:{}", e == null);
if (consumer != null) {
onComplete(metadata, e, consumer);
}
}
private void onComplete(RecordMetadata metadata, Exception e, BiConsumer<Boolean, ObjectNode> consumer) {
if (consumer == null) {
return;
}
ObjectNode objectNode = JacksonUtil.newObjectNode();
objectNode.put(OFFSET, String.valueOf(metadata.offset()));
objectNode.put(PARTITION, String.valueOf(metadata.partition()));
objectNode.put(TOPIC, metadata.topic());
if (e != null) {
objectNode.put(ERROR, e.getClass() + ": " + e.getMessage());
forwarderMessagesStats.incrementFailed();
} else {
forwarderMessagesStats.incrementSuccessful();
}
consumer.accept(e == null, objectNode);
if (consumer != null) {
ObjectNode objectNode = JacksonUtil.newObjectNode();
if (e != null) {
objectNode.put(ERROR, e.getClass() + ": " + e.getMessage());
}
objectNode.put(OFFSET, String.valueOf(metadata.offset()));
objectNode.put(PARTITION, String.valueOf(metadata.partition()));
objectNode.put(TOPIC, metadata.topic());
consumer.accept(e == null, objectNode);
}
}
}