…lacement
Motivation:
Multi-tenant HDFS clusters often need to pin a subset of data (identified by
its HDFS path) to a dedicated pool of DataNodes so that a noisy or
isolation-sensitive tenant does not share write/read capacity with the rest
of the cluster. The default BlockPlacementPolicy has no notion of such
path-to-DataNode affinity, so operators resort to separate clusters or
brittle rack hacks.
Approach:
Introduce a pluggable DatanodeAffinityManager abstraction resolved by
reflection from `dfs.datanode.affinity.manager.classname`. An affinity group
maps a source-path regex to a DataNode-hostname regex. On refresh the manager
resolves each DataNode-hostname regex against the live cluster and builds, per
group, a restricted NetworkTopology containing only that group's eligible
DataNodes.
- DatanodeManager instantiates the configured manager, removes affinity
DataNodes from the default NetworkTopology as they register (so the default
policy can never select them), prunes them from the affinity structures when
they are decommissioned/removed, and re-triggers a refresh on
`dfsadmin -refreshNodes`.
- BlockManager keeps one BlockPlacementPolicies per affinity group, each backed
by the group's restricted topology. Both initial block placement
(chooseTarget4NewBlock) and mid-write pipeline recovery
(chooseTarget4AdditionalDatanode) route through the matching group's policy
so a replacement node for a failed pipeline member stays inside the isolated
pool. When no group matches, the default policy is used.
- A built-in FileDatanodeAffinityManager loads affinity groups from a JSON file
(`dfs.datanode.affinity.file.path`), reloaded on `dfsadmin -refreshNodes`.
- `hdfs fsck <path> -favored-nodes` prints the affinity-resolved DataNodes for a
path as a dry-run, without requiring the path to exist.
Isolation vs. availability:
When an affinity group cannot place all requested replicas (under-provisioned
group, all group DataNodes down, or a datanodesRegex that matched no nodes),
the default behaviour falls back to the shared pool so the write still succeeds,
logging the spillover at WARN. Setting
`dfs.namenode.affinity.strict.isolation.enabled=true` makes initial placement
fail closed unless ALL requested replicas fit inside the group -- placing only
minReplication in-group would let the redundancy monitor repair the remainder
on shared-pool nodes and leak data out of the group.
Scope / limitations:
Affinity governs initial block placement and in-pipeline recovery. Background
replication / EC reconstruction still uses the default placement policy: making
it affinity-aware requires a reverse block->path lookup inside a lock-sensitive
hot loop and is intentionally left as follow-up work (documented in
BlockManager). DataNode-regex patterns are matched (via find()) against each
node's "hostname:port" address, so a fully anchored pattern must allow the
trailing ":port" (e.g. `^dn-tenant-a[0-9]+\.example\.com(:\d+)?$`).
Configuration:
- dfs.datanode.affinity.manager.classname (empty/disabled by default)
- dfs.datanode.affinity.file.path (JSON file for FileDatanodeAffinityManager)
- dfs.namenode.affinity.strict.isolation.enabled (default false)
Concurrency and correctness hardening:
- Both the strict and non-strict modes now require the affinity group to place
ALL requested replicas before returning in-group targets. Accepting a partial
placement (only >= minReplication) would create the block under-replicated and
let the redundancy monitor "repair" the remainder onto shared-pool nodes --
leaking data out of the isolated group (strict) or leaving a silent
under-replication window (non-strict). On a shortfall strict fails closed and
non-strict spills the WHOLE block to the shared pool via the default policy, so
it is fully replicated at write time.
- Symmetric registration/removal races during a lock-free config refresh are
reconciled: internalRefresh() takes a live snapshot before publishing, then
(a) postAffinityRefresh() re-evaluates every live DataNode against the current
patterns so a node that registered mid-refresh stays isolated instead of
leaking back into the default topology, and (b) a post-publish purge re-runs
onDatanodeRemoved() for any snapshot node no longer live so a node removed
mid-refresh cannot linger as a dead entry in the freshly published structures.
The isolated-address set is a stable final set mutated in place (never swapped)
so concurrent registration adds are not lost.
- DataNode replacement (same host:port re-registering with a new descriptor /
storage UUID) is reconciled in the per-group topology: because that topology
stores descriptor OBJECTS but membership is address-keyed, a replacement could
otherwise leave a STALE, dead descriptor selectable for placement.
onDatanodeRegistered() now swaps the stale descriptor for the live one
(object-identity guarded), removing every stale descriptor for the same
transfer endpoint regardless of the topology path it sits under (so a
replacement that also changes rack/IP cannot leave a dead leaf behind), and
postAffinityRefresh() routes every live node through it (no address
short-circuit) so the reconciliation always runs.
- The per-group restricted topology is never a storage-type-aware
DFSNetworkTopology. A DFSNetworkTopology captures each node's storage-type
counts at add() time and updates them only through the descriptor's parent
back-pointer; but an affinity node's shared descriptor has its parent nulled
when it is removed from the default topology, so those counts could never
update from later storage reports. Because a DataNode registers (and is added
to its group) with an EMPTY storage map -- storage reports arrive afterwards --
a DFSNetworkTopology group would freeze the count at ZERO and, via
storage-type-gated selection, return no target: group placement would silently
break after every NameNode restart (all affinity nodes re-register fresh)
until a manual `dfsadmin -refreshNodes`. So when the cluster uses
DFSNetworkTopology (the default) the group topology is built as a plain
NetworkTopology, which tracks no storage counts; the placement policy instead
selects a random eligible leaf and validates the storage type against the LIVE
descriptor, which is always current. When dfs.use.dfs.network.topology=false
the configured net.topology.impl is honored instead (e.g.
NetworkTopologyWithNodeGroup for a node-group replicator, whose placement
policy's initialize() rejects any other clusterMap type) -- those impls track
no storage counts, so they are immune to the staleness problem while keeping
the isolation feature from silently failing open under a node-group topology.
- Pipeline recovery (chooseTarget4AdditionalDatanode) now routes through the
affinity group's policy ONLY when the surviving pipeline replicas actually
live inside that group's restricted topology; otherwise it uses the default
policy. In non-strict availability mode an under-provisioned group spills the
WHOLE block to the shared pool, so the survivors are shared-pool nodes absent
from the tiny group topology. Forcing the group policy would then count those
out-of-topology survivors against the group in getMaxNodesPerRack (numChosen +
numAdditional exceeds the group's leaf count), drive the additional count to 0,
return no replacement, and -- with the client default best-effort=false --
fail the write. Recovering a spilled block through the default policy keeps
the write alive and never leaks isolation (the block already lives in the
shared pool, and affinity nodes are absent from the default topology).
Testing:
TestDatanodeAffinityBlockPlacement (14 tests) covers: all replicas of an
affinity path landing only on the group's DataNodes; non-affinity paths using
the full cluster; isolation surviving DataNode re-registration; stale-node
pruning on removal; strict isolation failing both an unsatisfiable and a
partially-provisioned write; non-strict fallback succeeding via the shared
pool for both an empty group and a partially-provisioned group (whole block
spilled, no replica left on the under-provisioned group); pipeline recovery of
a spilled (shared-pool) block succeeding via the default policy while in-group
recovery stays in-group; malformed (null-regex) records being skipped without
aborting refresh; a DataNode that registers during a config refresh being
reconciled into its group instead of leaking back into the default topology; a
replaced DataNode (same address, new UUID) having its group topology reconciled
to the live descriptor, both in place and when the replacement moves to a
different rack/path; and the per-group topology being a plain NetworkTopology
(not a storage-type-aware DFSNetworkTopology) so its node storage accounting
cannot go stale for affinity nodes. The removal-race purge reuses
onDatanodeRemoved(), whose pruning of the published structures is covered by the
removal test; the exact mid-refresh interleaving is not unit-tested because it
cannot be made deterministic without intrusive production test hooks.
Results:
Enables strict tenant isolation of write and read traffic onto a dedicated
DataNode pool while leaving the rest of the cluster on the default placement
policy, refreshable at runtime without a NameNode restart.
Description of PR
Multi-tenant HDFS clusters often need to pin a subset of data, identified by its HDFS path, to a dedicated pool of DataNodes. This prevents noisy or isolation-sensitive tenants from sharing read/write capacity with the rest of the cluster.
The default
BlockPlacementPolicyhas no notion of path-to-DataNode affinity, so operators typically resort to separate clusters or brittle rack-based workarounds.Approach
Introduce a pluggable
DatanodeAffinityManagerabstraction, resolved via reflection from:An affinity group maps:
On refresh, the manager resolves each DataNode hostname regex against the live cluster and builds a restricted
NetworkTopologyfor each group containing only that group's eligible DataNodes.Components
DatanodeManagerNetworkTopologyas they register, ensuring the default placement policy cannot select them.dfsadmin -refreshNodes.BlockManagerBlockPlacementPolicyper affinity group.chooseTarget4NewBlock) through the matching group's policy.chooseTarget4AdditionalDatanode) through the matching group's policy.FileDatanodeAffinityManagerBuilt-in implementation that loads affinity groups from a JSON file.
Configuration:
Reloads the configuration on
dfsadmin -refreshNodes.hdfs fsckSupports:
Prints the affinity-resolved DataNodes for a path as a dry run.
Does not require the path to exist.
Isolation vs. Availability
If an affinity group cannot place all requested replicas because of:
datanodesRegexmatching no nodes,the default behavior falls back to the shared pool so the write can still succeed. Spillover is logged at
WARN.Strict isolation can be enabled with:
dfs.namenode.affinity.strict.isolation.enabled=trueIn strict mode, initial placement fails closed unless all requested replicas can be placed inside the affinity group.
This is important because placing only
minReplicationreplicas in-group would allow the redundancy monitor to repair the remaining replicas onto shared-pool DataNodes, leaking data outside the isolated group.Scope and Limitations
Affinity currently governs:
Background replication and EC reconstruction still use the default placement policy.
Making these paths affinity-aware would require a reverse block-to-path lookup inside a lock-sensitive hot loop and is intentionally left as follow-up work. This is documented in
BlockManager.DataNode Regex Matching
DataNode regex patterns are matched using
find()against each node's:Therefore, a fully anchored pattern must account for the trailing port.
For example:
Configuration
dfs.datanode.affinity.manager.classnamedfs.datanode.affinity.file.pathFileDatanodeAffinityManagerdfs.namenode.affinity.strict.isolation.enabledfalseConcurrency and Correctness Hardening
1. Full-Replica Placement
Both strict and non-strict modes require the affinity group to place all requested replicas before returning in-group targets.
Accepting a partial placement, even when it satisfies
minReplication, creates an under-replicated block. The redundancy monitor could then repair the remaining replicas onto shared-pool DataNodes.Behavior on shortfall:
This guarantees that the block is fully replicated at write time and avoids both isolation leaks and silent under-replication windows.
2. Registration/Removal Races During Refresh
Registration and removal races during lock-free configuration refresh are reconciled.
internalRefresh()first takes a live snapshot before publishing the refreshed state. After publishing:postAffinityRefresh()re-evaluates every live DataNode against the current affinity patterns.A post-publish purge re-runs
onDatanodeRemoved()for any snapshot node that is no longer live.The isolated-address set is a stable set that is mutated in place rather than replaced. This ensures concurrent registration updates are not lost.
3. DataNode Replacement
A DataNode can re-register with the same
host:portbut a new descriptor or storage UUID.Affinity group topologies store descriptor objects while membership is address-keyed. Without reconciliation, a replacement could leave a stale descriptor selectable for placement.
onDatanodeRegistered()now:postAffinityRefresh()routes every live DataNode through this reconciliation logic rather than using an address-based short circuit.4. Per-Group Topology and Storage Types
The per-group restricted topology is intentionally not a storage-type-aware
DFSNetworkTopology.DFSNetworkTopologycaptures each node's storage-type counts when the node is added and updates them through the descriptor's parent back-pointer.Affinity DataNodes are removed from the default topology, which nulls their parent back-pointer. Consequently, storage-type counts in a
DFSNetworkTopologyaffinity topology could become stale.This is particularly problematic because:
DFSNetworkTopologygroup would retain a storage count of zero.After a NameNode restart, this could cause affinity placement to silently fail until a manual:
To avoid this:
DFSNetworkTopology, each affinity group uses a plainNetworkTopology.This ensures storage information remains current.
When:
dfs.use.dfs.network.topology=falsethe configured
net.topology.implis honored instead.For example:
can be used for a node-group-based placement policy. Such implementations do not maintain storage counts and therefore do not have the same stale-storage-accounting issue.
The placement policy's
initialize()still validates that the supplied cluster map is compatible with the configured topology implementation.5. Pipeline Recovery
chooseTarget4AdditionalDatanode()routes through the affinity group's policy only when the surviving pipeline replicas actually belong to that group's restricted topology.Otherwise, it uses the default placement policy.
This distinction is required for non-strict availability mode:
getMaxNodesPerRack()calculation.numChosen + numAdditionalcould exceed the group's leaf count.best-effort=false, the write could fail.Using the default policy for a spilled block avoids this failure.
It also does not introduce an isolation leak because:
Testing
TestDatanodeAffinityBlockPlacementcontains 14 tests covering:All replicas for an affinity path landing only on the group's DataNodes.
Non-affinity paths using the full cluster.
Isolation surviving DataNode re-registration.
Stale-node pruning on DataNode removal.
Strict isolation failing for:
Non-strict fallback succeeding through the shared pool for:
Whole-block spillover with no replica remaining on the under-provisioned affinity group.
Pipeline recovery of a spilled shared-pool block succeeding through the default policy.
In-group pipeline recovery remaining within the affinity group.
Malformed records with null regex values being skipped without aborting refresh.
DataNodes registering during configuration refresh being reconciled into their affinity group instead of leaking into the default topology.
Replaced DataNodes with the same address but a new UUID being reconciled to the live descriptor.
DataNode replacement working both:
Per-group topology being a plain
NetworkTopologyrather than a storage-type-awareDFSNetworkTopology.Storage accounting remaining current for affinity DataNodes.
The removal-race purge reuses
onDatanodeRemoved(), whose pruning behavior is covered by the removal test.The exact mid-refresh interleaving is not unit-tested because reproducing it deterministically would require intrusive production test hooks.
Results
The implementation enables strict tenant isolation of write and read traffic onto a dedicated DataNode pool while keeping the rest of the cluster on the default placement policy.
Affinity configuration can be refreshed at runtime without requiring a NameNode restart.
How was this patch tested?
Tested by newly added unit test
For code changes:
declared according to the connector-specific documentation? Note: Automated CI
testing doesn't cover all cases so manual testing with cloud storage is still
required.
LICENSE,LICENSE-binary,NOTICE-binaryfiles?AI Tooling
If an AI tool was used:
where is the name of the AI tool used.
https://www.apache.org/legal/generative-tooling.html