From 9c5e6efeaeb07606f654c63ae292ae2b84d54691 Mon Sep 17 00:00:00 2001 From: Himanshu Verma Date: Wed, 21 Oct 2026 10:22:41 +0530 Subject: [PATCH] feat(pd): commit schema keys and a per-graph record as one raft entry A schema DDL reaches PD as several separate puts, each its own raft entry, so a crash between them leaves half a change and no signal that anything changed. Add a txn rpc that applies compares (ABSENT, EQUALS), PUT, DELETE and DELETE_PREFIX ops and an optional per-graph record change as one raft entry and one RocksDB write batch. A compare or an op with an empty key is rejected before the TXN is proposed: to a DELETE_PREFIX an empty key is the whole KV space, and so is a compare, op or record whose enum field is left at its zero value, UNSPECIFIED. Both answer with the new error type TXN_NOT_PROPOSED, so a caller can tell a request PD never proposed from an unknown outcome. The node that proposes a TXN encodes its keys as it encodes the keys of put and ships them in the entry with its charset name, so every node writes the same bytes whatever its own default charset. A TXN that replaces or deletes a value put with a TTL deletes the TTL marker next to it in the same batch: the TTL cleanup is meant to delete the key a marker names once its time has passed, whatever the key holds by then (it deletes another name today, apache/hugegraph#3378). The marker's ASCII prefix is encoded in the proposer's charset and the stored key's bytes follow it as they are, so a key an older node wrote in another charset keeps its marker. The record, HUGEGRAPH/{cluster}/SCHEMA_SYNC/{graphspace}/{graph}, holds {"rev","inc","state"}: rev is the raft index of the TXN that wrote it, or the previous rev plus one when that is higher (a store that changed raft.enable starts its index low again), so PD allocates it at commit and it only rises per graph; inc is the graph's incarnation; state is LIVE or DROPPED. BUMP advances rev and creates the record (LIVE, incarnation 1) the first time a graph from before records is written. CREATE starts the next incarnation, LIVE, and marks the record "src":"create", which BUMP and DROP keep: the origin of a record, not a state. DROP marks the expected incarnation DROPPED. A BUMP or DROP of a DROPPED graph or of another incarnation is rejected, so an instance of an earlier incarnation cannot write into a recreated graph. A stored record that is not a JSON object with positive integers rev and inc and a known state fails the TXN with INVALID_RECORD instead of counting as absent, which would restart the incarnation at 1. The apply reads only the store and the request. jraft applies the log after the last snapshot again on restart, over the store that already holds it, so each TXN writes its index and every store write (TXN or plain put/delete) at or below the last TXN index is skipped on replay: a replayed plain put from before the upgrade would otherwise overwrite what a skipped TXN wrote. Plain writes after the last TXN replay in order to the same store. A TXN is not deduplicated, so it is sent once: KvClient.txn makes one attempt, plus one more on a new stub only when the connection was refused, so the TXN never left the client, and a leader whose apply failed answers with the error instead of forwarding the TXN, since the entry may be committed already and the next leader would apply it again. A failure is an unknown outcome for the caller. A store without raft counts its TXNs under a key of its own, so a store switched to raft later does not take that count for a raft index. The raft path stores the term with the index, and a replay is skipped only while the log has that term at that index and the stored index is within the log; otherwise the apply fails as below. Only a replay is skipped, an entry that was in the node's log when it started; an entry above that index that the store claims to hold means the store is ahead of its log (restored from a backup without the log), and the apply fails on that node instead of skipping it, since the peers apply the entry: jraft stops the state machine, a leader steps down, the probes report the error, and the node rejoins from a snapshot after its data directory is removed. A PD from before the TXN entry logs an unknown op and goes on, so a rolling upgrade could commit entries an old peer never applies. The leader proposes a TXN only once every other voter and learner of the configuration has answered its member probe as a release that applies the entry (a new field of the probe's response); the leader records the confirmed endpoints, itself included, through raft, so a new leader after a failover trusts them, an unconfirmed peer blocks the TXN with TXN_NOT_PROPOSED naming it, a confirmed peer that is unreachable later does not, and unconfirmed peers are probed again at most every 3 s and at once when the configuration changes. A peer that joins catches up by applying the log, TXN entries included, so both membership paths probe every peer the change adds before changing it and refuse an unconfirmed or older one; that also closes the window between a TXN's check and its proposal. Kept peers are covered by the check before every TXN. A node that lost the leadership before it could list its peers answers with TXN_NOT_PROPOSED. A successful TXN with a record calls the registered TxnApplyListener on the apply path, the hook of the schema sync dispatch. The closure adapter passes the apply result to the proposer, which it dropped before, and a TXN that fails in raft reaches the caller with its cause. The client throws a PDException when a txn call gets no response, where the other KV calls hit a NullPointerException, and one that carries the gRPC status when PD answered with one, so a call an older PD rejected (UNIMPLEMENTED) is told from a lost one. A failed txn drops the client's stub only on UNAVAILABLE and only if it still holds the channel of the call, so it drops no stub another thread installed since. hugegraph-pd/docs/api-reference.md documents the txn RPC, the send-once contract, the replay guard and the peer gate. --- hugegraph-pd/docs/api-reference.md | 128 +++ .../hugegraph/pd/client/AbstractClient.java | 84 ++ .../apache/hugegraph/pd/client/KvClient.java | 29 + .../org/apache/hugegraph/pd/KvService.java | 49 + .../apache/hugegraph/pd/raft/KVOperation.java | 14 + .../apache/hugegraph/pd/raft/RaftEngine.java | 159 +++ .../hugegraph/pd/raft/RaftReflectionUtil.java | 29 + .../hugegraph/pd/raft/RaftRpcProcessor.java | 7 + .../hugegraph/pd/raft/RaftStateMachine.java | 12 +- .../apache/hugegraph/pd/raft/TxnPeerGate.java | 269 +++++ .../apache/hugegraph/pd/store/HgKVStore.java | 12 + .../hugegraph/pd/store/HgKVStoreImpl.java | 45 + .../hugegraph/pd/store/KvTxnApplier.java | 508 +++++++++ .../hugegraph/pd/store/RaftKVStore.java | 220 +++- .../hugegraph/pd/store/TxnApplyListener.java | 45 + .../hg-pd-grpc/src/main/proto/kv.proto | 105 ++ .../hg-pd-grpc/src/main/proto/pdpb.proto | 4 + .../pd/service/KvServiceGrpcImpl.java | 27 + .../hugegraph/pd/service/PDService.java | 10 + .../pd/service/KvServiceGrpcImplTest.java | 154 +++ .../hugegraph/pd/client/KvClientTest.java | 316 +++++ .../hugegraph/pd/core/KvServiceTest.java | 61 + .../hugegraph/pd/core/PDCoreSuiteTest.java | 4 + .../pd/core/store/KvTxnApplierTest.java | 1013 +++++++++++++++++ .../raft/RaftEngineIpAuthIntegrationTest.java | 106 ++ .../pd/raft/RaftEngineReadinessTest.java | 43 + .../pd/raft/RaftReflectionUtilTest.java | 27 + .../hugegraph/pd/raft/TxnPeerGateTest.java | 234 ++++ 28 files changed, 3709 insertions(+), 5 deletions(-) create mode 100644 hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/raft/TxnPeerGate.java create mode 100644 hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/store/KvTxnApplier.java create mode 100644 hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/store/TxnApplyListener.java create mode 100644 hugegraph-pd/hg-pd-service/src/test/java/org/apache/hugegraph/pd/service/KvServiceGrpcImplTest.java create mode 100644 hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/core/store/KvTxnApplierTest.java create mode 100644 hugegraph-pd/hg-pd-test/src/main/java/org/apache/hugegraph/pd/raft/TxnPeerGateTest.java diff --git a/hugegraph-pd/docs/api-reference.md b/hugegraph-pd/docs/api-reference.md index 35d70fadd4..8cf592c7e7 100644 --- a/hugegraph-pd/docs/api-reference.md +++ b/hugegraph-pd/docs/api-reference.md @@ -474,6 +474,43 @@ for (Kv.KvPair kv : response.getKvsList()) { } ``` +#### Txn + +Atomic multi-key write: compares, ops and at most one graph record change, applied as one +Raft entry and one RocksDB write batch. See [KV transactions](#kv-transactions) for the +semantics and the retry contract. + +**Request**: +```protobuf +message TxnRequest { + pdpb.RequestHeader header = 1; + repeated TxnCompare compares = 2; // key, kind (ABSENT | EQUALS), value + repeated TxnOp ops = 3; // type (PUT | DELETE | DELETE_PREFIX), key, value + TxnRecord record = 4; // key, op (BUMP | CREATE | DROP), expected_incarnation +} +``` + +**Response**: +```protobuf +message TxnResponse { + pdpb.ResponseHeader header = 1; + bool succeeded = 2; + int64 revision = 3; // the record's rev after the TXN (the Raft index, or the + // previous rev plus one when higher); the index without a record + int64 incarnation = 4; // the record's incarnation after the TXN, or the stored one + Failure failure = 5; // NONE | COMPARE_FAILED | GRAPH_DROPPED | INCARNATION_MISMATCH | + // INVALID_RECORD + int32 failed_compare = 6; + string state = 7; // LIVE | DROPPED, or "" for an absent or unreadable record +} +``` + +A failed compare or a rejected record is an unsuccessful response, not an error. The zero +value of each enum is `UNSPECIFIED`: a compare, op or record whose field was never set is +rejected before the proposal, as is an empty key, with error type `TXN_NOT_PROPOSED` (nothing +was applied). The TXN is proposed by the leader; a follower forwards it. Each node applies the +entry with the keys as the proposing node encoded them, so every replica writes the same bytes. + ### 3. Pulse Service (HgPdPulseGrpc) Heartbeat and health monitoring for Store nodes. @@ -756,6 +793,94 @@ for (Map.Entry entry : results.entrySet()) { } ``` +### KV transactions + +`KvClient.txn(TxnRequest)` commits writes and an optional graph record update as one Raft +entry, applied in one RocksDB batch on every PD node. Only the leader proposes it: a follower +forwards it. With `raft.enable=false` there is no Raft entry, leader or forwarding: the PD +applies the txn in one RocksDB batch under its store's write lock, and every "Raft index" in +this section is the store's own TXN count. `txn` sends the request to one PD, and once more to +the leader found again only when the connection was refused, so the request was never sent; it +throws `PDException` when it gets no answer, or when PD answers with a gRPC status (the status +is in the message and the cause: `UNIMPLEMENTED` from a PD older than this RPC means the call +was rejected before any apply, although the exception's error type is `UNKNOWN`; `UNAVAILABLE` +and `DEADLINE_EXCEEDED` mean it was lost). A `PDException` of error type `TXN_NOT_PROPOSED` is +PD's answer to a request it did not propose (invalid, see below, a PD peer that lacks the RPC, +or a leader that lost the leadership before it could propose), so nothing was applied. Every +other failure is an unknown outcome. +The request has three parts: + +- **Compares.** Each names a key and a kind: `ABSENT` holds while the key does not exist (an + empty value is a value), `EQUALS` while its value equals the given one. If one fails, nothing + is written and the failure is `COMPARE_FAILED`, with the failed compare. +- **Ops.** In order: `PUT` a key and value, `DELETE` a key, `DELETE_PREFIX` every key under a + prefix. An op or compare with an empty key, or whose kind, type or record op was left at the + zero value `UNSPECIFIED`, is rejected before the TXN is proposed. A value that was put with a + TTL has a cleanup marker next to it; an op that replaces or deletes such a value deletes the + marker in the same batch, so the TTL cleanup, which deletes the key a marker names once its + time has passed, cannot delete what the TXN wrote. +- **Record.** At most one graph record, `HUGEGRAPH/{cluster}/SCHEMA_SYNC/{graphspace}/{graph}`, + with an op and the incarnation the writer expects: + - `BUMP` sets `rev` to the Raft index of the entry, or to the previous `rev` plus one when + that is higher (after `raft.enable` changed on the store, the index starts low again), so + `rev` only rises. With expected incarnation 0 it creates an absent record as `LIVE` + incarnation 1. It fails with `GRAPH_DROPPED` on a dropped graph and `INCARNATION_MISMATCH` + on another incarnation. + - `CREATE` starts the next incarnation, or 1, as `LIVE`, from an absent record or a valid + one in either state (an unreadable record fails it, as below), and marks + the record `"src":"create"`, which `BUMP` and `DROP` keep and nothing removes; a record a + `BUMP` created has no mark. A caller that must not take over a `LIVE` record guards the + `CREATE` with compares on its own keys and on the record as it read it (`ABSENT`, or + `EQUALS` the `DROPPED` value). + - `DROP` marks the expected incarnation `DROPPED`. It fails with `INCARNATION_MISMATCH` when + the record is absent or of another incarnation, and `GRAPH_DROPPED` when it is dropped + already. + - A stored record that is not `{"rev":,"inc":,"state":"LIVE"|"DROPPED"}` with positive + integers fails every txn on it with `INVALID_RECORD`. + +The response carries `succeeded`, the `revision` (the record's `rev` after the TXN, or the +Raft index of the entry for a TXN without a record), the record's `incarnation` and `state`, +and the `failure`. Keys and values share the namespace of `put`, +`get` and `scanPrefix`, and a TXN write fires no plain watch event. PD keeps no request ids +and a txn applied twice is not always harmless (a guarded `CREATE` fails its record compare +the second time), so `KvClient` sends a txn once (and once more only when the connection was +refused before anything was sent), and a PD that lost the leadership after it took a txn +answers with the error instead of forwarding it. When `txn` throws, the outcome is unknown, +except for `TXN_NOT_PROPOSED` and `UNIMPLEMENTED` above: read the state before retrying. This is the one call the retry advice under +[Best Practices](#4-retry-strategy) does not cover. + +A TXN is proposed only once every other node of the Raft group (voters and learners) has +answered the leader's member probe as a release that applies the TXN entry; the leader +records the confirmed nodes, itself included, through Raft, so a new leader after a failover +trusts them, and a peer that has not answered yet, or answered as an older release, blocks +every `txn` with `TXN_NOT_PROPOSED` naming it, while a confirmed peer that is unreachable +later does not block, on this leader or the next. Unconfirmed peers are probed again on a `txn` +attempt at most every 3 seconds, and at once when the group's configuration changes, so a node +added to the group blocks `txn` until it answers. A peer that joins applies the log, `txn` entries included, +so a membership change (`changePeerList`, `updatePdRaft`) probes every peer it adds first and +is refused with `TXN_NOT_PROPOSED` when one is unconfirmed or older: an old-release PD can no +longer be added to an upgraded group. Peers the change keeps are not probed, since the check +before every `txn` covers them, so a kept member that is down or not upgraded yet does not stop +a membership change. Upgrade every PD node before +the first `txn`; a PD downgrade after it is unsupported. + +On a restart without a snapshot, PD applies the Raft log after the last snapshot again over +the store that holds it. Every TXN records its Raft index in the store, and a restarted node +skips every store write at or below that index, plain puts included: a plain put from before +the first TXN, applied again over what a later TXN wrote, would undo the TXN. Only a replay is +skipped: an entry that was in the node's log when it started. An entry above that index is new +in this run, and a store that claims to hold it is ahead of its log (a store restored from a +backup without the log): the other nodes apply the entry, and this node fails the apply instead +of skipping it, so that it does not fall behind them in silence. jraft then stops this node's +state machine, a leader steps down, the readiness probe reports the error, and the node logs +the entry and the stored index at error level. The same happens when the stored index is beyond +the end of the log at start, or when the log has another term at the stored index than the one +stored with it (the log was rebuilt or replaced while the store was kept). If the node cannot +read its own raft log, it replays without that check and logs once, at error level, that the log +generation could not be verified, since fencing every node on restart would leave none to rejoin +from. Recovery: remove the node's data directory and let it rejoin from a snapshot. A PD run without Raft (`raft.enable=false`) counts its TXNs +under a key of its own, so switching such a store to Raft does not trip this. + ### KV watch recovery `KvClient.listen()` and `listenPrefix()` restore subscriptions after retryable @@ -1057,6 +1182,9 @@ ManagedChannel channel = ManagedChannelBuilder - Implement exponential backoff for retries - Retry on transient errors (UNAVAILABLE, DEADLINE_EXCEEDED) - Don't retry on permanent errors (NOT_FOUND, INVALID_ARGUMENT) +- Except for `txn`: a lost `txn` may have committed, so it is not retried once it may have + been sent (it goes once more only when the connection was refused before sending), and the + caller reads the state before retrying (see [KV transactions](#kv-transactions)) ```java int maxRetries = 3; diff --git a/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/AbstractClient.java b/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/AbstractClient.java index 44ca5dccee..58d3a6d4b0 100644 --- a/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/AbstractClient.java +++ b/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/AbstractClient.java @@ -18,6 +18,7 @@ package org.apache.hugegraph.pd.client; import java.io.Closeable; +import java.net.ConnectException; import java.util.LinkedList; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -42,6 +43,7 @@ import io.grpc.ManagedChannel; import io.grpc.ManagedChannelBuilder; import io.grpc.MethodDescriptor; +import io.grpc.Status; import io.grpc.StatusRuntimeException; import io.grpc.stub.AbstractBlockingStub; import io.grpc.stub.AbstractStub; @@ -279,6 +281,88 @@ protected RespT blockingUnaryCall( return null; } + /** + * Sends the request once, for a call that is not idempotent: it can't be sent again when + * its response is lost. An UNAVAILABLE status drops the stub if it still holds the channel + * of the call, so the next call connects again; other failures keep it, since calls of + * other threads may be running on the channel and a reconnect would cut them off. A call + * whose connection was refused never left this client, so it is sent once more on a new + * stub, as a PD that stopped a moment ago leaves it. + * + * @return null when the call failed + */ + protected RespT blockingUnaryCallOnce( + MethodDescriptor method, ReqT req) throws PDException { + return blockingUnaryCallOnce(method, req, false); + } + + private RespT blockingUnaryCallOnce( + MethodDescriptor method, ReqT req, boolean resent) throws PDException { + AbstractBlockingStub stub; + Channel attemptChannel; + synchronized (this) { + stub = getBlockingStub(); + // The stub of the call wraps the channel with its interceptors: compare the stored one + attemptChannel = proxy.getBlockingStub().getChannel(); + } + try { + return ClientCalls.blockingUnaryCall(stub.getChannel(), method, + stub.getCallOptions(), req); + } catch (StatusRuntimeException e) { + boolean unsent = !resent && refused(e); + log.warn("{} failed, {}: {}", method.getFullMethodName(), + unsent ? "never sent, sending it once more" : "not sent again", + e.getStatus()); + if (e.getStatus().getCode() == Status.Code.UNAVAILABLE) { + synchronized (this) { + AbstractBlockingStub current = proxy.getBlockingStub(); + if (current != null && current.getChannel() == attemptChannel) { + proxy.setBlockingStub(null); + } + } + } + if (unsent) { + return blockingUnaryCallOnce(method, req, true); + } + throw failed(method, e); + } catch (RuntimeException e) { + log.error(method.getFullMethodName() + " exception, not sent again", e); + return null; + } + } + + /** + * The failure of a call made once, with the gRPC status, so the caller can tell a call PD + * rejected (UNIMPLEMENTED from an older PD, PERMISSION_DENIED) from one that was lost + * (UNAVAILABLE, DEADLINE_EXCEEDED) and has an unknown outcome + */ + private PDException failed(MethodDescriptor method, StatusRuntimeException e) { + Status status = e.getStatus(); + boolean lost = status.getCode() == Status.Code.UNAVAILABLE || + status.getCode() == Status.Code.DEADLINE_EXCEEDED; + String message = method.getFullMethodName() + " failed with " + status.getCode() + + (status.getDescription() == null ? "" : + ": " + status.getDescription()) + + (lost ? ", pd.peers=" + config.getServerHost() : ""); + return new PDException(lost ? Pdpb.ErrorType.PD_UNREACHABLE_VALUE : + Pdpb.ErrorType.UNKNOWN_VALUE, message, e); + } + + /** + * Whether the call failed to connect, so its request was never written + */ + private static boolean refused(StatusRuntimeException e) { + if (e.getStatus().getCode() != Status.Code.UNAVAILABLE) { + return false; + } + for (Throwable cause = e.getCause(); cause != null; cause = cause.getCause()) { + if (cause instanceof ConnectException) { + return true; + } + } + return false; + } + // this.stubs = new ConcurrentHashMap(hosts.length); private AbstractBlockingStub getConcurrentBlockingStub(String address) { AbstractBlockingStub stub = stubs.get(address); diff --git a/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java b/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java index e513fa62e7..7480091352 100644 --- a/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java +++ b/hugegraph-pd/hg-pd-client/src/main/java/org/apache/hugegraph/pd/client/KvClient.java @@ -49,6 +49,8 @@ import org.apache.hugegraph.pd.grpc.kv.ScanPrefixResponse; import org.apache.hugegraph.pd.grpc.kv.TTLRequest; import org.apache.hugegraph.pd.grpc.kv.TTLResponse; +import org.apache.hugegraph.pd.grpc.kv.TxnRequest; +import org.apache.hugegraph.pd.grpc.kv.TxnResponse; import org.apache.hugegraph.pd.grpc.kv.WatchEvent; import org.apache.hugegraph.pd.grpc.kv.WatchKv; import org.apache.hugegraph.pd.grpc.kv.WatchRequest; @@ -191,6 +193,33 @@ public TTLResponse putTTL(String key, String value, long ttl) throws PDException return response; } + /** + * Atomic multi-key write. A failed compare or a rejected record is not an error: check + * succeeded on the response. The call is sent once: a TXN applied twice is not always + * harmless, a guarded CREATE fails its compare the second time. Only a call whose + * connection was refused, so never sent, goes once more to the leader found again. When + * no response arrives, or the PD that took the TXN lost the leadership before answering, + * it throws. + * + * @throws PDException with an unknown outcome: PD may have applied the TXN, except with + * error type TXN_NOT_PROPOSED, which PD answers before it proposes a + * request it rejects (invalid, or a PD peer without the txn opcode), so + * nothing was applied. A call PD answered with a gRPC status carries + * that status in its message and as its cause (UNIMPLEMENTED: a PD + * from before this RPC rejected it before any apply; UNAVAILABLE, + * DEADLINE_EXCEEDED: the call was lost) + */ + public TxnResponse txn(TxnRequest request) throws PDException { + TxnResponse response = blockingUnaryCallOnce(KvServiceGrpc.getTxnMethod(), request); + if (response == null) { + throw new PDException(Pdpb.ErrorType.PD_UNREACHABLE_VALUE, + "No response to a txn from PD, pd.peers=" + + config.getServerHost()); + } + handleErrors(response.getHeader()); + return response; + } + private void onEvent(WatchResponse value, Consumer consumer) { log.debug("receive message for {},event Count:{}", value, value.getEventsCount()); if (value.getEventsCount() != 0) { diff --git a/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/KvService.java b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/KvService.java index abb71cc8ec..4bc4dd2262 100644 --- a/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/KvService.java +++ b/hugegraph-pd/hg-pd-core/src/main/java/org/apache/hugegraph/pd/KvService.java @@ -18,6 +18,8 @@ package org.apache.hugegraph.pd; import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; import java.util.HashMap; import java.util.LinkedList; import java.util.List; @@ -26,10 +28,13 @@ import org.apache.hugegraph.pd.common.PDException; import org.apache.hugegraph.pd.config.PDConfig; import org.apache.hugegraph.pd.grpc.kv.Kv; +import org.apache.hugegraph.pd.grpc.kv.TxnRequest; +import org.apache.hugegraph.pd.grpc.kv.TxnResponse; import org.apache.hugegraph.pd.grpc.kv.V; import org.apache.hugegraph.pd.meta.MetadataKeyHelper; import org.apache.hugegraph.pd.meta.MetadataRocksDBStore; import org.apache.hugegraph.pd.store.KV; +import org.apache.hugegraph.pd.store.KvTxnApplier; import org.springframework.stereotype.Service; import com.google.protobuf.InvalidProtocolBufferException; @@ -71,6 +76,40 @@ public static byte[] getKeyBytes(Object... keys) { return key.getBytes(Charset.defaultCharset()); } + /** + * The TTL marker {@link #getTTLStoreKey(String, long)} wrote next to the store key of a + * value that expires at expireAt (the ttl field of its {@link V}), so every raft node + * names the same marker, the one getTTLStoreKey wrote. The TTL cleanup is meant to delete + * the key a marker names once the marker's time has passed, whatever the key holds by then + * (it deletes another name today, apache/hugegraph#3378), so a write that replaces a TTL + * value deletes its marker too. + */ + public static byte[] ttlMarkerKey(byte[] storeKey, long expireAt, String charsetName) { + /* + * K@T@