Skip to content

HDDS-14379. Implement basic Hadoop OM client proxy provider to read from followers #9641

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
szetszwo merged 19 commits into from
Jan 22, 2026
+2,574 −10
Merged

HDDS-14379. Implement basic Hadoop OM client proxy provider to read from followers #9641

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
e05dfad
HDDS-14379. Implement basic Hadoop OM client proxy provider to read f…
ivandika3 Jan 16, 2026
2fe5704
Fix leak in test
ivandika3 Jan 17, 2026
435ef92
Try to use assertThat in slow node test
ivandika3 Jan 18, 2026
f2db9cf
Change from isFollowerReadEnabled to useFollowerRead
ivandika3 Jan 22, 2026
0d663a4
Update SuppressWarnings
ivandika3 Jan 22, 2026
406e34d
Use stream API for combinedInfo
ivandika3 Jan 22, 2026
fdd6d6e
Make parseOMRequest throw ServiceException
ivandika3 Jan 22, 2026
c7e9a44
More detailed error message in parseOMRequest
ivandika3 Jan 22, 2026
d804991
Better declaring class update
ivandika3 Jan 22, 2026
3abd0f9
Check useFollowerRead while iterating in follower
ivandika3 Jan 22, 2026
6a3f9b7
Change wrapInServiceException to throwServiceException
ivandika3 Jan 22, 2026
fb6b942
Remove redundant toString
ivandika3 Jan 22, 2026
77939e5
Remove redundant throws clause
ivandika3 Jan 22, 2026
1a1a591
Remove unused import
ivandika3 Jan 22, 2026
b03db5a
Merge branch 'master' into HDDS-14379
ivandika3 Jan 22, 2026
d6ba3be
Adhere to HDDS-14455
ivandika3 Jan 22, 2026
0bffb04
Revert "Better declaring class update"
ivandika3 Jan 22, 2026
02cbdf9
Handle null args[1] in parseOMRequest
ivandika3 Jan 22, 2026
8592a99
Fix checkstyle
ivandika3 Jan 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
import java.io.File;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.UndeclaredThrowableException;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.nio.file.Path;
Expand Down Expand Up @@ -615,6 +616,9 @@ public static Throwable getUnwrappedException(Exception ex) {
if (t instanceof RemoteException) {
t = ((RemoteException) t).unwrapRemoteException();
}
if (t instanceof UndeclaredThrowableException) {
t = ((UndeclaredThrowableException) t).getUndeclaredThrowable();
}
while (t != null) {
if (t instanceof RpcException ||
t instanceof AccessControlException ||
Expand Down
5 changes: 5 additions & 0 deletions hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -488,6 +488,11 @@ public final class OzoneConfigKeys {
public static final long OZONE_CLIENT_WAIT_BETWEEN_RETRIES_MILLIS_DEFAULT =
2000;

// Ozone Client Follower Read
public static final String OZONE_CLIENT_FOLLOWER_READ_ENABLED_KEY =
"ozone.client.follower.read.enabled";
public static final boolean OZONE_CLIENT_FOLLOWER_READ_ENABLED_DEFAULT = false;

public static final String OZONE_FREON_HTTP_ENABLED_KEY =
"ozone.freon.http.enabled";
public static final String OZONE_FREON_HTTP_BIND_HOST_KEY =
Expand Down
7 changes: 7 additions & 0 deletions hadoop-hdds/common/src/main/resources/ozone-default.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3141,6 +3141,13 @@
wait time is introduced after all the OM proxies have been attempted once.
</description>
</property>
<property>
<name>ozone.client.follower.read.enabled</name>
<value>false</value>
<description>
Enable client to read from OM followers.
</description>
</property>
<property>
<name>ozone.om.admin.protocol.max.retries</name>
<value>20</value>
Expand Down
133 changes: 131 additions & 2 deletions hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,7 @@
import org.apache.hadoop.ozone.om.helpers.RepeatedOmKeyInfo;
import org.apache.hadoop.ozone.om.helpers.ServiceInfo;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand Down Expand Up @@ -228,8 +229,7 @@ public static int getOmRpcPort(ConfigurationSource conf) {
* @param omRequest OMRequest proto
* @return True if its readOnly, false otherwise.
*/
public static boolean isReadOnly(
OzoneManagerProtocolProtos.OMRequest omRequest) {
public static boolean isReadOnly(OMRequest omRequest) {
OzoneManagerProtocolProtos.Type cmdType = omRequest.getCmdType();
switch (cmdType) {
case CheckVolumeAccess:
Expand Down Expand Up @@ -352,6 +352,135 @@ public static boolean isReadOnly(
}
}

/**
* Checks if the OM request should be sent to the follower or leader.
* <p>
* Note that this method is not equivalent to {@link OmUtils#isReadOnly(OMRequest)}
* since there are cases that a "read" requests (ones that do not go through Ratis) requires
* to be sent to the leader.
* @param omRequest OMRequest proto
* @return True if the request should be sent to the follower.
*/
public static boolean shouldSendToFollower(OMRequest omRequest) {
OzoneManagerProtocolProtos.Type cmdType = omRequest.getCmdType();
switch (cmdType) {
case CheckVolumeAccess:
case InfoVolume:
case ListVolume:
case InfoBucket:
case ListBuckets:
case LookupKey:
case ListKeys:
case ListKeysLight:
case ListTrash:
// ListTrash is deprecated by HDDS-11251. Keeping this in here
// As protobuf currently doesn't support deprecating enum fields
// TODO: Remove once migrated to proto3 and mark fields in proto
// as deprecated
case ListOpenFiles:
case ListMultiPartUploadParts:
case GetFileStatus:
case LookupFile:
case ListStatus:
case ListStatusLight:
case GetAcl:
case ListMultipartUploads:
case FinalizeUpgradeProgress:
case PrepareStatus:
case GetS3VolumeContext:
case ListTenant:
case TenantGetUserInfo:
case TenantListUser:
case ListSnapshot:
case RefetchSecretKey:
case GetKeyInfo:
case GetSnapshotInfo:
case GetObjectTagging:
return true;
case CreateVolume:
case SetVolumeProperty:
case DeleteVolume:
case CreateBucket:
case SetBucketProperty:
case DeleteBucket:
case CreateKey:
case RenameKey:
case RenameKeys:
case DeleteKey:
case DeleteKeys:
case CommitKey:
case AllocateBlock:
case InitiateMultiPartUpload:
case CommitMultiPartUpload:
case CompleteMultiPartUpload:
case AbortMultiPartUpload:
case GetS3Secret:
case GetDelegationToken:
case RenewDelegationToken:
case CancelDelegationToken:
case CreateDirectory:
case CreateFile:
case RemoveAcl:
case SetAcl:
case AddAcl:
case PurgeKeys:
case RecoverTrash:
// RecoverTrash is deprecated by HDDS-11251. Keeping this in here
// As protobuf currently doesn't support deprecating enum fields
// TODO: Remove once migrated to proto3 and mark fields in proto
// as deprecated
case FinalizeUpgrade:
case Prepare:
case CancelPrepare:
case DeleteOpenKeys:
case SetS3Secret:
case RevokeS3Secret:
case PurgeDirectories:
case PurgePaths:
case CreateTenant:
case DeleteTenant:
case TenantAssignUserAccessId:
case TenantRevokeUserAccessId:
case TenantAssignAdmin:
case TenantRevokeAdmin:
case SetRangerServiceVersion:
case CreateSnapshot:
case DeleteSnapshot:
case RenameSnapshot:
case SnapshotMoveDeletedKeys:
case SnapshotMoveTableKeys:
case SnapshotPurge:
case RecoverLease:
case SetTimes:
case AbortExpiredMultiPartUploads:
case SetSnapshotProperty:
case QuotaRepair:
case PutObjectTagging:
case DeleteObjectTagging:
case ServiceList: // OM leader should have the most up-to-date OM service list info
case RangerBGSync: // Ranger Background Sync task is only run on leader
case SnapshotDiff:
case CancelSnapshotDiff:
case ListSnapshotDiffJobs:
case PrintCompactionLogDag:
// Snapshot diff is a local to a single OM node so we should not send it arbitrarily
// to any OM nodes
case TransferLeadership: // Transfer leadership should be initiated by the leader
case SetSafeMode: // SafeMode should be initiated by the leader
case StartQuotaRepair:
case GetQuotaRepairStatus:
// Quota repair lifecycle request should be initiated by the leader
case DBUpdates: // We are currently only interested on the leader DB info
case UnknownCommand:
return false;
case EchoRPC:
return omRequest.getEchoRPCRequest().getReadOnly();
default:
LOG.error("CmdType {} is not categorized to be sent to follower.", cmdType);
return false;
}
}

public static byte[] getSHADigest() throws IOException {
try {
SRAND.nextBytes(randomBytes);
Expand Down
9 changes: 7 additions & 2 deletions .../common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFailoverProxyProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@
public class HadoopRpcOMFailoverProxyProvider<T> extends
OMFailoverProxyProviderBase<T> {

private static final Logger LOG =
protected static final Logger LOG =
LoggerFactory.getLogger(HadoopRpcOMFailoverProxyProvider.class);

private final Text delegationTokenService;
Expand Down Expand Up @@ -123,7 +123,7 @@ public synchronized ProxyInfo<T> getProxy() {
/**
* Creates proxy object.
*/
protected ProxyInfo<T> createOMProxyIfNeeded(OMProxyInfo<T> omProxyInfo) {
protected synchronized ProxyInfo<T> createOMProxyIfNeeded(OMProxyInfo<T> omProxyInfo) {
if (omProxyInfo.proxy == null) {
Comment on lines +126 to 127
Copy link
Contributor

@szetszwo szetszwo Jan 20, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The synchronized is for omProxyInfo.proxy. We probably should do it in OMProxyInfo. Let me fix it in HDDS-14470.

Copy link
Contributor Author

@ivandika3 ivandika3 Jan 22, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok will wait until HDDS-14470 is merged.

Copy link
Contributor

@szetszwo szetszwo Jan 22, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let merge this first if it is ready. I can update the change for HDDS-14470.

try {
omProxyInfo.proxy = createOMProxy(omProxyInfo.getAddress());
Expand All @@ -136,6 +136,11 @@ protected ProxyInfo<T> createOMProxyIfNeeded(OMProxyInfo<T> omProxyInfo) {
return omProxyInfo;
}

protected synchronized ProxyInfo<T> createOMProxyIfNeeded(String omNodeId) {
OMProxyInfo<T> omProxyInfo = getOmProxy(omNodeId);
return createOMProxyIfNeeded(omProxyInfo);
}

public Text getCurrentProxyDelegationToken() {
return delegationTokenService;
}
Expand Down
Loading