rdhabalia opened a new pull request, #8715:
URL: https://github.com/apache/hadoop/pull/8715
### 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 `BlockPlacementPolicy` has no notion of path-to-DataNode
affinity, so operators typically resort to separate clusters or brittle
rack-based workarounds.
### Approach
Introduce a pluggable `DatanodeAffinityManager` abstraction, resolved via
reflection from:
```properties
dfs.datanode.affinity.manager.classname
```
An affinity group maps:
* **Source path regex** → **DataNode hostname regex**
On refresh, the manager resolves each DataNode hostname regex against the
live cluster and builds a restricted `NetworkTopology` for each group
containing only that group's eligible DataNodes.
### Components
* **`DatanodeManager`**
* Instantiates the configured affinity manager.
* Removes affinity DataNodes from the default `NetworkTopology` as they
register, ensuring the default placement policy cannot select them.
* Prunes DataNodes from affinity structures when they are decommissioned
or removed.
* Re-triggers an affinity refresh on `dfsadmin -refreshNodes`.
* **`BlockManager`**
* Maintains one `BlockPlacementPolicy` per affinity group.
* Each policy is backed by the group's restricted topology.
* Routes initial block placement (`chooseTarget4NewBlock`) through the
matching group's policy.
* Routes mid-write pipeline recovery (`chooseTarget4AdditionalDatanode`)
through the matching group's policy.
* Ensures replacement nodes for failed pipeline members remain within the
isolated pool.
* Uses the default placement policy when no affinity group matches.
* **`FileDatanodeAffinityManager`**
* Built-in implementation that loads affinity groups from a JSON file.
* Configuration:
```properties
dfs.datanode.affinity.file.path
```
* Reloads the configuration on `dfsadmin -refreshNodes`.
* **`hdfs fsck`**
* Supports:
```bash
hdfs fsck <path> -favored-nodes
```
* 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:
* an under-provisioned group,
* all group DataNodes being unavailable, or
* a `datanodesRegex` matching 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:
```properties
dfs.namenode.affinity.strict.isolation.enabled=true
```
In strict mode, initial placement fails closed unless **all requested
replicas** can be placed inside the affinity group.
This is important because placing only `minReplication` replicas 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:
* Initial block placement.
* In-pipeline recovery.
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:
```text
hostname:port
```
Therefore, a fully anchored pattern must account for the trailing port.
For example:
```regex
^dn-tenant-a[0-9]+\.example\.com(:\d+)?$
```
## Configuration
| Property | Description
| Default |
| ------------------------------------------------ |
--------------------------------------------------------- | ---------------- |
| `dfs.datanode.affinity.manager.classname` | Configured affinity
manager implementation | Empty / disabled |
| `dfs.datanode.affinity.file.path` | JSON configuration file
for `FileDatanodeAffinityManager` | - |
| `dfs.namenode.affinity.strict.isolation.enabled` | Enable strict affinity
isolation | `false` |
## Concurrency 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:
* **Strict mode:** Fail closed.
* **Non-strict mode:** Spill the **entire block** to the shared pool using
the default placement policy.
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:
1. `postAffinityRefresh()` re-evaluates every live DataNode against the
current affinity patterns.
* A DataNode that registered during the refresh is still moved into its
appropriate affinity group.
* The node cannot leak back into the default topology.
2. A post-publish purge re-runs `onDatanodeRemoved()` for any snapshot node
that is no longer live.
* A DataNode removed during refresh cannot remain as a stale entry in the
newly published structures.
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:port` but 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:
* Detects replacement DataNodes.
* Swaps stale descriptors for the live descriptor.
* Uses object identity to guard the replacement.
* Removes stale descriptors for the same transfer endpoint regardless of
which topology path they reside under.
* Handles replacements that also move to a different rack or IP.
`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`.
`DFSNetworkTopology` captures 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
`DFSNetworkTopology` affinity topology could become stale.
This is particularly problematic because:
1. A DataNode registers with an empty storage map.
2. The DataNode is added to its affinity group.
3. Storage reports arrive later.
4. A `DFSNetworkTopology` group would retain a storage count of zero.
5. Storage-type-gated selection could therefore return no target.
After a NameNode restart, this could cause affinity placement to silently
fail until a manual:
```bash
dfsadmin -refreshNodes
```
To avoid this:
* When the cluster uses the default `DFSNetworkTopology`, each affinity
group uses a plain `NetworkTopology`.
* The topology does not maintain storage-type counts.
* The placement policy selects an eligible leaf and validates the storage
type against the **live DataNode descriptor**.
This ensures storage information remains current.
When:
```properties
dfs.use.dfs.network.topology=false
```
the configured `net.topology.impl` is honored instead.
For example:
```text
NetworkTopologyWithNodeGroup
```
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:
1. An under-provisioned affinity group spills the **whole block** to the
shared pool.
2. The surviving pipeline replicas are therefore shared-pool DataNodes.
3. Those DataNodes are not present in the affinity group's restricted
topology.
4. Forcing the affinity policy would count out-of-topology survivors against
the group's `getMaxNodesPerRack()` calculation.
5. `numChosen + numAdditional` could exceed the group's leaf count.
6. The additional target count could become zero.
7. No replacement DataNode would be returned.
8. With the client default `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:
* The block is already entirely in the shared pool.
* Affinity DataNodes are excluded from the default topology.
## Testing
`TestDatanodeAffinityBlockPlacement` contains **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:
* Unsatisfiable placement.
* Partially provisioned groups.
* Non-strict fallback succeeding through the shared pool for:
* Empty groups.
* Partially provisioned groups.
* 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:
* In place.
* When the replacement moves to a different rack/path.
* Per-group topology being a plain `NetworkTopology` rather than a
storage-type-aware `DFSNetworkTopology`.
* 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.
<!--
Thanks for sending a pull request!
1. If this is your first time, please read our contributor guidelines:
https://cwiki.apache.org/confluence/display/HADOOP/How+To+Contribute
2. Make sure your PR title starts with JIRA issue id, e.g.,
'HADOOP-17799. Your PR title ...'.
-->
### How was this patch tested?
Tested by newly added unit test
### For code changes:
- [ ] Does the title of this PR start with the corresponding JIRA issue id
(e.g. 'HADOOP-17799. Your PR title ...')?
- [ ] Object storage: Have the integration tests been executed and the
endpoint
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.*
- [ ] If adding new dependencies to the code, are these dependencies
licensed in a way that is compatible for inclusion under [ASF
2.0](http://www.apache.org/legal/resolved.html#category-a)?
- [ ] If applicable, have you updated the `LICENSE`, `LICENSE-binary`,
`NOTICE-binary` files?
### AI Tooling
If an AI tool was used:
- [ ] The PR includes the phrase "Contains content generated by <tool>"
where <tool> is the name of the AI tool used.
- [ ] My use of AI contributions follows the ASF legal policy
https://www.apache.org/legal/generative-tooling.html
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
