daib commented on issue #10474:
URL: https://github.com/apache/gravitino/issues/10474#issuecomment-4435340335
# Addressing TreeLock Limitations for Gravitino HA Deployment
## Background
Gravitino uses `TreeLock` (implemented in
`core/src/main/java/org/apache/gravitino/lock/`) to ensure consistency and
atomicity of metadata operations. The lock follows a hierarchical read-write
locking strategy — acquiring read locks on ancestor nodes and a write lock on
the target node. This design is managed by `LockManager`, which maintains an
in-memory tree of `TreeLockNode` instances backed by Java's `ReadWriteLock`.
The core problem in HA mode is straightforward: `TreeLock` is entirely
in-process. Each Gravitino instance maintains its own isolated lock tree in JVM
memory. Multiple instances behind a load balancer have no visibility into each
other's lock state, enabling concurrent conflicting writes to shared metadata.
---
## Analysis of Proposed Options
### Option A — Distributed Lock (ZooKeeper, etcd, Redis)
Option A preserves existing hierarchical lock semantics by replacing the
in-memory lock tree with a distributed coordination service. While semantically
correct, it introduces a fundamental performance problem: every metadata
operation requires a network round trip to acquire and release a lock.
Gravitino is a metadata catalog — a control plane, not a data plane. But
even at moderate operation frequencies, serializing every metadata read and
write through an external coordination service creates an unnecessary
bottleneck. More critically, this approach introduces a new external dependency
that becomes a potential single point of failure, and adds the operational
complexity of managing distributed lock lease expiry, fencing tokens, and
network partition handling.
I have direct production experience with this failure class. While operating
a large-scale distributed system, I diagnosed a chronic ingestion reliability
issue caused by Zookeeper observer node stale reads — a subtle but severe
consistency failure that took months to identify. The lesson: distributed
coordination services are complex failure surfaces. Adding one to every
metadata operation is an architectural liability, not just a performance
concern.
**Recommendation: reject Option A** for the general case. The coordination
overhead is unjustified given the availability of better alternatives.
### Option B — Remove TreeLock; Rely on Storage-Level Consistency
Option B eliminates `TreeLock` entirely, pushing concurrency control into
the storage layer via optimistic locking, CAS operations, or serializable
transactions. This is architecturally clean — the database is already shared
across all instances, making it the natural coordination point.
However, Option B has two significant problems. First, it requires
refactoring every manager class that calls `TreeLockUtils.doWithTreeLock()` — a
large and risky change surface spanning `CatalogManager`, `MetalakeManager`,
`TableOperationDispatcher`, and many others. Second, and more fundamentally,
pure optimistic locking cannot easily express coarse-grained parent-level write
locks. Consider two concurrent operations: renaming schema `db1` to `db2` while
simultaneously creating table `db1.newtable`. These conflict at the parent
level — version-based CAS on individual rows cannot detect this hierarchical
conflict without additional coordination.
**Recommendation: reject Option B** as the primary approach. The refactoring
risk is high and the hierarchical locking semantics are genuinely difficult to
replicate at the storage layer.
---
## Proposed Solution: Option C — Partitioned TreeLock
The core insight is to reframe the problem. Rather than asking "how do we
distribute the lock?", we ask "how do we distribute ownership so each instance
can run its existing local lock without coordination?"
### Design Overview
**Partition the metadata tree** at the metalake level (or catalog level for
finer granularity). Each Gravitino instance owns one or more partitions and is
the sole authority for TreeLock operations within those partitions. Within a
partition, the existing in-memory `TreeLock` runs completely unchanged — no
refactoring, no semantic changes, no new coordination overhead for the common
case.
**Add a routing layer** that directs each incoming request to the instance
that owns the relevant partition. The routing layer reads partition ownership
from a shared database table, ensuring all routers have a consistent view.
**Use the database for coordination** — no new external dependencies. Two
new tables provide everything needed:
```sql
-- Partition ownership and failover configuration
CREATE TABLE partition_map (
partition_key VARCHAR, -- e.g. 'metalake1'
primary_id VARCHAR, -- current active instance
standby_id VARCHAR, -- designated failover instance
tree_version BIGINT, -- incremented on every ownership change
promoted_at TIMESTAMP, -- when current primary was promoted
updated_at TIMESTAMP
);
-- Instance liveness detection
CREATE TABLE instance_heartbeats (
instance_id VARCHAR,
last_seen TIMESTAMP
);
```
**Cross-partition operations** (rare — schema moves, metalake-level
operations) use database advisory locks. The common case — operations within a
single partition — never touches distributed coordination.
### Key Mechanism: TreeLock Version Tag
When a standby instance is promoted to primary, `tree_version` is
incremented in `partition_map`. Incoming requests carry the current
`tree_version` in their header. On receiving a request with a version higher
than its local version, the instance:
1. Invalidates (clears) its local `TreeLock` subtree for that partition
2. Processes the request normally — the `TreeLock` rebuilds lazily as
requests arrive
3. Updates its local version to the new value
This version tag mechanism elegantly solves the stale state problem without
requiring pre-warming, changelog replay, or any additional synchronization. The
database is always the source of truth — the `TreeLock` is purely an ephemeral
coordination cache that rebuilds on demand.
### Routing Layer
The routing layer is the primary new piece of code. Its logic is
straightforward:
```
Request arrives for path /metalake1/catalog/db/table
→ Check watchdog — terminate self if stale (see below)
→ Extract partition key: metalake1
→ Read local cache of partition_map: primary_id = instance_A, tree_version =
7
→ Am I instance_A?
→ Yes: handle locally with existing TreeLock
→ No: proxy to instance_A with tree_version=7 in request header
→ Instance_A receives request, checks version header
→ Version matches current: handle normally
→ Version higher than current: invalidate subtree, rebuild lazily, handle
```
One competent engineer can implement this routing layer in approximately two
weeks. The existing `TreeLock` and all manager classes remain completely
unchanged.
### Router Watchdog — Staleness Self-Termination
A critical correctness requirement: a router must never forward requests
based on stale partition ownership. A router with a stale database connection
could route to an instance that is no longer primary, causing correctness
violations.
This is addressed through a **watchdog mechanism**:
- A background thread polls `partition_map` continuously, updating the
router's local cache
- Each successful poll resets the watchdog timer
- **Before forwarding any request**, the router checks the watchdog timestamp
- If the watchdog has not been updated within **one second** — the router
**terminates itself**
```
Per-request invariant check:
IF watchdog_last_updated < NOW() - 1 second:
→ Kill self (process exit)
ELSE:
→ Forward using current local cache
```
**Why termination rather than pausing:** A paused router is still alive and
may resume forwarding on stale state. Termination is unambiguous — the router
is definitively gone. Kubernetes or the process supervisor restarts it
immediately. On restart the router reads fresh state from `partition_map`
before accepting any requests.
**The invariant this enforces:** Every routing decision is made with
partition ownership information no more than one second old. No router ever
operates on stale state.
### Failover Protocol — Grace Period
When any router detects that primary instance A has failed (via stale
heartbeat in `instance_heartbeats`), it promotes the standby B via CAS on
`partition_map`:
```sql
UPDATE partition_map
SET primary_id = 'instance-b',
tree_version = tree_version + 1,
promoted_at = NOW()
WHERE partition_key = 'metalake1'
AND primary_id = 'instance-a' -- CAS: only one router wins
```
All routers observe the promotion on their next database poll — **within one
second**, enforced by the watchdog. On detecting a new promotion, each router
enters a **grace period**:
- New requests for that partition are **buffered** locally — not rejected,
not forwarded
- In-flight requests to A are allowed to complete or time out naturally
- No requests are forwarded to B during the grace period
After the grace period expires, routers forward buffered and new requests to
B with the new `tree_version`.
**Grace period duration:**
```
grace_period = max(router_poll_interval, max_request_timeout) + safety_buffer
≈ 30–60 seconds (configurable per deployment)
```
**Why this requires no router-to-router coordination:**
The watchdog guarantees every active router reads the promotion within 1
second. The grace period is set longer than the maximum in-flight request
lifetime. Therefore after the grace period:
- Every active router has seen the promotion — guaranteed by watchdog
- Every in-flight request to A has completed or timed out — guaranteed by
grace period
- No router is forwarding to A — follows from both guarantees above
No confirmation tables, no quorum calculation, no router heartbeats, no
router-to-router communication required.
**Client experience:** Clients observe a brief latency increase (buffering
during grace period) rather than errors. Significantly better than approaches
where in-flight requests fail and require client-side retry.
### Failure Scenarios
#### Clean Crash
```
T=0 Instance A crashes
T=1 Router R1 detects stale heartbeat for A
T=1 R1 promotes B via CAS: tree_version → 8, promoted_at = T=1
T=2 All routers read promotion on next poll (watchdog enforces <1s)
T=2 All routers buffer new requests, allow in-flight to A to expire
T=32 Grace period expires (30 seconds)
T=32 All routers forward buffered + new requests to B with tree_version=8
T=32 B receives first request, detects version 8 > local version 7
T=32 B invalidates subtree, rebuilds lazily, serves request
Result: brief buffering latency, no failed requests
```
#### Intermittent Failure
The routing layer tracks a rolling health score per instance based on recent
response success rates. When A's health score drops below a configurable
threshold, any router initiates promotion to B — triggering the same grace
period protocol even if A's heartbeat is still arriving. A cannot interfere
with B's operation because `tree_version` acts as a fencing token: requests
carrying a stale version are rejected at the instance level.
For persistent hardware degradation (bad machine, memory pressure), the
monitoring system pages the on-call engineer. The affected machine is cordoned
and the partition is permanently reassigned. This is the appropriate resolution
— not complex automatic recovery that might mask underlying hardware issues.
#### Router Failure During Failover
If a router loses database connectivity during the grace period:
- Watchdog detects the failure within one second
- Router terminates itself
- Load balancer stops sending traffic to the dead router
- On restart: router reads fresh `partition_map`, observes the promotion,
respects remaining grace period or forwards immediately if it has already passed
The system self-heals without any coordination overhead.
#### Stale Write Prevention — Database-Level Fencing
A slow instance may stall between receiving a request and executing the
database write. Consider:
```
T=0 Request routed to A with tree_version=7
T=1 A acquires TreeLock, begins processing
T=2 A stalls — GC pause, slow disk, memory pressure
T=5 A's heartbeat stops, router promotes B, tree_version → 8
T=40 B processes retry, commits successfully
T=45 A recovers from stall, proceeds to execute write
A's in-process tree_version check passed at T=1 — before the stall
A writes to database — CORRUPTION
```
**In-process checks are insufficient.** The vulnerability exists precisely
in `TreeLockUtils.doWithTreeLock()` — between `lock.lock()` acquiring the
in-memory TreeLock and `executable.execute()` completing the database write:
```java
public static <R, E extends Exception> R doWithTreeLock(
NameIdentifier identifier, LockType lockType, Executable<R, E>
executable) throws E {
TreeLock lock =
GravitinoEnv.getInstance().lockManager().createTreeLock(identifier);
try {
lock.lock(lockType);
return executable.execute(); // ← stall can occur here, after lock
acquired
// before or during database write
} finally {
lock.unlock();
}
}
```
A JVM stall of arbitrary duration can occur inside `executable.execute()`.
Any in-process version check before this call has already passed — it cannot
protect the write. This is a classic TOCTOU (Time-Of-Check-To-Time-Of-Use)
race. A post-write check is also insufficient — the committed write is already
visible to other instances before the check fires, potentially causing
cascading inconsistency.
**The fencing token must be enforced before the write commits — atomically,
inside the same transaction.**
**Solution: ThreadLocal + SessionUtils injection + trigger**
All metadata writes flow through `SessionUtils` — specifically
`doWithCommit()`, `doWithCommitAndFetchResult()`, and `doMultipleWithCommit()`.
These are the transaction boundaries. The fix injects the `tree_version` as a
database session variable on the exact connection used for the write,
guaranteed by injecting immediately after the `SqlSession` connection is
obtained.
**Step 1 — `doWithTreeLock()` sets a ThreadLocal:**
```java
public static <R, E extends Exception> R doWithTreeLock(
NameIdentifier identifier, LockType lockType, Executable<R, E>
executable) throws E {
TreeLock lock =
GravitinoEnv.getInstance().lockManager().createTreeLock(identifier);
try {
lock.lock(lockType);
// Set ThreadLocal — SessionUtils reads it on the same thread
TreeVersionContext.set(getTreeVersion(identifier));
return executable.execute();
} finally {
lock.unlock();
TreeVersionContext.clear();
}
}
```
**Step 2 — `SessionUtils` injects session variable after obtaining
connection:**
```java
public static <T> void doWithCommit(Class<T> mapperClazz, Consumer<T>
consumer) {
try {
T mapper = SqlSessions.getMapper(mapperClazz);
SqlSessions.setTreeVersionIfPresent(); // ← inject on same connection
consumer.accept(mapper);
SqlSessions.commitAndCloseSqlSession();
} catch (Throwable t) {
SqlSessions.rollbackAndCloseSqlSession();
throw t;
}
}
public static <T, R> R doWithCommitAndFetchResult(
Class<T> mapperClazz, Function<T, R> func) {
try {
T mapper = SqlSessions.getMapper(mapperClazz);
SqlSessions.setTreeVersionIfPresent(); // ← inject on same connection
R result = func.apply(mapper);
SqlSessions.commitAndCloseSqlSession();
return result;
} catch (Throwable t) {
SqlSessions.rollbackAndCloseSqlSession();
throw t;
}
}
public static void doMultipleWithCommit(Runnable... operations) {
SqlSessions.getSqlSession();
try {
SqlSessions.setTreeVersionIfPresent(); // ← inject on same connection
Arrays.stream(operations).forEach(Runnable::run);
SqlSessions.commitAndCloseSqlSession();
} catch (Throwable t) {
SqlSessions.rollbackAndCloseSqlSession();
throw t;
}
}
```
**`SqlSessions.setTreeVersionIfPresent()` reads the ThreadLocal:**
```java
public static void setTreeVersionIfPresent() {
Long version = TreeVersionContext.get();
if (version != null) {
// PostgreSQL: SET LOCAL gravitino.tree_version = <version>
// MySQL: SET @gravitino_tree_version = <version>
setSessionVariable("gravitino.tree_version", version);
}
}
```
**Step 3 — Database trigger enforces fencing before commit:**
```sql
CREATE OR REPLACE FUNCTION check_tree_version_fencing()
RETURNS TRIGGER AS $$
DECLARE
request_version BIGINT;
current_version BIGINT;
p_key VARCHAR;
BEGIN
request_version := current_setting('gravitino.tree_version', true)::BIGINT;
-- Only enforce if tree_version was injected (HA mode)
IF request_version IS NULL THEN
RETURN NEW;
END IF;
p_key := get_partition_key(NEW.entity_path);
SELECT tree_version INTO current_version
FROM partition_map
WHERE partition_key = p_key;
IF request_version != current_version THEN
RAISE EXCEPTION
'Stale partition ownership: request version % != current version %',
request_version, current_version;
END IF;
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER enforce_tree_version_fencing
BEFORE INSERT OR UPDATE ON metadata_table
FOR EACH ROW EXECUTE FUNCTION check_tree_version_fencing();
```
**Why this is connection-safe:**
```
doWithTreeLock()
→ TreeVersionContext.set(7) — ThreadLocal on calling thread
→ executable.execute()
→ doWithCommit() / doMultipleWithCommit()
→ SqlSessions.getMapper() — connection C1 obtained
→ setTreeVersionIfPresent() — session variable set on C1
→ mapper executes SQL on C1 — same connection guaranteed
→ trigger fires BEFORE commit — reads session variable from C1
→ version 7 != 8 → EXCEPTION — transaction rolls back
→ TreeVersionContext.clear()
```
The ThreadLocal carries the version across the call stack on the same
thread. `SessionUtils` sets the session variable immediately after obtaining
the connection — on that exact connection. The trigger fires inside the
transaction — write is never committed.
**Read-only operations** — `getWithoutCommit()` and `doWithoutCommit()` —
are unchanged. No session variable injection needed — these don't write, the
trigger does not fire.
**Deployment prerequisite:**
This approach requires either no connection pooler or session-mode pooling
(e.g. pgbouncer in session mode). Transaction-mode or statement-mode poolers
may reset session variables between statements. For such deployments,
`setTreeVersionIfPresent()` should be called within an explicit transaction
boundary to guarantee the session variable persists through the write.
**What this achieves:**
- Write rejected before commit — no inconsistency window possible
- All manager classes — completely unchanged
- All SQL statements — completely unchanged
- `Executable` interface and all call sites — completely unchanged
- `doWithTreeLock()` — two lines added: set/clear ThreadLocal
- `SessionUtils` — one line added to each of the three write methods
- `SqlSessions` — one new method `setTreeVersionIfPresent()`
- New class `TreeVersionContext` — ~10 lines, ThreadLocal wrapper
#### Retry Safety
When a client times out and retries, both the original stalled instance and
the retry target may attempt the same write. All write operations carry a
client-generated idempotent request ID as a second line of defense:
```sql
INSERT INTO metadata_operations (request_id, operation, entity_path, ...)
ON CONFLICT (request_id) DO NOTHING
```
The database deduplicates on `request_id`. However, the database-level
fencing token is the primary and unconditional guard — it rejects stale writes
regardless of retry behavior, request IDs, or client implementation.
### Deployment Tiers
**Tier 1 — Lazy Takeover (minimal complexity):**
Any healthy instance claims an orphaned partition via CAS on
`partition_map`. No designated standby required. Suitable for development
environments and deployments where grace period buffering latency is acceptable.
**Tier 2 — Designated Standby (faster, more predictable):**
Each partition has a designated standby registered in `partition_map`. On
primary failure, the standby claims the partition immediately without a
claiming race. Suitable for production deployments with stricter availability
requirements.
Both tiers use the same watchdog and grace period protocol — only the
standby assignment logic differs.
### Correctness Bound
A key property of this design: **it is never worse than Option A in any
scenario.**
Let:
- N = total metadata operations
- C = cross-partition operations (0 ≤ C ≤ N)
- `d_distributed` = cost of distributed lock operation
- `d_local` = cost of local TreeLock operation
Option A cost: N × d_distributed
Option C cost: C × d_distributed + (N - C) × d_local
Since d_local << d_distributed and C ≤ N:
**Option C cost ≤ Option A cost always.**
In the common case where C approaches 0, Option C approaches pure local
TreeLock performance — identical to current single-instance behavior. In the
degenerate case where every operation is cross-partition, Option C degrades to
Option A — never worse than the baseline.
### Future Scalability
The partitioning primitive extends naturally to database scalability. When a
single database becomes a bottleneck, each partition maps to a dedicated
database shard:
```
metalake1 → Instance A → Database Shard 1
metalake2 → Instance B → Database Shard 2
```
The routing layer already knows the partition key for every request.
Database shard routing is a natural extension of the same logic. Cross-shard
operations use the same distributed lock mechanism as cross-partition
operations today.
---
## Migration Strategy
The migration is entirely additive. No existing code changes are required.
**Phase 1 — Infrastructure (no behavior change):**
Deploy `partition_map` and `instance_heartbeats`. Configure a single
instance owning all partitions. All requests route to that instance — identical
to current behavior.
**Phase 2 — Multi-instance validation:**
Add a second instance with distinct partitions. Validate routing correctness
and TreeLock version invalidation. Monitor cross-partition operation frequency
to calibrate partition granularity.
**Phase 3 — Standby configuration:**
Enable designated standby instances. Validate grace period failover
behavior. Tune watchdog interval and grace period duration for the deployment's
SLA.
**Phase 4 — Full HA:**
Scale to desired instance count. Each phase is independently deployable and
reversible.
---
## Implementation Notes
I am happy to implement the routing layer as part of this contribution. I
estimate approximately two weeks for a complete, tested implementation
including:
- Partition map management and instance heartbeat renewal
- Request routing with tree_version tag propagation
- Watchdog mechanism with self-termination on database staleness
- Grace period buffering on promotion detection
- Health score monitoring for intermittent failure detection
- Cross-partition distributed lock acquisition
- Idempotent request ID handling
- ThreadLocal injection in `doWithTreeLock()`, `SessionUtils`, and database
trigger
**New database tables required (2 total):**
```sql
partition_map -- partition ownership, tree_version, promoted_at
instance_heartbeats -- instance liveness detection
```
**Existing code changes (minimal):**
- `TreeLockUtils.doWithTreeLock()` — set/clear `TreeVersionContext`
ThreadLocal around `executable.execute()`
- `SessionUtils.doWithCommit()` — one line after `getMapper()`
- `SessionUtils.doWithCommitAndFetchResult()` — one line after `getMapper()`
- `SessionUtils.doMultipleWithCommit()` — one line after `getSqlSession()`
- `SqlSessions` — one new method `setTreeVersionIfPresent()`
- `TreeVersionContext` — new ThreadLocal wrapper class, ~10 lines
- Database trigger on metadata write tables
All manager classes, all SQL statements, `TreeLock`, `LockManager`,
`TreeLockNode`, and all call sites remain completely unchanged.
---
## Summary
| Aspect | Option A | Option B | Option C |
|--------|----------|----------|----------|
| External dependency | Yes (ZK/etcd) | No | No |
| Code changes | Minimal | Significant refactor | `doWithTreeLock()` +
`SessionUtils` + trigger |
| Performance | Every op has network cost | Local + retry overhead | Local
for common case |
| Correctness | Strong | Complex to implement | Equivalent to Option A |
| Router safety | N/A | N/A | Watchdog self-termination |
| Failover mechanism | Complex lease management | N/A | Grace period +
version tag |
| Router coordination | N/A | N/A | None required |
| Migration risk | Low | High | Very low |
| New DB tables | External service | 0 | 2 |
| Future scalability | Limited | Good | Extends to DB sharding |
Option C preserves existing `TreeLock` semantics entirely, adds zero new
external dependencies, and is provably never worse than Option A. The watchdog
mechanism ensures no router ever operates on stale partition state —
self-termination on database staleness eliminates the need for router-to-router
coordination entirely. The grace period approach provides clean failover with
no failed requests and no complex synchronization protocols. Database-level
fencing tokens enforce ownership atomically at the write layer, closing the
TOCTOU gap that in-process checks cannot. The design provides a clear migration
path from single-instance to fully partitioned HA deployment, and extends
naturally to database-level sharding as Gravitino scales.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
