From 9cb4e558200c90e96414fcda3cccd7bd1536baf1 Mon Sep 17 00:00:00 2001 From: Himanshu Verma Date: Tue, 20 Oct 2026 22:30:57 +0530 Subject: [PATCH] fix(pd): wait for a ReadIndex before serving get and scanPrefix A leader elected a moment ago may not have applied every committed write yet, so a KV get or scanPrefix answered from its local store could return a value older than one a client already saw committed. Both reads now wait for a raft ReadIndex first, the barrier #3368 added for the id counters, and fail with an error and no value when the leader cannot confirm it within raft.rpc-timeout. The configuration doc says the deadline also bounds these two reads. --- hugegraph-pd/docs/configuration.md | 2 +- .../pd/service/KvServiceGrpcImpl.java | 8 + .../pd/service/KvServiceGrpcImplTest.java | 205 ++++++++++++++++++ 3 files changed, 214 insertions(+), 1 deletion(-) create mode 100644 hugegraph-pd/hg-pd-service/src/test/java/org/apache/hugegraph/pd/service/KvServiceGrpcImplTest.java diff --git a/hugegraph-pd/docs/configuration.md b/hugegraph-pd/docs/configuration.md index c2d6eecc9f..815f1e5d4c 100644 --- a/hugegraph-pd/docs/configuration.md +++ b/hugegraph-pd/docs/configuration.md @@ -119,7 +119,7 @@ raft: |-----------|------|---------|-------------| | `raft.address` | String | `127.0.0.1:8610` | Raft service address for this PD node. Format: `:`. Must be unique across all PD nodes. | | `raft.peers-list` | String | `127.0.0.1:8610` | Comma-separated list of all PD nodes' Raft addresses. Used for cluster formation and leader election. | -| `raft.rpc-timeout` | Integer | `10000` | Timeout of a Raft RPC between PD nodes, in milliseconds. It is also the deadline of the ReadIndex wait PD runs before it reads an id counter: `getId` and `getCId` fail when the leader cannot confirm and apply a ReadIndex in this time, for example while a new leader is still committing its first entry. | +| `raft.rpc-timeout` | Integer | `10000` | Timeout of a Raft RPC between PD nodes, in milliseconds. It is also the deadline of the ReadIndex wait PD runs before it reads an id counter or serves a KV read: `getId`, `getCId`, and the KV `get` and `scanPrefix` fail with an error and no value when the leader cannot confirm and apply a ReadIndex in this time, for example while a new leader is still committing its first entry. | | `raft.rpc-connect-timeout` | Integer | `1000` | Timeout for opening a connection to a peer, in milliseconds. A candidate opens its connections one peer at a time while it holds the Raft node lock, so an election waits this long for each peer that accepts the connection but does not answer (a stopped process, a lost host). It is the timeout of the Ping that opens a connection; requests on a connection that is already open use `raft.rpc-timeout`. | | `raft.rpc-install-snapshot-timeout` | Integer | `0` | Timeout for sending a snapshot to a follower that fell behind the log, in milliseconds. It bounds the transfer of the whole metadata store. `0` uses `raft.rpc-timeout`, the value this timeout had before it became a separate option; raise it if large snapshots time out. | diff --git a/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/service/KvServiceGrpcImpl.java b/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/service/KvServiceGrpcImpl.java index 94b41a0f0a..c53f4276bb 100644 --- a/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/service/KvServiceGrpcImpl.java +++ b/hugegraph-pd/hg-pd-service/src/main/java/org/apache/hugegraph/pd/service/KvServiceGrpcImpl.java @@ -158,6 +158,12 @@ public void get(K request, StreamObserver responseObserver) { KResponse response; KResponse.Builder builder = KResponse.newBuilder(); try { + /* + * A leader elected a moment ago may not have applied every committed write yet, + * and a Server that read such a value would cache schema older than a revision it + * acknowledged: read after a ReadIndex barrier. + */ + RaftEngine.getInstance().waitReadIndex(); String value = this.kvService.get(request.getKey()); builder.setHeader(getResponseHeader()); if (value != null) { @@ -260,6 +266,8 @@ public void scanPrefix(K request, StreamObserver responseObs ScanPrefixResponse response; ScanPrefixResponse.Builder builder = ScanPrefixResponse.newBuilder(); try { + // After a ReadIndex barrier, as in get + RaftEngine.getInstance().waitReadIndex(); Map kvs = this.kvService.scanWithPrefix(request.getKey()); response = builder.setHeader(getResponseHeader()).putAllKvs(kvs).build(); } catch (PDException e) { diff --git a/hugegraph-pd/hg-pd-service/src/test/java/org/apache/hugegraph/pd/service/KvServiceGrpcImplTest.java b/hugegraph-pd/hg-pd-service/src/test/java/org/apache/hugegraph/pd/service/KvServiceGrpcImplTest.java new file mode 100644 index 0000000000..ceb2d62045 --- /dev/null +++ b/hugegraph-pd/hg-pd-service/src/test/java/org/apache/hugegraph/pd/service/KvServiceGrpcImplTest.java @@ -0,0 +1,205 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hugegraph.pd.service; + +import java.io.File; +import java.lang.reflect.Proxy; +import java.nio.file.Files; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.apache.commons.io.FileUtils; +import org.apache.hugegraph.pd.KvService; +import org.apache.hugegraph.pd.config.PDConfig; +import org.apache.hugegraph.pd.grpc.Pdpb; +import org.apache.hugegraph.pd.grpc.kv.K; +import org.apache.hugegraph.pd.grpc.kv.KResponse; +import org.apache.hugegraph.pd.grpc.kv.ScanPrefixResponse; +import org.apache.hugegraph.pd.raft.RaftEngine; +import org.apache.hugegraph.testutil.Whitebox; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import com.alipay.sofa.jraft.Node; +import com.alipay.sofa.jraft.Status; +import com.alipay.sofa.jraft.closure.ReadIndexClosure; +import com.alipay.sofa.jraft.error.RaftError; + +import io.grpc.stub.StreamObserver; + +/** + * get and scanPrefix read the local store of the leader. A leader elected a moment ago may not + * have applied every committed write, so both wait for a ReadIndex first, and return an error + * rather than a stored value when it fails. The raft node answers each ReadIndex the way + * jraft does: it sets the result and runs the closure. + */ +public class KvServiceGrpcImplTest { + + private final List calls = new CopyOnWriteArrayList<>(); + private volatile Status readIndex = Status.OK(); + private Node originalRaftNode; + private PDConfig.Raft originalRaftConfig; + private KvServiceGrpcImpl service; + + @Before + public void setUp() throws Exception { + // The store is opened once per JVM, in the directory of the first test that makes a + // KvService, and stays open: the directory has to outlive the test + File dataPath = Files.createTempDirectory("kv-service-grpc").toFile(); + Runtime.getRuntime().addShutdownHook(new Thread(() -> FileUtils.deleteQuietly(dataPath))); + PDConfig config = new PDConfig(); + config.setDataPath(dataPath.getAbsolutePath()); + PDConfig.Raft raft = config.new Raft(); + raft.setEnable(false); + // Bounds the wait for the ReadIndex: one that times out is reported at this deadline + raft.setRpcTimeout(300); + config.setRaft(raft); + + RaftEngine engine = RaftEngine.getInstance(); + this.originalRaftNode = engine.getRaftNode(); + this.originalRaftConfig = Whitebox.getInternalState(engine, "config"); + Whitebox.setInternalState(engine, "raftNode", leaderNode()); + Whitebox.setInternalState(engine, "config", raft); + + this.service = new KvServiceGrpcImpl() { + @Override + public boolean isLeader() { + return true; + } + }; + // The reads alone: what they return is what the store holds when they run + this.service.kvService = new KvService(config) { + @Override + public String get(String key) { + KvServiceGrpcImplTest.this.calls.add("get"); + return "stored"; + } + + @Override + public Map scanWithPrefix(String key) { + KvServiceGrpcImplTest.this.calls.add("scan"); + return Map.of(key + "a", "stored"); + } + }; + } + + @After + public void tearDown() { + RaftEngine engine = RaftEngine.getInstance(); + Whitebox.setInternalState(engine, "raftNode", this.originalRaftNode); + Whitebox.setInternalState(engine, "config", this.originalRaftConfig); + } + + @Test + public void testGetAndScanReadAfterTheReadIndex() { + KResponse got = get("k"); + Assert.assertEquals(Pdpb.ErrorType.OK, got.getHeader().getError().getType()); + Assert.assertEquals("stored", got.getValue()); + Assert.assertEquals(List.of("readIndex", "get"), this.calls); + + this.calls.clear(); + ScanPrefixResponse scanned = scan("k/"); + Assert.assertEquals(Pdpb.ErrorType.OK, scanned.getHeader().getError().getType()); + Assert.assertEquals(Map.of("k/a", "stored"), scanned.getKvsMap()); + Assert.assertEquals(List.of("readIndex", "scan"), this.calls); + } + + @Test + public void testFailedReadIndexReturnsNoStoredValue() { + // As on a leader that cannot confirm its log with a quorum + this.readIndex = new Status(RaftError.ETIMEDOUT, "no quorum"); + + KResponse got = get("k"); + Assert.assertEquals(Pdpb.ErrorType.UNKNOWN, got.getHeader().getError().getType()); + Assert.assertEquals("", got.getValue()); + + ScanPrefixResponse scanned = scan("k/"); + Assert.assertEquals(Pdpb.ErrorType.UNKNOWN, scanned.getHeader().getError().getType()); + Assert.assertTrue(scanned.getKvsMap().isEmpty()); + // The store was not read at all + Assert.assertEquals(List.of("readIndex", "readIndex"), this.calls); + } + + private KResponse get(String key) { + Capture observer = new Capture<>(); + this.service.get(K.newBuilder().setKey(key).build(), observer); + return observer.result(); + } + + private ScanPrefixResponse scan(String prefix) { + Capture observer = new Capture<>(); + this.service.scanPrefix(K.newBuilder().setKey(prefix).build(), observer); + return observer.result(); + } + + private Node leaderNode() { + return (Node) Proxy.newProxyInstance( + Node.class.getClassLoader(), new Class[]{Node.class}, + (proxy, method, args) -> { + switch (method.getName()) { + case "isLeader": + return true; + case "readIndex": + this.calls.add("readIndex"); + Status status = this.readIndex; + ReadIndexClosure closure = (ReadIndexClosure) args[args.length - 1]; + closure.setResult(status.isOk() ? 42L : + ReadIndexClosure.INVALID_LOG_INDEX, new byte[0]); + closure.run(status); + return null; + case "toString": + return "leader raft node"; + case "hashCode": + return System.identityHashCode(proxy); + case "equals": + return proxy == args[0]; + default: + throw new AssertionError("unexpected call " + method.getName()); + } + }); + } + + private static final class Capture implements StreamObserver { + + private T value; + private boolean completed; + + @Override + public void onNext(T value) { + this.value = value; + } + + @Override + public void onError(Throwable t) { + throw new AssertionError("unexpected stream error", t); + } + + @Override + public void onCompleted() { + this.completed = true; + } + + private T result() { + Assert.assertTrue(this.completed); + return this.value; + } + } +}