From a3d5b54d02586491e4f5b440e0dcef96359524df Mon Sep 17 00:00:00 2001 From: w1am <33353798+w1am@users.noreply.github.com> Date: Fri, 6 Nov 2026 20:22:30 +0400 Subject: [PATCH 1/4] fix: reconnect to leader after not-leader rejection on appendRecords and multiStreamAppend --- .../java/io/kurrent/dbclient/GrpcUtils.java | 22 +++++++++++++++ .../dbclient/LeaderRedirectUnitTest.java | 27 +++++++++++++++++++ 2 files changed, 49 insertions(+) diff --git a/src/main/java/io/kurrent/dbclient/GrpcUtils.java b/src/main/java/io/kurrent/dbclient/GrpcUtils.java index 60f5b8c8..ddfd08fd 100644 --- a/src/main/java/io/kurrent/dbclient/GrpcUtils.java +++ b/src/main/java/io/kurrent/dbclient/GrpcUtils.java @@ -2,10 +2,14 @@ import io.kurrent.dbclient.proto.shared.Shared; import io.kurrent.dbclient.proto.streams.StreamsOuterClass; +import io.kurrentdb.protocol.v2.common.errors.NotLeaderNodeErrorDetails; +import com.google.protobuf.Any; import com.google.protobuf.ByteString; +import com.google.protobuf.InvalidProtocolBufferException; import io.grpc.Metadata; import io.grpc.Status; import io.grpc.StatusRuntimeException; +import io.grpc.protobuf.StatusProto; import io.grpc.stub.*; import java.util.concurrent.CompletableFuture; @@ -27,6 +31,24 @@ static boolean handleNotLeaderError(Throwable t, CompletableFuture dest) { return true; } } + + com.google.rpc.Status status = StatusProto.fromThrowable(e); + + if (status != null) { + for (Any detail : status.getDetailsList()) { + if (!detail.is(NotLeaderNodeErrorDetails.class)) + continue; + + try { + NotLeaderNodeErrorDetails.NodeInfo leader = + detail.unpack(NotLeaderNodeErrorDetails.class).getCurrentLeader(); + dest.completeExceptionally(new NotLeaderException(leader.getHost(), leader.getPort())); + return true; + } catch (InvalidProtocolBufferException ignored) { + return false; + } + } + } } return false; diff --git a/src/test/java/io/kurrent/dbclient/LeaderRedirectUnitTest.java b/src/test/java/io/kurrent/dbclient/LeaderRedirectUnitTest.java index 0a37246a..f00a1379 100644 --- a/src/test/java/io/kurrent/dbclient/LeaderRedirectUnitTest.java +++ b/src/test/java/io/kurrent/dbclient/LeaderRedirectUnitTest.java @@ -1,13 +1,19 @@ package io.kurrent.dbclient; +import com.google.protobuf.Any; +import com.google.rpc.ErrorInfo; import io.grpc.Metadata; import io.grpc.Status; import io.grpc.StatusRuntimeException; +import io.grpc.protobuf.StatusProto; +import io.kurrentdb.protocol.v2.common.errors.NotLeaderNodeErrorDetails; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.net.InetSocketAddress; import java.util.UUID; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; @@ -85,4 +91,25 @@ public void onCancelled(Subscription subscription, Throwable exception) { Assertions.assertTrue(cancelErr.get() instanceof NotLeaderException, "listener should receive a NotLeaderException, got: " + cancelErr.get()); } + + @Test + public void handleNotLeaderErrorReadsNotLeaderNodeErrorDetails() { + com.google.rpc.Status status = com.google.rpc.Status.newBuilder() + .setCode(Status.Code.FAILED_PRECONDITION.value()) + .setMessage("The server is not the leader node and cannot handle the request.") + .addDetails(Any.pack(ErrorInfo.newBuilder().setReason("SERVER_ERROR_NOT_LEADER_NODE").build())) + .addDetails(Any.pack(NotLeaderNodeErrorDetails.newBuilder() + .setCurrentLeader(NotLeaderNodeErrorDetails.NodeInfo.newBuilder().setHost("127.0.0.1").setPort(2113)) + .build())) + .build(); + CompletableFuture dest = new CompletableFuture<>(); + + Assertions.assertTrue(GrpcUtils.handleNotLeaderError(StatusProto.toStatusRuntimeException(status), dest)); + + ExecutionException error = Assertions.assertThrows(ExecutionException.class, dest::get); + Assertions.assertTrue(error.getCause() instanceof NotLeaderException, + "should complete with a NotLeaderException, got: " + error.getCause()); + Assertions.assertEquals(new InetSocketAddress("127.0.0.1", 2113), + ((NotLeaderException) error.getCause()).getLeaderEndpoint()); + } } From 275a87f393ac5b81e91e81e92a674ef68fdb5174 Mon Sep 17 00:00:00 2001 From: w1am <33353798+w1am@users.noreply.github.com> Date: Fri, 6 Nov 2026 20:52:05 +0400 Subject: [PATCH 2/4] fix: complete future when rich status code mismatches gRPC status --- src/main/java/io/kurrent/dbclient/GrpcUtils.java | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/src/main/java/io/kurrent/dbclient/GrpcUtils.java b/src/main/java/io/kurrent/dbclient/GrpcUtils.java index ddfd08fd..3fd6d16a 100644 --- a/src/main/java/io/kurrent/dbclient/GrpcUtils.java +++ b/src/main/java/io/kurrent/dbclient/GrpcUtils.java @@ -32,7 +32,13 @@ static boolean handleNotLeaderError(Throwable t, CompletableFuture dest) { } } - com.google.rpc.Status status = StatusProto.fromThrowable(e); + com.google.rpc.Status status; + try { + status = StatusProto.fromThrowable(e); + } catch (IllegalArgumentException ignored) { + dest.completeExceptionally(t); + return true; + } if (status != null) { for (Any detail : status.getDetailsList()) { From 263fffee401bc34ea09262858bf9678131c94d75 Mon Sep 17 00:00:00 2001 From: w1am <33353798+w1am@users.noreply.github.com> Date: Fri, 6 Nov 2026 20:59:17 +0400 Subject: [PATCH 3/4] fix: complete future when leader endpoint in not-leader error is invalid --- .../java/io/kurrent/dbclient/GrpcUtils.java | 36 +++++++++---------- 1 file changed, 17 insertions(+), 19 deletions(-) diff --git a/src/main/java/io/kurrent/dbclient/GrpcUtils.java b/src/main/java/io/kurrent/dbclient/GrpcUtils.java index 3fd6d16a..a5609dc7 100644 --- a/src/main/java/io/kurrent/dbclient/GrpcUtils.java +++ b/src/main/java/io/kurrent/dbclient/GrpcUtils.java @@ -17,8 +17,12 @@ final class GrpcUtils { static boolean handleNotLeaderError(Throwable t, CompletableFuture dest) { - if (t instanceof StatusRuntimeException) { - StatusRuntimeException e = (StatusRuntimeException) t; + if (!(t instanceof StatusRuntimeException)) + return false; + + StatusRuntimeException e = (StatusRuntimeException) t; + + try { Metadata trailers = e.getTrailers(); if (trailers != null) { @@ -26,35 +30,29 @@ static boolean handleNotLeaderError(Throwable t, CompletableFuture dest) { String leaderPort = trailers.get(Metadata.Key.of("leader-endpoint-port", Metadata.ASCII_STRING_MARSHALLER)); if (leaderHost != null && leaderPort != null) { - NotLeaderException reason = new NotLeaderException(leaderHost, Integer.parseInt(leaderPort)); - dest.completeExceptionally(reason); + dest.completeExceptionally(new NotLeaderException(leaderHost, Integer.parseInt(leaderPort))); return true; } } - com.google.rpc.Status status; - try { - status = StatusProto.fromThrowable(e); - } catch (IllegalArgumentException ignored) { - dest.completeExceptionally(t); - return true; - } + com.google.rpc.Status status = StatusProto.fromThrowable(e); if (status != null) { for (Any detail : status.getDetailsList()) { if (!detail.is(NotLeaderNodeErrorDetails.class)) continue; - try { - NotLeaderNodeErrorDetails.NodeInfo leader = - detail.unpack(NotLeaderNodeErrorDetails.class).getCurrentLeader(); - dest.completeExceptionally(new NotLeaderException(leader.getHost(), leader.getPort())); - return true; - } catch (InvalidProtocolBufferException ignored) { - return false; - } + NotLeaderNodeErrorDetails.NodeInfo leader = + detail.unpack(NotLeaderNodeErrorDetails.class).getCurrentLeader(); + dest.completeExceptionally(new NotLeaderException(leader.getHost(), leader.getPort())); + return true; } } + } catch (IllegalArgumentException ignored) { + dest.completeExceptionally(t); + return true; + } catch (InvalidProtocolBufferException ignored) { + return false; } return false; From d9e6f2b1a04e47423e27c8741cb8d44b662b350e Mon Sep 17 00:00:00 2001 From: w1am <33353798+w1am@users.noreply.github.com> Date: Fri, 6 Nov 2026 21:08:25 +0400 Subject: [PATCH 4/4] test: cover invalid leader endpoint and status code mismatch in not-leader handling --- .../dbclient/LeaderRedirectUnitTest.java | 40 +++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/src/test/java/io/kurrent/dbclient/LeaderRedirectUnitTest.java b/src/test/java/io/kurrent/dbclient/LeaderRedirectUnitTest.java index f00a1379..548296c3 100644 --- a/src/test/java/io/kurrent/dbclient/LeaderRedirectUnitTest.java +++ b/src/test/java/io/kurrent/dbclient/LeaderRedirectUnitTest.java @@ -5,6 +5,7 @@ import io.grpc.Metadata; import io.grpc.Status; import io.grpc.StatusRuntimeException; +import io.grpc.protobuf.ProtoUtils; import io.grpc.protobuf.StatusProto; import io.kurrentdb.protocol.v2.common.errors.NotLeaderNodeErrorDetails; import org.junit.jupiter.api.Assertions; @@ -112,4 +113,43 @@ public void handleNotLeaderErrorReadsNotLeaderNodeErrorDetails() { Assertions.assertEquals(new InetSocketAddress("127.0.0.1", 2113), ((NotLeaderException) error.getCause()).getLeaderEndpoint()); } + + @Test + public void handleNotLeaderErrorFailsWithOriginalErrorOnInvalidTrailerPort() { + Metadata trailers = new Metadata(); + trailers.put(Metadata.Key.of("leader-endpoint-host", Metadata.ASCII_STRING_MARSHALLER), "127.0.0.1"); + trailers.put(Metadata.Key.of("leader-endpoint-port", Metadata.ASCII_STRING_MARSHALLER), "abc"); + + assertFailsWithOriginalError(new StatusRuntimeException(Status.NOT_FOUND, trailers)); + } + + @Test + public void handleNotLeaderErrorFailsWithOriginalErrorOnOutOfRangeDetailsPort() { + com.google.rpc.Status status = com.google.rpc.Status.newBuilder() + .setCode(Status.Code.FAILED_PRECONDITION.value()) + .addDetails(Any.pack(NotLeaderNodeErrorDetails.newBuilder() + .setCurrentLeader(NotLeaderNodeErrorDetails.NodeInfo.newBuilder().setHost("127.0.0.1").setPort(99999)) + .build())) + .build(); + + assertFailsWithOriginalError(StatusProto.toStatusRuntimeException(status)); + } + + @Test + public void handleNotLeaderErrorFailsWithOriginalErrorOnStatusCodeMismatch() { + Metadata trailers = new Metadata(); + trailers.put(Metadata.Key.of("grpc-status-details-bin", ProtoUtils.metadataMarshaller(com.google.rpc.Status.getDefaultInstance())), + com.google.rpc.Status.newBuilder().setCode(Status.Code.FAILED_PRECONDITION.value()).build()); + + assertFailsWithOriginalError(new StatusRuntimeException(Status.INTERNAL, trailers)); + } + + private static void assertFailsWithOriginalError(StatusRuntimeException original) { + CompletableFuture dest = new CompletableFuture<>(); + + Assertions.assertTrue(GrpcUtils.handleNotLeaderError(original, dest)); + + ExecutionException error = Assertions.assertThrows(ExecutionException.class, dest::get); + Assertions.assertSame(original, error.getCause()); + } }