From 30fde947038460a05308a5584d277f2d4e07d4ab Mon Sep 17 00:00:00 2001 From: hqbfzwang Date: Wed, 22 Jul 2026 14:58:23 +0800 Subject: [PATCH 1/4] fix: Completely remove the dependency of fastjson1 --- dledger/pom.xml | 11 +++++++++++ .../storage/dledger/DLedgerEntryPusher.java | 2 +- .../storage/dledger/DLedgerLeaderElector.java | 2 +- .../storage/dledger/DLedgerRpcNettyService.java | 2 +- .../dledger/client/DLedgerClientRpcNettyService.java | 2 +- .../dledger/snapshot/file/FileSnapshotReader.java | 2 +- .../dledger/snapshot/file/FileSnapshotWriter.java | 2 +- .../storage/dledger/snapshot/SnapshotManagerTest.java | 2 +- .../storage/dledger/snapshot/SnapshotReaderTest.java | 3 +-- .../storage/dledger/snapshot/SnapshotStoreTest.java | 1 - .../storage/dledger/snapshot/SnapshotWriterTest.java | 2 +- .../dledger/statemachine/StateMachineCallerTest.java | 2 +- .../dledger/example/appender/AppenderDLedger.java | 2 +- .../example/appender/command/AppendCommand.java | 2 +- .../dledger/example/appender/command/GetCommand.java | 2 +- .../common/command/LeadershipTransferCommand.java | 2 +- .../example/common/command/ReadFileCommand.java | 2 +- .../dledger/example/register/command/ReadCommand.java | 2 +- pom.xml | 2 +- 19 files changed, 28 insertions(+), 19 deletions(-) diff --git a/dledger/pom.xml b/dledger/pom.xml index bb519545..26d32ad8 100644 --- a/dledger/pom.xml +++ b/dledger/pom.xml @@ -30,6 +30,17 @@ org.apache.rocketmq rocketmq-remoting + + + com.alibaba + fastjson + + + + + com.alibaba.fastjson2 + fastjson2 + 2.0.59 org.slf4j diff --git a/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerEntryPusher.java b/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerEntryPusher.java index 5682fd57..0c715659 100644 --- a/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerEntryPusher.java +++ b/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerEntryPusher.java @@ -16,7 +16,7 @@ package io.openmessaging.storage.dledger; -import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson2.JSON; import io.openmessaging.storage.dledger.common.Closure; import io.openmessaging.storage.dledger.common.ShutdownAbleThread; import io.openmessaging.storage.dledger.common.Status; diff --git a/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerLeaderElector.java b/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerLeaderElector.java index e23bff61..e99eb429 100644 --- a/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerLeaderElector.java +++ b/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerLeaderElector.java @@ -16,7 +16,7 @@ package io.openmessaging.storage.dledger; -import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson2.JSON; import io.openmessaging.storage.dledger.common.ShutdownAbleThread; import io.openmessaging.storage.dledger.protocol.DLedgerResponseCode; import io.openmessaging.storage.dledger.protocol.HeartBeatRequest; diff --git a/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerRpcNettyService.java b/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerRpcNettyService.java index c53fda36..1e93f363 100644 --- a/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerRpcNettyService.java +++ b/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerRpcNettyService.java @@ -16,7 +16,7 @@ package io.openmessaging.storage.dledger; -import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson2.JSON; import io.netty.channel.ChannelHandlerContext; import io.openmessaging.storage.dledger.common.NamedThreadFactory; import io.openmessaging.storage.dledger.protocol.AppendEntryRequest; diff --git a/dledger/src/main/java/io/openmessaging/storage/dledger/client/DLedgerClientRpcNettyService.java b/dledger/src/main/java/io/openmessaging/storage/dledger/client/DLedgerClientRpcNettyService.java index 57d2a703..1664acaa 100644 --- a/dledger/src/main/java/io/openmessaging/storage/dledger/client/DLedgerClientRpcNettyService.java +++ b/dledger/src/main/java/io/openmessaging/storage/dledger/client/DLedgerClientRpcNettyService.java @@ -16,7 +16,7 @@ package io.openmessaging.storage.dledger.client; -import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson2.JSON; import io.openmessaging.storage.dledger.protocol.AppendEntryRequest; import io.openmessaging.storage.dledger.protocol.AppendEntryResponse; import io.openmessaging.storage.dledger.protocol.DLedgerRequestCode; diff --git a/dledger/src/main/java/io/openmessaging/storage/dledger/snapshot/file/FileSnapshotReader.java b/dledger/src/main/java/io/openmessaging/storage/dledger/snapshot/file/FileSnapshotReader.java index cd4427c5..69fec25b 100644 --- a/dledger/src/main/java/io/openmessaging/storage/dledger/snapshot/file/FileSnapshotReader.java +++ b/dledger/src/main/java/io/openmessaging/storage/dledger/snapshot/file/FileSnapshotReader.java @@ -16,7 +16,7 @@ package io.openmessaging.storage.dledger.snapshot.file; -import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson2.JSON; import io.openmessaging.storage.dledger.snapshot.DownloadSnapshot; import io.openmessaging.storage.dledger.snapshot.SnapshotManager; import io.openmessaging.storage.dledger.snapshot.SnapshotMeta; diff --git a/dledger/src/main/java/io/openmessaging/storage/dledger/snapshot/file/FileSnapshotWriter.java b/dledger/src/main/java/io/openmessaging/storage/dledger/snapshot/file/FileSnapshotWriter.java index 5009cdd3..20302bd0 100644 --- a/dledger/src/main/java/io/openmessaging/storage/dledger/snapshot/file/FileSnapshotWriter.java +++ b/dledger/src/main/java/io/openmessaging/storage/dledger/snapshot/file/FileSnapshotWriter.java @@ -16,7 +16,7 @@ package io.openmessaging.storage.dledger.snapshot.file; -import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson2.JSON; import io.openmessaging.storage.dledger.snapshot.SnapshotManager; import io.openmessaging.storage.dledger.snapshot.SnapshotMeta; import io.openmessaging.storage.dledger.snapshot.SnapshotStatus; diff --git a/dledger/src/test/java/io/openmessaging/storage/dledger/snapshot/SnapshotManagerTest.java b/dledger/src/test/java/io/openmessaging/storage/dledger/snapshot/SnapshotManagerTest.java index f570d12d..4dd72205 100644 --- a/dledger/src/test/java/io/openmessaging/storage/dledger/snapshot/SnapshotManagerTest.java +++ b/dledger/src/test/java/io/openmessaging/storage/dledger/snapshot/SnapshotManagerTest.java @@ -1,6 +1,6 @@ package io.openmessaging.storage.dledger.snapshot; -import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson2.JSON; import io.openmessaging.storage.dledger.DLedgerConfig; import io.openmessaging.storage.dledger.DLedgerServer; import io.openmessaging.storage.dledger.MemberState; diff --git a/dledger/src/test/java/io/openmessaging/storage/dledger/snapshot/SnapshotReaderTest.java b/dledger/src/test/java/io/openmessaging/storage/dledger/snapshot/SnapshotReaderTest.java index f538351b..2081019a 100644 --- a/dledger/src/test/java/io/openmessaging/storage/dledger/snapshot/SnapshotReaderTest.java +++ b/dledger/src/test/java/io/openmessaging/storage/dledger/snapshot/SnapshotReaderTest.java @@ -1,6 +1,6 @@ package io.openmessaging.storage.dledger.snapshot; -import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson2.JSON; import io.openmessaging.storage.dledger.snapshot.file.FileSnapshotReader; import io.openmessaging.storage.dledger.util.FileTestUtil; import io.openmessaging.storage.dledger.utils.IOUtils; @@ -9,7 +9,6 @@ import java.io.File; import java.io.IOException; -import java.util.UUID; public class SnapshotReaderTest { diff --git a/dledger/src/test/java/io/openmessaging/storage/dledger/snapshot/SnapshotStoreTest.java b/dledger/src/test/java/io/openmessaging/storage/dledger/snapshot/SnapshotStoreTest.java index 51d61704..12f5b7ac 100644 --- a/dledger/src/test/java/io/openmessaging/storage/dledger/snapshot/SnapshotStoreTest.java +++ b/dledger/src/test/java/io/openmessaging/storage/dledger/snapshot/SnapshotStoreTest.java @@ -1,6 +1,5 @@ package io.openmessaging.storage.dledger.snapshot; -import com.alibaba.fastjson.JSON; import io.openmessaging.storage.dledger.snapshot.file.FileSnapshotStore; import io.openmessaging.storage.dledger.util.FileTestUtil; import io.openmessaging.storage.dledger.utils.IOUtils; diff --git a/dledger/src/test/java/io/openmessaging/storage/dledger/snapshot/SnapshotWriterTest.java b/dledger/src/test/java/io/openmessaging/storage/dledger/snapshot/SnapshotWriterTest.java index 21df4363..8a46cdb9 100644 --- a/dledger/src/test/java/io/openmessaging/storage/dledger/snapshot/SnapshotWriterTest.java +++ b/dledger/src/test/java/io/openmessaging/storage/dledger/snapshot/SnapshotWriterTest.java @@ -1,6 +1,6 @@ package io.openmessaging.storage.dledger.snapshot; -import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson2.JSON; import io.openmessaging.storage.dledger.snapshot.file.FileSnapshotStore; import io.openmessaging.storage.dledger.snapshot.file.FileSnapshotWriter; import io.openmessaging.storage.dledger.util.FileTestUtil; diff --git a/dledger/src/test/java/io/openmessaging/storage/dledger/statemachine/StateMachineCallerTest.java b/dledger/src/test/java/io/openmessaging/storage/dledger/statemachine/StateMachineCallerTest.java index b9e5a0ac..ae35b077 100644 --- a/dledger/src/test/java/io/openmessaging/storage/dledger/statemachine/StateMachineCallerTest.java +++ b/dledger/src/test/java/io/openmessaging/storage/dledger/statemachine/StateMachineCallerTest.java @@ -16,6 +16,7 @@ package io.openmessaging.storage.dledger.statemachine; +import com.alibaba.fastjson2.JSON; import io.openmessaging.storage.dledger.common.Status; import io.openmessaging.storage.dledger.common.WriteClosure; import io.openmessaging.storage.dledger.common.WriteTask; @@ -25,7 +26,6 @@ import java.util.UUID; import java.util.concurrent.CountDownLatch; -import com.alibaba.fastjson.JSON; import io.openmessaging.storage.dledger.DLedgerConfig; import io.openmessaging.storage.dledger.DLedgerServer; import io.openmessaging.storage.dledger.MemberState; diff --git a/example/src/main/java/io/openmessaging/storage/dledger/example/appender/AppenderDLedger.java b/example/src/main/java/io/openmessaging/storage/dledger/example/appender/AppenderDLedger.java index cc3fad80..6785e74b 100644 --- a/example/src/main/java/io/openmessaging/storage/dledger/example/appender/AppenderDLedger.java +++ b/example/src/main/java/io/openmessaging/storage/dledger/example/appender/AppenderDLedger.java @@ -16,7 +16,7 @@ package io.openmessaging.storage.dledger.example.appender; -import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson2.JSON; import io.openmessaging.storage.dledger.DLedgerConfig; import io.openmessaging.storage.dledger.proxy.DLedgerProxy; import java.util.Collections; diff --git a/example/src/main/java/io/openmessaging/storage/dledger/example/appender/command/AppendCommand.java b/example/src/main/java/io/openmessaging/storage/dledger/example/appender/command/AppendCommand.java index 4bce09ad..7959d489 100644 --- a/example/src/main/java/io/openmessaging/storage/dledger/example/appender/command/AppendCommand.java +++ b/example/src/main/java/io/openmessaging/storage/dledger/example/appender/command/AppendCommand.java @@ -16,7 +16,7 @@ package io.openmessaging.storage.dledger.example.appender.command; -import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson2.JSON; import com.beust.jcommander.Parameter; import com.beust.jcommander.Parameters; import io.openmessaging.storage.dledger.client.DLedgerClient; diff --git a/example/src/main/java/io/openmessaging/storage/dledger/example/appender/command/GetCommand.java b/example/src/main/java/io/openmessaging/storage/dledger/example/appender/command/GetCommand.java index 87ebf407..283b5326 100644 --- a/example/src/main/java/io/openmessaging/storage/dledger/example/appender/command/GetCommand.java +++ b/example/src/main/java/io/openmessaging/storage/dledger/example/appender/command/GetCommand.java @@ -16,7 +16,7 @@ package io.openmessaging.storage.dledger.example.appender.command; -import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson2.JSON; import com.beust.jcommander.Parameter; import com.beust.jcommander.Parameters; import io.openmessaging.storage.dledger.client.DLedgerClient; diff --git a/example/src/main/java/io/openmessaging/storage/dledger/example/common/command/LeadershipTransferCommand.java b/example/src/main/java/io/openmessaging/storage/dledger/example/common/command/LeadershipTransferCommand.java index 3b004f86..54d3b3e8 100644 --- a/example/src/main/java/io/openmessaging/storage/dledger/example/common/command/LeadershipTransferCommand.java +++ b/example/src/main/java/io/openmessaging/storage/dledger/example/common/command/LeadershipTransferCommand.java @@ -16,7 +16,7 @@ package io.openmessaging.storage.dledger.example.common.command; -import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson2.JSON; import com.beust.jcommander.Parameter; import com.beust.jcommander.Parameters; import io.openmessaging.storage.dledger.client.DLedgerClient; diff --git a/example/src/main/java/io/openmessaging/storage/dledger/example/common/command/ReadFileCommand.java b/example/src/main/java/io/openmessaging/storage/dledger/example/common/command/ReadFileCommand.java index d7872672..ac9a9481 100644 --- a/example/src/main/java/io/openmessaging/storage/dledger/example/common/command/ReadFileCommand.java +++ b/example/src/main/java/io/openmessaging/storage/dledger/example/common/command/ReadFileCommand.java @@ -16,7 +16,7 @@ package io.openmessaging.storage.dledger.example.common.command; -import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson2.JSON; import com.beust.jcommander.Parameter; import com.beust.jcommander.Parameters; import io.openmessaging.storage.dledger.entry.DLedgerEntry; diff --git a/example/src/main/java/io/openmessaging/storage/dledger/example/register/command/ReadCommand.java b/example/src/main/java/io/openmessaging/storage/dledger/example/register/command/ReadCommand.java index 924cbbab..ed1814b7 100644 --- a/example/src/main/java/io/openmessaging/storage/dledger/example/register/command/ReadCommand.java +++ b/example/src/main/java/io/openmessaging/storage/dledger/example/register/command/ReadCommand.java @@ -16,7 +16,7 @@ package io.openmessaging.storage.dledger.example.register.command; -import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson2.JSON; import com.beust.jcommander.Parameter; import com.beust.jcommander.Parameters; import io.openmessaging.storage.dledger.common.ReadMode; diff --git a/pom.xml b/pom.xml index 061a743d..a2bd68eb 100644 --- a/pom.xml +++ b/pom.xml @@ -42,7 +42,7 @@ 1.7.36 1.30 1.72 - 5.1.0 + 5.5.0 jacoco 3.18.0 1.26.0 From 9bb081c1293ec6d082400b962938a46f56119b53 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 22 Jul 2026 08:16:01 +0000 Subject: [PATCH 2/4] Initial plan From a9d6e8bbad378b6439ab8d6fb5f55e091034cc44 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 22 Jul 2026 08:55:26 +0000 Subject: [PATCH 3/4] fix: handle InvokeCallback.operationFail() for rocketmq-remoting 5.5.0 compatibility rocketmq-remoting 5.5.0 changed InvokeCallback semantics: operationFail() is now called directly when connections fail (null channel, inactive channel, connection timeout). DLedger's lambda callbacks only implemented operationComplete(), using the default no-op operationFail(). This caused CompletableFutures to never complete when connections failed, leading to TimeoutException after 3 seconds and test failures across LeaderElectorTest, AppendAndPushTest, BatchPushTest, CommitIndexTest, and SnapshotManagerTest. Fix: replace all lambda InvokeCallback instances in DLedgerRpcNettyService with anonymous classes that implement both operationComplete() and operationFail(), immediately completing futures with NETWORK_ERROR on failure. --- .../dledger/DLedgerRpcNettyService.java | 160 ++++++++++++------ 1 file changed, 110 insertions(+), 50 deletions(-) diff --git a/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerRpcNettyService.java b/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerRpcNettyService.java index 1e93f363..2c85d516 100644 --- a/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerRpcNettyService.java +++ b/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerRpcNettyService.java @@ -52,11 +52,13 @@ import java.util.concurrent.Executors; import org.apache.rocketmq.remoting.ChannelEventListener; +import org.apache.rocketmq.remoting.InvokeCallback; import org.apache.rocketmq.remoting.netty.NettyClientConfig; import org.apache.rocketmq.remoting.netty.NettyRemotingClient; import org.apache.rocketmq.remoting.netty.NettyRemotingServer; import org.apache.rocketmq.remoting.netty.NettyRequestProcessor; import org.apache.rocketmq.remoting.netty.NettyServerConfig; +import org.apache.rocketmq.remoting.netty.ResponseFuture; import org.apache.rocketmq.remoting.protocol.RemotingCommand; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -156,13 +158,22 @@ public CompletableFuture heartBeat(HeartBeatRequest request) try { RemotingCommand wrapperRequest = RemotingCommand.createRequestCommand(DLedgerRequestCode.HEART_BEAT.getCode(), null); wrapperRequest.setBody(JSON.toJSONBytes(request)); - remotingClient.invokeAsync(getPeerAddr(request.getGroup(), request.getRemoteId()), wrapperRequest, 3000, responseFuture -> { - RemotingCommand responseCommand = responseFuture.getResponseCommand(); - if (responseCommand != null) { - HeartBeatResponse response = JSON.parseObject(responseCommand.getBody(), HeartBeatResponse.class); - future.complete(response); - } else { - LOGGER.error("HeartBeat request time out, {}", request.baseInfo()); + remotingClient.invokeAsync(getPeerAddr(request.getGroup(), request.getRemoteId()), wrapperRequest, 3000, new InvokeCallback() { + @Override + public void operationComplete(ResponseFuture responseFuture) { + RemotingCommand responseCommand = responseFuture.getResponseCommand(); + if (responseCommand != null) { + HeartBeatResponse response = JSON.parseObject(responseCommand.getBody(), HeartBeatResponse.class); + future.complete(response); + } else { + LOGGER.error("HeartBeat request time out, {}", request.baseInfo()); + future.complete(new HeartBeatResponse().code(DLedgerResponseCode.NETWORK_ERROR.getCode())); + } + } + + @Override + public void operationFail(Throwable throwable) { + LOGGER.error("HeartBeat request failed, {}", request.baseInfo(), throwable); future.complete(new HeartBeatResponse().code(DLedgerResponseCode.NETWORK_ERROR.getCode())); } }); @@ -181,13 +192,22 @@ public CompletableFuture vote(VoteRequest request) { try { RemotingCommand wrapperRequest = RemotingCommand.createRequestCommand(DLedgerRequestCode.VOTE.getCode(), null); wrapperRequest.setBody(JSON.toJSONBytes(request)); - this.remotingClient.invokeAsync(getPeerAddr(request.getGroup(), request.getRemoteId()), wrapperRequest, 3000, responseFuture -> { - RemotingCommand responseCommand = responseFuture.getResponseCommand(); - if (responseCommand != null) { - VoteResponse response = JSON.parseObject(responseCommand.getBody(), VoteResponse.class); - future.complete(response); - } else { - LOGGER.error("Vote request time out, {}", request.baseInfo()); + this.remotingClient.invokeAsync(getPeerAddr(request.getGroup(), request.getRemoteId()), wrapperRequest, 3000, new InvokeCallback() { + @Override + public void operationComplete(ResponseFuture responseFuture) { + RemotingCommand responseCommand = responseFuture.getResponseCommand(); + if (responseCommand != null) { + VoteResponse response = JSON.parseObject(responseCommand.getBody(), VoteResponse.class); + future.complete(response); + } else { + LOGGER.error("Vote request time out, {}", request.baseInfo()); + future.complete(new VoteResponse()); + } + } + + @Override + public void operationFail(Throwable throwable) { + LOGGER.error("Vote request failed, {}", request.baseInfo(), throwable); future.complete(new VoteResponse()); } }); @@ -212,18 +232,28 @@ public CompletableFuture append(AppendEntryRequest request) try { RemotingCommand wrapperRequest = RemotingCommand.createRequestCommand(DLedgerRequestCode.APPEND.getCode(), null); wrapperRequest.setBody(JSON.toJSONBytes(request)); - remotingClient.invokeAsync(getPeerAddr(request.getGroup(), request.getRemoteId()), wrapperRequest, 3000, responseFuture -> { - RemotingCommand responseCommand = responseFuture.getResponseCommand(); - - AppendEntryResponse response; - if (responseCommand != null) { - response = JSON.parseObject(responseCommand.getBody(), AppendEntryResponse.class); - } else { - response = new AppendEntryResponse(); + remotingClient.invokeAsync(getPeerAddr(request.getGroup(), request.getRemoteId()), wrapperRequest, 3000, new InvokeCallback() { + @Override + public void operationComplete(ResponseFuture responseFuture) { + RemotingCommand responseCommand = responseFuture.getResponseCommand(); + AppendEntryResponse response; + if (responseCommand != null) { + response = JSON.parseObject(responseCommand.getBody(), AppendEntryResponse.class); + } else { + response = new AppendEntryResponse(); + response.copyBaseInfo(request); + response.setCode(DLedgerResponseCode.NETWORK_ERROR.getCode()); + } + future.complete(response); + } + + @Override + public void operationFail(Throwable throwable) { + AppendEntryResponse response = new AppendEntryResponse(); response.copyBaseInfo(request); response.setCode(DLedgerResponseCode.NETWORK_ERROR.getCode()); + future.complete(response); } - future.complete(response); }); } catch (Throwable t) { LOGGER.error("Send append request failed, {}", request.baseInfo(), t); @@ -257,18 +287,28 @@ public CompletableFuture push(PushEntryRequest request) { try { RemotingCommand wrapperRequest = RemotingCommand.createRequestCommand(DLedgerRequestCode.PUSH.getCode(), null); wrapperRequest.setBody(JSON.toJSONBytes(request)); - remotingClient.invokeAsync(getPeerAddr(request.getGroup(), request.getRemoteId()), wrapperRequest, 3000, responseFuture -> { - RemotingCommand responseCommand = responseFuture.getResponseCommand(); - - PushEntryResponse response; - if (responseCommand != null) { - response = JSON.parseObject(responseCommand.getBody(), PushEntryResponse.class); - } else { - response = new PushEntryResponse(); + remotingClient.invokeAsync(getPeerAddr(request.getGroup(), request.getRemoteId()), wrapperRequest, 3000, new InvokeCallback() { + @Override + public void operationComplete(ResponseFuture responseFuture) { + RemotingCommand responseCommand = responseFuture.getResponseCommand(); + PushEntryResponse response; + if (responseCommand != null) { + response = JSON.parseObject(responseCommand.getBody(), PushEntryResponse.class); + } else { + response = new PushEntryResponse(); + response.copyBaseInfo(request); + response.setCode(DLedgerResponseCode.NETWORK_ERROR.getCode()); + } + future.complete(response); + } + + @Override + public void operationFail(Throwable throwable) { + PushEntryResponse response = new PushEntryResponse(); response.copyBaseInfo(request); response.setCode(DLedgerResponseCode.NETWORK_ERROR.getCode()); + future.complete(response); } - future.complete(response); }); } catch (Throwable t) { LOGGER.error("Send push request failed, {}", request.baseInfo(), t); @@ -287,18 +327,28 @@ public CompletableFuture installSnapshot(InstallSnapsho try { RemotingCommand wrapperRequest = RemotingCommand.createRequestCommand(DLedgerRequestCode.INSTALL_SNAPSHOT.getCode(), null); wrapperRequest.setBody(JSON.toJSONBytes(request)); - remotingClient.invokeAsync(getPeerAddr(request.getGroup(), request.getRemoteId()), wrapperRequest, 3000, responseFuture -> { - RemotingCommand responseCommand = responseFuture.getResponseCommand(); - - InstallSnapshotResponse response; - if (responseCommand != null) { - response = JSON.parseObject(responseFuture.getResponseCommand().getBody(), InstallSnapshotResponse.class); - } else { - response = new InstallSnapshotResponse(); + remotingClient.invokeAsync(getPeerAddr(request.getGroup(), request.getRemoteId()), wrapperRequest, 3000, new InvokeCallback() { + @Override + public void operationComplete(ResponseFuture responseFuture) { + RemotingCommand responseCommand = responseFuture.getResponseCommand(); + InstallSnapshotResponse response; + if (responseCommand != null) { + response = JSON.parseObject(responseFuture.getResponseCommand().getBody(), InstallSnapshotResponse.class); + } else { + response = new InstallSnapshotResponse(); + response.copyBaseInfo(request); + response.setCode(DLedgerResponseCode.NETWORK_ERROR.getCode()); + } + future.complete(response); + } + + @Override + public void operationFail(Throwable throwable) { + InstallSnapshotResponse response = new InstallSnapshotResponse(); response.copyBaseInfo(request); response.setCode(DLedgerResponseCode.NETWORK_ERROR.getCode()); + future.complete(response); } - future.complete(response); }); } catch (Throwable t) { LOGGER.error("Send install snapshot request failed, {}", request.baseInfo(), t); @@ -317,18 +367,28 @@ public CompletableFuture leadershipTransfer( try { RemotingCommand wrapperRequest = RemotingCommand.createRequestCommand(DLedgerRequestCode.LEADERSHIP_TRANSFER.getCode(), null); wrapperRequest.setBody(JSON.toJSONBytes(request)); - remotingClient.invokeAsync(getPeerAddr(request.getGroup(), request.getRemoteId()), wrapperRequest, 3000, responseFuture -> { - RemotingCommand responseCommand = responseFuture.getResponseCommand(); - - LeadershipTransferResponse response; - if (responseCommand != null) { - response = JSON.parseObject(responseFuture.getResponseCommand().getBody(), LeadershipTransferResponse.class); - } else { - response = new LeadershipTransferResponse(); + remotingClient.invokeAsync(getPeerAddr(request.getGroup(), request.getRemoteId()), wrapperRequest, 3000, new InvokeCallback() { + @Override + public void operationComplete(ResponseFuture responseFuture) { + RemotingCommand responseCommand = responseFuture.getResponseCommand(); + LeadershipTransferResponse response; + if (responseCommand != null) { + response = JSON.parseObject(responseFuture.getResponseCommand().getBody(), LeadershipTransferResponse.class); + } else { + response = new LeadershipTransferResponse(); + response.copyBaseInfo(request); + response.setCode(DLedgerResponseCode.NETWORK_ERROR.getCode()); + } + future.complete(response); + } + + @Override + public void operationFail(Throwable throwable) { + LeadershipTransferResponse response = new LeadershipTransferResponse(); response.copyBaseInfo(request); response.setCode(DLedgerResponseCode.NETWORK_ERROR.getCode()); + future.complete(response); } - future.complete(response); }); } catch (Throwable t) { LOGGER.error("Send leadershipTransfer request failed, {}", request.baseInfo(), t); From 61f87aebb217d6b2678c3fbd9609a7d78b27ccd4 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 22 Jul 2026 08:57:02 +0000 Subject: [PATCH 4/4] fix: improve log messages for operationFail to distinguish from timeout case --- .../openmessaging/storage/dledger/DLedgerRpcNettyService.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerRpcNettyService.java b/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerRpcNettyService.java index 2c85d516..5f75fa7a 100644 --- a/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerRpcNettyService.java +++ b/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerRpcNettyService.java @@ -173,7 +173,7 @@ public void operationComplete(ResponseFuture responseFuture) { @Override public void operationFail(Throwable throwable) { - LOGGER.error("HeartBeat request failed, {}", request.baseInfo(), throwable); + LOGGER.error("HeartBeat request failed due to network error, {}", request.baseInfo(), throwable); future.complete(new HeartBeatResponse().code(DLedgerResponseCode.NETWORK_ERROR.getCode())); } }); @@ -207,7 +207,7 @@ public void operationComplete(ResponseFuture responseFuture) { @Override public void operationFail(Throwable throwable) { - LOGGER.error("Vote request failed, {}", request.baseInfo(), throwable); + LOGGER.error("Vote request failed due to network error, {}", request.baseInfo(), throwable); future.complete(new VoteResponse()); } });