|
@@ -267,22 +267,8 @@ public final class KafkaContextHolder {
|
|
|
DelayRedisQueue<String> queue = getTransactionQueue();
|
|
|
queue.offer(message, Duration.ofSeconds(60));
|
|
|
|
|
|
- // 处理本地事务
|
|
|
- boolean committable;
|
|
|
- try {
|
|
|
- committable = callback.test(id);
|
|
|
- } catch (Exception e) {
|
|
|
- // 本地事务处理异常则删除事务延时消息
|
|
|
- try {
|
|
|
- queue.remove(message);
|
|
|
- } catch (Throwable t) {
|
|
|
- log.error("Kafka transaction message remove failed: {}", message, t);
|
|
|
- }
|
|
|
- throw e;
|
|
|
- }
|
|
|
-
|
|
|
- // 本地事务处理成功则发送Kafka消息
|
|
|
- if (committable) {
|
|
|
+ // 处理本地事务并发送消息
|
|
|
+ if (callback.test(id)) {
|
|
|
KafkaTemplate<K, V> template = getKafkaTemplate();
|
|
|
try {
|
|
|
template.send(topic, partition, ordering, payload).get();
|