diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java index 84b49e1b57e0..9d7f5c6ab73a 100644 --- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java +++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java @@ -62,6 +62,7 @@ import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs; import org.apache.hadoop.ozone.om.helpers.OzoneFileStatus; import org.apache.hadoop.ozone.om.helpers.OzoneFileStatusLight; +import org.apache.hadoop.ozone.om.helpers.ReadConsistency; import org.apache.hadoop.ozone.om.helpers.S3SecretValue; import org.apache.hadoop.ozone.om.helpers.S3VolumeContext; import org.apache.hadoop.ozone.om.helpers.TenantStateList; @@ -1378,17 +1379,44 @@ OzoneKey headObject(String volumeName, String bucketName, */ void setThreadLocalS3Auth(S3Auth s3Auth); + /** + * Sets the read consistency hint for the current request thread. + * @param readConsistency read consistency selected by the client. + */ + void setThreadLocalReadConsistency(ReadConsistency readConsistency); + + /** + * Sets the read consistency hint and optional local lease context for the + * current request thread. + * @param readConsistency read consistency selected by the client. + * @param localLeaseLogLimit optional local lease log limit. + * @param localLeaseTimeMs optional local lease duration in milliseconds. + */ + void setThreadLocalReadConsistency(ReadConsistency readConsistency, + Long localLeaseLogLimit, Long localLeaseTimeMs); + /** * Gets the S3 Authentication information that is attached to the thread. * @return S3 Authentication information. */ S3Auth getThreadLocalS3Auth(); + /** + * Gets the read consistency hint that is attached to the thread. + * @return request-local read consistency. + */ + ReadConsistency getThreadLocalReadConsistency(); + /** * Clears the S3 Authentication information attached to the thread. */ void clearThreadLocalS3Auth(); + /** + * Clears the read consistency hint attached to the thread. + */ + void clearThreadLocalReadConsistency(); + default ThreadLocal getS3CredentialsProvider() { return null; } diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java index 9c121bde4c89..0b91d0be9579 100644 --- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java +++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java @@ -162,6 +162,7 @@ import org.apache.hadoop.ozone.om.helpers.OzoneFSUtils; import org.apache.hadoop.ozone.om.helpers.OzoneFileStatus; import org.apache.hadoop.ozone.om.helpers.OzoneFileStatusLight; +import org.apache.hadoop.ozone.om.helpers.ReadConsistency; import org.apache.hadoop.ozone.om.helpers.S3SecretValue; import org.apache.hadoop.ozone.om.helpers.S3VolumeContext; import org.apache.hadoop.ozone.om.helpers.ServiceInfo; @@ -2922,16 +2923,38 @@ public void setThreadLocalS3Auth( this.s3gUgi = UserGroupInformation.createRemoteUser(getThreadLocalS3Auth().getUserPrincipal()); } + @Override + public void setThreadLocalReadConsistency(ReadConsistency readConsistency) { + ozoneManagerClient.setThreadLocalReadConsistency(readConsistency); + } + + @Override + public void setThreadLocalReadConsistency(ReadConsistency readConsistency, + Long localLeaseLogLimit, Long localLeaseTimeMs) { + ozoneManagerClient.setThreadLocalReadConsistency(readConsistency, + localLeaseLogLimit, localLeaseTimeMs); + } + @Override public S3Auth getThreadLocalS3Auth() { return ozoneManagerClient.getThreadLocalS3Auth(); } + @Override + public ReadConsistency getThreadLocalReadConsistency() { + return ozoneManagerClient.getThreadLocalReadConsistency(); + } + @Override public void clearThreadLocalS3Auth() { ozoneManagerClient.clearThreadLocalS3Auth(); } + @Override + public void clearThreadLocalReadConsistency() { + ozoneManagerClient.clearThreadLocalReadConsistency(); + } + @Override public ThreadLocal getS3CredentialsProvider() { return ozoneManagerClient.getS3CredentialsProvider(); diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java index 38a7bbbb5bb2..f96913ce50eb 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFollowerReadFailoverProxyProvider.java @@ -76,11 +76,11 @@ public class HadoopRpcOMFollowerReadFailoverProxyProvider implements FailoverPro /** The combined proxy which redirects to other proxies as necessary. */ private final ProxyInfo combinedProxy; - /** - * Whether reading from follower is enabled. If this is false, all read - * requests will still go to OM leader. - */ - private volatile boolean useFollowerRead; + /** Whether follower reads are supported by the OM service. */ + private volatile boolean omServiceSupportsFollowerRead; + + /** Whether eligible reads without an explicit consistency hint use followers. */ + private final boolean defaultFollowerReadEnabled; /** * The current index of the underlying leader-based proxy provider's omNodesInOrder currently being used. @@ -96,17 +96,11 @@ public class HadoopRpcOMFollowerReadFailoverProxyProvider implements FailoverPro /** The read consistency hint used when follower read is disabled or when follower read fails. */ private final ReadConsistencyHint leaderReadConsistency; - public HadoopRpcOMFollowerReadFailoverProxyProvider( - HadoopRpcOMFailoverProxyProvider leaderProxy - ) { - this(leaderProxy, ReadConsistency.LINEARIZABLE_ALLOW_FOLLOWER, ReadConsistency.DEFAULT, true); - } - public HadoopRpcOMFollowerReadFailoverProxyProvider( HadoopRpcOMFailoverProxyProvider leaderProxy, ReadConsistency followerReadConsistencyType, ReadConsistency leaderReadConsistencyType, - boolean useFollowerRead) { + boolean defaultFollowerReadEnabled) { Preconditions.assertTrue(followerReadConsistencyType.allowFollowerRead(), "Invalid follower read consistency " + followerReadConsistencyType); Preconditions.assertTrue(!leaderReadConsistencyType.allowFollowerRead(), @@ -121,7 +115,8 @@ public HadoopRpcOMFollowerReadFailoverProxyProvider( FollowerReadInvocationHandler.class.getClassLoader(), new Class[] {OzoneManagerProtocolPB.class}, new FollowerReadInvocationHandler()); combinedProxy = new ProxyInfo<>(wrappedProxy, combinedInfo); - this.useFollowerRead = useFollowerRead; + this.omServiceSupportsFollowerRead = true; + this.defaultFollowerReadEnabled = defaultFollowerReadEnabled; this.followerReadConsistency = followerReadConsistencyType.getHint(); this.leaderReadConsistency = leaderReadConsistencyType.getHint(); } @@ -178,6 +173,19 @@ private static OMRequest parseOMRequest(Object[] args) throws ServiceException { return (OMRequest) args[1]; } + private static boolean allowFollowerRead(OMRequest omRequest) { + return !omRequest.hasReadConsistencyHint() + || ReadConsistency.fromProto(omRequest.getReadConsistencyHint() + .getReadConsistency()).allowFollowerRead(); + } + + private static ReadConsistency getReadConsistency(OMRequest omRequest) { + return omRequest.hasReadConsistencyHint() + ? ReadConsistency.fromProto(omRequest.getReadConsistencyHint() + .getReadConsistency()) + : ReadConsistency.DEFAULT; + } + @VisibleForTesting public ProxyInfo getLastProxy() { return lastProxy; @@ -218,6 +226,23 @@ private synchronized OMProxyInfo changeProxy(OMProxyInfo return currentProxy; } + private OMProxyInfo getCurrentProxy( + ReadConsistency readConsistency) { + OMProxyInfo current = getCurrentProxy(); + if (readConsistency != ReadConsistency.LOCAL_LEASE) { + return current; + } + + String leaderNodeId = leaderProxy.getCurrentProxyOMNodeId(); + for (int i = 0; i < leaderProxy.getOMProxyMap().size(); i++) { + if (!current.getNodeId().equals(leaderNodeId)) { + return current; + } + current = changeProxy(current); + } + return null; + } + /** * An InvocationHandler to handle incoming requests. This class's invoke * method contains the primary logic for redirecting to followers. @@ -242,12 +267,10 @@ public Object invoke(Object proxy, final Method method, final Object[] args) OMRequest omRequest = parseOMRequest(args); // Apply default consistency hint once, before any routing decision. - // In the future, we will support per-request hints which allows client (e.g. S3 clients) - // to specify a custom request header (e.g. x-ozone-read-consistency) as a consistency hint - // for read requests. - boolean isFollowerReadEligible = useFollowerRead && OmUtils.shouldSendToFollower(omRequest); + boolean isReadRequest = OmUtils.shouldSendToFollower(omRequest); if (!omRequest.hasReadConsistencyHint()) { - final ReadConsistencyHint defaultReadConsistency = isFollowerReadEligible + final ReadConsistencyHint defaultReadConsistency = omServiceSupportsFollowerRead + && defaultFollowerReadEnabled && isReadRequest ? followerReadConsistency : leaderReadConsistency; if (defaultReadConsistency != null) { omRequest = omRequest.toBuilder() @@ -256,11 +279,22 @@ public Object invoke(Object proxy, final Method method, final Object[] args) args[1] = omRequest; } } + ReadConsistency readConsistency = getReadConsistency(omRequest); + // Requests with explicit hints (for example S3 read consistency headers) + // can narrow routing below, such as forcing leader-only reads. + boolean isExplicitFollowerRead = omRequest.hasReadConsistencyHint() + && allowFollowerRead(omRequest); + boolean isFollowerReadEligible = omServiceSupportsFollowerRead && isReadRequest + && (defaultFollowerReadEnabled || isExplicitFollowerRead); if (isFollowerReadEligible) { int failedCount = 0; - for (int i = 0; useFollowerRead && i < leaderProxy.getOMProxyMap().size(); i++) { - OMProxyInfo current = getCurrentProxy(); + for (int i = 0; i < leaderProxy.getOMProxyMap().size(); i++) { + OMProxyInfo current = + getCurrentProxy(readConsistency); + if (current == null) { + break; + } LOG.debug("Attempting to service {} with cmdType {} using proxy {}", method.getName(), omRequest.getCmdType(), current.proxyInfo); try { @@ -292,7 +326,7 @@ public Object invoke(Object proxy, final Method method, final Object[] args) // the OM follower does not support / disable follower read or something is misconfigured LOG.debug("Encountered OMNotLeaderException from {}. " + "Disable OM follower read and retry OM leader directly.", current.proxyInfo); - useFollowerRead = false; + omServiceSupportsFollowerRead = false; // Break here instead of throwing exception so that it is not counted // as a failover break; @@ -363,19 +397,24 @@ public Object invoke(Object proxy, final Method method, final Object[] args) // Either all followers have failed, follower reads are disabled, // or this is a write request. In any case, forward the request to // the leader OM. + return invokeLeader(method, args); + } + + private Object invokeLeader(Method method, Object[] args) throws Throwable { LOG.debug("Using leader-based failoverProxy to service {}", method.getName()); - final OMProxyInfo currentLeaderProxy = leaderProxy.getProxy(); - Object retVal = null; + final OMProxyInfo currentLeaderProxy = + leaderProxy.getProxy(); try { - retVal = method.invoke(currentLeaderProxy.getProxy(), args); + Object retVal = method.invoke(currentLeaderProxy.getProxy(), args); + lastProxy = currentLeaderProxy; + return retVal; } catch (InvocationTargetException e) { LOG.debug("Exception thrown from leader-based failoverProxy", e.getCause()); // This exception will be handled by the OMFailoverProxyProviderBase#getRetryPolicy // (see getRetryPolicy). This ensures that the leader-only failover should still work. throwServiceException(e.getCause()); + return null; } - lastProxy = currentLeaderProxy; - return retVal; } @Override @@ -395,7 +434,7 @@ public ConnectionId getConnectionId() { // visibility hazard, not a tearing one -- but the outcome is the // same: a stale proxy whose underlying connection has been // stopped is dialed instead of the live replacement. - return RPC.getConnectionIdForProxy(useFollowerRead + return RPC.getConnectionIdForProxy(omServiceSupportsFollowerRead ? getCurrentProxy().getProxy() : leaderProxy.getProxy().getProxy()); } } @@ -408,8 +447,8 @@ public synchronized void close() throws IOException { } @VisibleForTesting - public boolean isUseFollowerRead() { - return useFollowerRead; + public boolean isOmServiceSupportsFollowerRead() { + return omServiceSupportsFollowerRead; } @VisibleForTesting diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/GrpcOmTransport.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/GrpcOmTransport.java index 21baa053e44d..f7bd799ebb74 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/GrpcOmTransport.java +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/GrpcOmTransport.java @@ -99,7 +99,8 @@ public class GrpcOmTransport implements OmTransport { private RetryPolicy retryPolicy; private final GrpcOMFailoverProxyProvider omFailoverProxyProvider; - private volatile boolean useFollowerRead; + private final boolean defaultFollowerReadEnabled; + private volatile boolean omServiceSupportsFollowerRead; private final ReadConsistencyHint followerReadConsistency; private final ReadConsistencyHint leaderReadConsistency; private int currentFollowerReadIndex = -1; @@ -128,9 +129,10 @@ public GrpcOmTransport(ConfigurationSource conf, omServiceId, OzoneManagerProtocolPB.class); - this.useFollowerRead = conf.getBoolean( + this.defaultFollowerReadEnabled = conf.getBoolean( OzoneConfigKeys.OZONE_CLIENT_FOLLOWER_READ_ENABLED_KEY, OzoneConfigKeys.OZONE_CLIENT_FOLLOWER_READ_ENABLED_DEFAULT); + this.omServiceSupportsFollowerRead = true; String defaultFollowerReadConsistencyStr = conf.get( OzoneConfigKeys.OZONE_CLIENT_FOLLOWER_READ_DEFAULT_CONSISTENCY_KEY, OzoneConfigKeys.OZONE_CLIENT_FOLLOWER_READ_DEFAULT_CONSISTENCY_DEFAULT @@ -206,20 +208,29 @@ public void start() throws IOException { @Override public OMResponse submitRequest(OMRequest payload) throws IOException { - if (useFollowerRead && OmUtils.shouldSendToFollower(payload)) { + if (shouldUseFollowerRead(payload)) { return submitRequestWithFollowerRead(payload); } return submitRequestToLeader(addReadConsistencyHint(payload, leaderReadConsistency)); } + private boolean shouldUseFollowerRead(OMRequest payload) { + if (!omServiceSupportsFollowerRead || !OmUtils.shouldSendToFollower(payload)) { + return false; + } + return defaultFollowerReadEnabled || payload.hasReadConsistencyHint() + && ReadConsistency.fromProto(payload.getReadConsistencyHint() + .getReadConsistency()).allowFollowerRead(); + } + private OMResponse submitRequestWithFollowerRead(OMRequest payload) throws IOException { OMRequest followerPayload = addReadConsistencyHint(payload, followerReadConsistency); int failedCount = 0; - for (int i = 0; useFollowerRead && - i < omFailoverProxyProvider.getOMProxyMap().getNodeIds().size(); i++) { + for (int i = 0; + i < omFailoverProxyProvider.getOMProxyMap().getNodeIds().size(); i++) { String nodeId = getCurrentFollowerReadNodeId(); String followerHost = omFailoverProxyProvider.getGrpcProxyAddress(nodeId); try { @@ -234,7 +245,7 @@ private OMResponse submitRequestWithFollowerRead(OMRequest payload) if (OMFailoverProxyProviderBase.getNotLeaderException(unwrapped) != null) { LOG.debug("Encountered OMNotLeaderException from {}. Disable OM follower read and retry OM leader directly.", followerHost); - useFollowerRead = false; + omServiceSupportsFollowerRead = false; break; } if (OMFailoverProxyProviderBase.getLeaderNotReadyException(unwrapped) != null) { diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/Hadoop3OmTransport.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/Hadoop3OmTransport.java index 98fcd66e7622..f09d7e19a2fb 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/Hadoop3OmTransport.java +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/Hadoop3OmTransport.java @@ -64,7 +64,7 @@ public Hadoop3OmTransport(ConfigurationSource conf, this.omFailoverProxyProvider = new HadoopRpcOMFailoverProxyProvider<>( conf, ugi, omServiceId, OzoneManagerProtocolPB.class); - boolean followerReadEnabled = conf.getBoolean( + boolean defaultFollowerReadEnabled = conf.getBoolean( OzoneConfigKeys.OZONE_CLIENT_FOLLOWER_READ_ENABLED_KEY, OzoneConfigKeys.OZONE_CLIENT_FOLLOWER_READ_ENABLED_DEFAULT); @@ -90,7 +90,7 @@ public Hadoop3OmTransport(ConfigurationSource conf, new HadoopRpcOMFollowerReadFailoverProxyProvider(omFailoverProxyProvider, defaultFollowerReadConsistency, defaultLeaderReadConsistency, - followerReadEnabled); + defaultFollowerReadEnabled); this.rpcProxy = OzoneManagerProtocolPB.newProxy(followerReadFailoverProxyProvider, maxFailovers); } diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerClientProtocol.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerClientProtocol.java index de38f275c6e3..00adf4f49af9 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerClientProtocol.java +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerClientProtocol.java @@ -17,6 +17,7 @@ package org.apache.hadoop.ozone.om.protocolPB; +import org.apache.hadoop.ozone.om.helpers.ReadConsistency; import org.apache.hadoop.ozone.om.protocol.OzoneManagerProtocol; import org.apache.hadoop.ozone.om.protocol.S3Auth; @@ -33,9 +34,18 @@ public interface OzoneManagerClientProtocol extends OzoneManagerProtocol { */ void setThreadLocalS3Auth(S3Auth s3Auth); + void setThreadLocalReadConsistency(ReadConsistency readConsistency); + + void setThreadLocalReadConsistency(ReadConsistency readConsistency, + Long localLeaseLogLimit, Long localLeaseTimeMs); + S3Auth getThreadLocalS3Auth(); + ReadConsistency getThreadLocalReadConsistency(); + void clearThreadLocalS3Auth(); + void clearThreadLocalReadConsistency(); + ThreadLocal getS3CredentialsProvider(); } diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java index b5a9e3a2bbcf..3be03adb1a73 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java @@ -90,6 +90,7 @@ import org.apache.hadoop.ozone.om.helpers.OpenKeySession; import org.apache.hadoop.ozone.om.helpers.OzoneFileStatus; import org.apache.hadoop.ozone.om.helpers.OzoneFileStatusLight; +import org.apache.hadoop.ozone.om.helpers.ReadConsistency; import org.apache.hadoop.ozone.om.helpers.S3SecretValue; import org.apache.hadoop.ozone.om.helpers.S3VolumeContext; import org.apache.hadoop.ozone.om.helpers.ServiceInfo; @@ -207,6 +208,7 @@ import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PutObjectTaggingRequest; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.RangerBGSyncRequest; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.RangerBGSyncResponse; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.ReadConsistencyHint; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.RecoverLeaseRequest; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.RecoverLeaseResponse; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.RefetchSecretKeyRequest; @@ -280,6 +282,8 @@ public final class OzoneManagerProtocolClientSideTranslatorPB private OmTransport transport; private ThreadLocal threadLocalS3Auth = new ThreadLocal<>(); + private ThreadLocal threadLocalReadConsistencyHint + = new ThreadLocal<>(); private boolean s3AuthCheck; public static final int BLOCK_ALLOCATION_RETRY_COUNT = 90; @@ -357,6 +361,10 @@ private OMResponse submitRequest(OMRequest omRequest) throw new IllegalArgumentException("S3 Auth expected to " + "be set but is null " + omRequest.toString()); } + if (!builder.hasReadConsistencyHint() + && threadLocalReadConsistencyHint.get() != null) { + builder.setReadConsistencyHint(threadLocalReadConsistencyHint.get()); + } if (threadLocalS3Auth.get() != null) { if (!Strings.isNullOrEmpty(threadLocalS3Auth.get().getAccessID())) { String caller = OM_S3_CALLER_CONTEXT_PREFIX + @@ -2218,12 +2226,45 @@ public void setThreadLocalS3Auth( this.threadLocalS3Auth.set(s3Auth); } + @Override + @SkipTracing + public void setThreadLocalReadConsistency(ReadConsistency readConsistency) { + this.threadLocalReadConsistencyHint.set(readConsistency.getHint()); + } + + @Override + @SkipTracing + public void setThreadLocalReadConsistency(ReadConsistency readConsistency, + Long localLeaseLogLimit, Long localLeaseTimeMs) { + ReadConsistencyHint.Builder hint = ReadConsistencyHint.newBuilder() + .setReadConsistency(readConsistency.toProto()); + if (readConsistency == ReadConsistency.LOCAL_LEASE + && (localLeaseLogLimit != null || localLeaseTimeMs != null)) { + ReadConsistencyHint.LocalLeaseContext.Builder localLeaseContext = + ReadConsistencyHint.LocalLeaseContext.newBuilder(); + if (localLeaseLogLimit != null) { + localLeaseContext.setLogLimit(localLeaseLogLimit); + } + if (localLeaseTimeMs != null) { + localLeaseContext.setLeaseTimeMs(localLeaseTimeMs); + } + hint.setLocalLeaseContext(localLeaseContext); + } + this.threadLocalReadConsistencyHint.set(hint.build()); + } + @Override @SkipTracing public void clearThreadLocalS3Auth() { this.threadLocalS3Auth.remove(); } + @Override + @SkipTracing + public void clearThreadLocalReadConsistency() { + this.threadLocalReadConsistencyHint.remove(); + } + @Override @SkipTracing public ThreadLocal getS3CredentialsProvider() { @@ -2236,6 +2277,16 @@ public S3Auth getThreadLocalS3Auth() { return this.threadLocalS3Auth.get(); } + @Override + @SkipTracing + public ReadConsistency getThreadLocalReadConsistency() { + ReadConsistencyHint hint = this.threadLocalReadConsistencyHint.get(); + if (hint == null) { + return null; + } + return ReadConsistency.fromProto(hint.getReadConsistency()); + } + /** * Get File Status for an Ozone key. * diff --git a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/ha/TestHadoopRpcOMFollowerReadFailoverProxyProvider.java b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/ha/TestHadoopRpcOMFollowerReadFailoverProxyProvider.java index d3275d22be6b..bb59244f441c 100644 --- a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/ha/TestHadoopRpcOMFollowerReadFailoverProxyProvider.java +++ b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/ha/TestHadoopRpcOMFollowerReadFailoverProxyProvider.java @@ -18,6 +18,7 @@ package org.apache.hadoop.ozone.om.ha; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_FAILOVER_MAX_ATTEMPTS_KEY; +import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_FOLLOWER_READ_ENABLED_KEY; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_WAIT_BETWEEN_RETRIES_MILLIS_KEY; import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ADDRESS_KEY; import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_NODES_KEY; @@ -58,6 +59,7 @@ import org.apache.hadoop.ozone.ha.ConfUtils; import org.apache.hadoop.ozone.om.exceptions.OMLeaderNotReadyException; import org.apache.hadoop.ozone.om.exceptions.OMNotLeaderException; +import org.apache.hadoop.ozone.om.helpers.ReadConsistency; import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolPB; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.CreateKeyRequest; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.GetKeyInfoRequest; @@ -95,7 +97,7 @@ void testWriteOperationOnLeader() throws Exception { doWrite(); assertHandledBy(2); - assertTrue(proxyProvider.isUseFollowerRead()); + assertTrue(proxyProvider.isOmServiceSupportsFollowerRead()); // Although the write request is forwarded to the leader, // the follower read proxy provider should still point to first OM follower assertEquals(proxyProvider.getCurrentProxy().getNodeId(), omNodeIds[0]); @@ -123,7 +125,7 @@ void testWriteOperationOnLeaderNotReady() throws Exception { "Write operation finished earlier than expected"); assertHandledBy(0); - assertTrue(proxyProvider.isUseFollowerRead()); + assertTrue(proxyProvider.isOmServiceSupportsFollowerRead()); } @Test @@ -160,7 +162,62 @@ void testReadOperationOnFollower() throws Exception { doRead(); assertHandledBy(0); - assertTrue(proxyProvider.isUseFollowerRead()); + assertTrue(proxyProvider.isOmServiceSupportsFollowerRead()); + } + + @Test + void testExplicitFollowerReadWhenDisabledByDefault() throws Exception { + OzoneConfiguration config = new OzoneConfiguration(); + config.setBoolean(OZONE_CLIENT_FOLLOWER_READ_ENABLED_KEY, false); + setupProxyProvider(3, config); + omNodeAnswers[0].isLeader = true; + + doRead(); + assertHandledBy(0); + + doRead(ReadConsistency.LOCAL_LEASE); + assertHandledBy(1); + } + + @Test + void testLinearizableAllowFollowerReadSticksToCurrentProxy() + throws Exception { + setupProxyProvider(3); + omNodeAnswers[2].isLeader = true; + + doRead(ReadConsistency.LINEARIZABLE_ALLOW_FOLLOWER); + assertHandledBy(0); + doRead(ReadConsistency.LINEARIZABLE_ALLOW_FOLLOWER); + assertHandledBy(0); + doRead(ReadConsistency.LINEARIZABLE_ALLOW_FOLLOWER); + assertHandledBy(0); + } + + @Test + void testLocalLeaseReadSticksToFollowerBeforeLeader() + throws Exception { + setupProxyProvider(3); + omNodeAnswers[1].isLeader = true; + doWrite(); + + doRead(ReadConsistency.LOCAL_LEASE); + assertHandledBy(0); + doRead(ReadConsistency.LOCAL_LEASE); + assertHandledBy(0); + doRead(ReadConsistency.LOCAL_LEASE); + assertHandledBy(0); + } + + @Test + void testLeaderOnlyReadBypassesFollowerReadProxy() throws Exception { + setupProxyProvider(3); + omNodeAnswers[0].isLeader = true; + + doRead(ReadConsistency.LINEARIZABLE_LEADER_ONLY); + + assertHandledBy(0); + assertTrue(proxyProvider.isOmServiceSupportsFollowerRead()); + assertEquals(proxyProvider.getCurrentProxy().getNodeId(), omNodeIds[0]); } @Test @@ -172,7 +229,7 @@ void testReadOperationOnLeader() throws Exception { // Follower read can still read from OM leader assertHandledBy(0); - assertTrue(proxyProvider.isUseFollowerRead()); + assertTrue(proxyProvider.isOmServiceSupportsFollowerRead()); } @Test @@ -197,7 +254,7 @@ void testReadOperationOnLeaderNotReady() throws Exception { "Read operation finished earlier than expected"); assertHandledBy(0); - assertTrue(proxyProvider.isUseFollowerRead()); + assertTrue(proxyProvider.isOmServiceSupportsFollowerRead()); } @Test @@ -214,7 +271,7 @@ void testReadOperationOnFollowerWhenFollowerReadUnsupported() throws Exception { assertHandledBy(1); // Since OMNotLeaderException is thrown during follower read, the // proxy will keep sending reads from the leader from now on - assertFalse(proxyProvider.isUseFollowerRead()); + assertFalse(proxyProvider.isOmServiceSupportsFollowerRead()); // Try to simulate leader change omNodeAnswers[1].isLeader = false; @@ -223,7 +280,7 @@ void testReadOperationOnFollowerWhenFollowerReadUnsupported() throws Exception { doRead(); assertHandledBy(2); - assertFalse(proxyProvider.isUseFollowerRead()); + assertFalse(proxyProvider.isOmServiceSupportsFollowerRead()); } @Test @@ -356,7 +413,9 @@ void testReadException() throws Exception { } private void setupProxyProvider(int omNodeCount) throws Exception { - setupProxyProvider(omNodeCount, new OzoneConfiguration()); + OzoneConfiguration config = new OzoneConfiguration(); + config.setBoolean(OZONE_CLIENT_FOLLOWER_READ_ENABLED_KEY, true); + setupProxyProvider(omNodeCount, config); } private void setupProxyProvider(int omNodeCount, OzoneConfiguration config) throws Exception { @@ -435,8 +494,12 @@ protected List> initOmProxiesFromConfigs( }; // Wrap the leader-based failover proxy provider with follower read proxy provider - proxyProvider = new HadoopRpcOMFollowerReadFailoverProxyProvider(underlyingProxyProvider); - assertTrue(proxyProvider.isUseFollowerRead()); + boolean defaultFollowerReadEnabled = config.getBoolean( + OZONE_CLIENT_FOLLOWER_READ_ENABLED_KEY, true); + proxyProvider = new HadoopRpcOMFollowerReadFailoverProxyProvider( + underlyingProxyProvider, ReadConsistency.LINEARIZABLE_ALLOW_FOLLOWER, + ReadConsistency.DEFAULT, defaultFollowerReadEnabled); + assertTrue(proxyProvider.isOmServiceSupportsFollowerRead()); // Wrap the follower read proxy provider in retry proxy to allow automatic failover retryProxy = (OzoneManagerProtocolPB) RetryProxy.create( OzoneManagerProtocolPB.class, proxyProvider, @@ -451,6 +514,10 @@ private void doRead() throws Exception { doRead(retryProxy); } + private void doRead(ReadConsistency readConsistency) throws Exception { + doRead(retryProxy, readConsistency); + } + private void doWrite() throws Exception { doWrite(retryProxy); } @@ -475,6 +542,11 @@ private static void doWrite(OzoneManagerProtocolPB client) throws Exception { } private static void doRead(OzoneManagerProtocolPB client) throws Exception { + doRead(client, null); + } + + private static void doRead(OzoneManagerProtocolPB client, + ReadConsistency readConsistency) throws Exception { KeyArgs keyArgs = KeyArgs.newBuilder() .setVolumeName("volume") .setBucketName("bucket") @@ -483,14 +555,16 @@ private static void doRead(OzoneManagerProtocolPB client) throws Exception { GetKeyInfoRequest.Builder req = GetKeyInfoRequest.newBuilder() .setKeyArgs(keyArgs); - OMRequest omRequest = OMRequest.newBuilder() + OMRequest.Builder omRequest = OMRequest.newBuilder() .setVersion(ClientVersion.CURRENT_VERSION) .setClientId(ClientId.randomId().toString()) .setCmdType(Type.GetKeyInfo) - .setGetKeyInfoRequest(req) - .build(); + .setGetKeyInfoRequest(req); + if (readConsistency != null) { + omRequest.setReadConsistencyHint(readConsistency.getHint()); + } - client.submitRequest(null, omRequest); + client.submitRequest(null, omRequest.build()); } private void assertHandledBy(int omNodeIdx) { @@ -509,7 +583,6 @@ private static class OMAnswer { private volatile boolean isFollowerReadSupported = true; private volatile boolean isThrowReadIndexException = false; private volatile boolean isThrowReadException = false; - private OMProtocolAnswer clientAnswer = new OMProtocolAnswer(); private class OMProtocolAnswer implements Answer { @@ -545,6 +618,18 @@ public OMResponse answer(InvocationOnMock invocationOnMock) throws Throwable { break; case GetKeyInfo: if (!isLeader) { + if (omRequest.hasReadConsistencyHint() + && ReadConsistency.fromProto(omRequest + .getReadConsistencyHint() + .getReadConsistency()) == + ReadConsistency.LINEARIZABLE_LEADER_ONLY) { + throw new ServiceException( + new RemoteException( + OMNotLeaderException.class.getCanonicalName(), + "Read can only be done on leader" + ) + ); + } if (!isFollowerReadSupported) { throw new ServiceException( new RemoteException( diff --git a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/protocolPB/TestOzoneManagerProtocolClientSideTranslatorPB.java b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/protocolPB/TestOzoneManagerProtocolClientSideTranslatorPB.java index e08d1e35df74..b0cd17e6e8bf 100644 --- a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/protocolPB/TestOzoneManagerProtocolClientSideTranslatorPB.java +++ b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/protocolPB/TestOzoneManagerProtocolClientSideTranslatorPB.java @@ -28,8 +28,11 @@ import java.io.IOException; import java.util.Collections; import java.util.List; +import org.apache.hadoop.io.Text; +import org.apache.hadoop.ozone.om.helpers.ReadConsistency; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.ServiceListResponse; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.StartQuotaRepairRequest; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.StartQuotaRepairResponse; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Status; @@ -99,4 +102,92 @@ void testStartQuotaRepairWithNullBuckets() { verifyNoInteractions(omTransport); } + @Test + void submitRequestAddsThreadLocalReadConsistencyHint() throws Exception { + CapturingTransport transport = new CapturingTransport(); + OzoneManagerProtocolClientSideTranslatorPB client = + new OzoneManagerProtocolClientSideTranslatorPB(transport, "client-id"); + + client.setThreadLocalReadConsistency(ReadConsistency.LOCAL_LEASE); + + client.getServiceList(); + + assertThat(transport.getLastRequest().hasReadConsistencyHint()).isTrue(); + assertThat(transport.getLastRequest() + .getReadConsistencyHint() + .getReadConsistency()) + .isEqualTo(ReadConsistency.LOCAL_LEASE.toProto()); + } + + @Test + void submitRequestAddsThreadLocalLocalLeaseContext() throws Exception { + CapturingTransport transport = new CapturingTransport(); + OzoneManagerProtocolClientSideTranslatorPB client = + new OzoneManagerProtocolClientSideTranslatorPB(transport, "client-id"); + + client.setThreadLocalReadConsistency(ReadConsistency.LOCAL_LEASE, + 10L, 100L); + + client.getServiceList(); + + assertThat(transport.getLastRequest().hasReadConsistencyHint()).isTrue(); + assertThat(transport.getLastRequest() + .getReadConsistencyHint() + .getReadConsistency()) + .isEqualTo(ReadConsistency.LOCAL_LEASE.toProto()); + assertThat(transport.getLastRequest() + .getReadConsistencyHint() + .hasLocalLeaseContext()) + .isTrue(); + assertThat(transport.getLastRequest() + .getReadConsistencyHint() + .getLocalLeaseContext() + .getLogLimit()) + .isEqualTo(10L); + assertThat(transport.getLastRequest() + .getReadConsistencyHint() + .getLocalLeaseContext() + .getLeaseTimeMs()) + .isEqualTo(100L); + } + + @Test + void submitRequestOmitsReadConsistencyHintByDefault() throws Exception { + CapturingTransport transport = new CapturingTransport(); + OzoneManagerProtocolClientSideTranslatorPB client = + new OzoneManagerProtocolClientSideTranslatorPB(transport, "client-id"); + + client.getServiceList(); + + assertThat(transport.getLastRequest().hasReadConsistencyHint()).isFalse(); + } + + private static final class CapturingTransport implements OmTransport { + private OMRequest lastRequest; + + @Override + public OMResponse submitRequest(OMRequest payload) { + lastRequest = payload; + return OMResponse.newBuilder() + .setCmdType(payload.getCmdType()) + .setStatus(Status.OK) + .setSuccess(true) + .setServiceListResponse(ServiceListResponse.newBuilder()) + .build(); + } + + @Override + public Text getDelegationTokenService() { + return new Text(); + } + + @Override + public void close() throws IOException { + } + + private OMRequest getLastRequest() { + return lastRequest; + } + } + } diff --git a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/protocolPB/TestS3GrpcOmTransport.java b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/protocolPB/TestS3GrpcOmTransport.java index 3d2a5fedda34..63d51d77ea28 100644 --- a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/protocolPB/TestS3GrpcOmTransport.java +++ b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/protocolPB/TestS3GrpcOmTransport.java @@ -372,6 +372,46 @@ public void testFollowerReadKeepsExistingConsistencyHint() throws Exception { followerRequest.get().getReadConsistencyHint().getReadConsistency()); } + @Test + public void testExplicitFollowerReadWhenDisabledByDefault() throws Exception { + conf.setBoolean(OzoneConfigKeys.OZONE_CLIENT_FOLLOWER_READ_ENABLED_KEY, false); + configureHaOmService("om0", "om1"); + + AtomicInteger leaderRequestCount = new AtomicInteger(); + AtomicInteger followerRequestCount = new AtomicInteger(); + AtomicReference leaderRequest = new AtomicReference<>(); + AtomicReference followerRequest = new AtomicReference<>(); + + client = new GrpcOmTransport(conf, ugi, omServiceId); + client.startClient("om0", createNodeChannel("om0", + leaderRequestCount, leaderRequest)); + client.startClient("om1", createNodeChannel("om1", + followerRequestCount, followerRequest)); + client.changeLeaderProxyForTest("om0"); + client.changeFollowerReadInitialProxy("om1"); + + OMRequest request = OMRequest.newBuilder() + .setCmdType(Type.ListVolume) + .setVersion(CURRENT_VERSION) + .setClientId("test") + .build(); + client.submitRequest(request); + + assertEquals(1, leaderRequestCount.get()); + assertEquals(0, followerRequestCount.get()); + assertEquals(ReadConsistencyProto.DEFAULT, + leaderRequest.get().getReadConsistencyHint().getReadConsistency()); + + client.submitRequest(request.toBuilder() + .setReadConsistencyHint(ReadConsistencyHint.newBuilder() + .setReadConsistency(ReadConsistencyProto.LOCAL_LEASE)) + .build()); + + assertEquals(1, followerRequestCount.get()); + assertEquals(ReadConsistencyProto.LOCAL_LEASE, + followerRequest.get().getReadConsistencyHint().getReadConsistency()); + } + @Test public void testFollowerReadFallsBackToLeaderOnNotLeaderException() throws Exception { conf.setBoolean(OzoneConfigKeys.OZONE_CLIENT_FOLLOWER_READ_ENABLED_KEY, true); diff --git a/hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v1/AbstractS3SDKV1Tests.java b/hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v1/AbstractS3SDKV1Tests.java index 42e8140dcb87..f8cce986b2be 100644 --- a/hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v1/AbstractS3SDKV1Tests.java +++ b/hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v1/AbstractS3SDKV1Tests.java @@ -183,6 +183,10 @@ public abstract class AbstractS3SDKV1Tests extends OzoneTestBase implements NonH // server-side limitation private static final int MAX_UPLOADS_LIMIT = 1000; + private static final String READ_CONSISTENCY_HEADER = + "x-ozone-read-consistency"; + private static final String LOCAL_LEASE_LOG_LIMIT_HEADER = + "x-ozone-local-lease-log-limit"; /** * There are still some unsupported S3 operations. @@ -576,6 +580,50 @@ public void testPutObject() { assertEquals("37b51d194a7513e45b56f6524f2d51f2", putObjectResult.getETag()); } + @ParameterizedTest + @ValueSource(strings = {"follower-stale", "follower-linearizable", + "leader-only"}) + public void testGetObjectWithReadConsistencyHeader(String readConsistency) + throws IOException { + final String bucketName = getBucketName(); + final String keyName = getKeyName(); + final String content = "bar"; + s3Client.createBucket(bucketName); + s3Client.putObject(bucketName, keyName, content); + + GetObjectRequest request = new GetObjectRequest(bucketName, keyName); + request.putCustomRequestHeader(READ_CONSISTENCY_HEADER, + readConsistency); + if ("follower-stale".equals(readConsistency)) { + request.putCustomRequestHeader(LOCAL_LEASE_LOG_LIMIT_HEADER, "10"); + } + + try (S3Object object = s3Client.getObject(request); + S3ObjectInputStream objectContent = object.getObjectContent()) { + assertEquals(content, IOUtils.toString(objectContent, + StandardCharsets.UTF_8)); + } + } + + @Test + public void testGetObjectWithInvalidReadConsistencyHeader() { + final String bucketName = getBucketName(); + final String keyName = getKeyName(); + s3Client.createBucket(bucketName); + s3Client.putObject(bucketName, keyName, "bar"); + + GetObjectRequest request = new GetObjectRequest(bucketName, keyName); + request.putCustomRequestHeader(READ_CONSISTENCY_HEADER, + "invalid"); + + AmazonServiceException ase = assertThrows(AmazonServiceException.class, + () -> s3Client.getObject(request)); + + assertEquals(ErrorType.Client, ase.getErrorType()); + assertEquals(400, ase.getStatusCode()); + assertEquals("InvalidArgument", ase.getErrorCode()); + } + @Test public void testPutObjectWithEmptyContentType() { final String bucketName = getBucketName(); @@ -594,6 +642,25 @@ public void testPutObjectWithEmptyContentType() { putObjectResult.getETag()); } + @Test + public void testGetObjectWithMalformedReadConsistencyHeader() { + final String bucketName = getBucketName(); + final String keyName = getKeyName(); + s3Client.createBucket(bucketName); + s3Client.putObject(bucketName, keyName, "bar"); + + GetObjectRequest request = new GetObjectRequest(bucketName, keyName); + request.putCustomRequestHeader(READ_CONSISTENCY_HEADER, + "follower-stale;logLimit=10"); + + AmazonServiceException ase = assertThrows(AmazonServiceException.class, + () -> s3Client.getObject(request)); + + assertEquals(ErrorType.Client, ase.getErrorType()); + assertEquals(400, ase.getStatusCode()); + assertEquals("InvalidArgument", ase.getErrorCode()); + } + @Test public void testPutObjectIfNoneMatch() { final String bucketName = getBucketName(); diff --git a/hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v2/AbstractS3SDKV2Tests.java b/hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v2/AbstractS3SDKV2Tests.java index bfcf5b741bc3..893869a0afc4 100644 --- a/hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v2/AbstractS3SDKV2Tests.java +++ b/hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v2/AbstractS3SDKV2Tests.java @@ -105,6 +105,7 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; +import org.junit.jupiter.params.provider.ValueSource; import software.amazon.awssdk.core.ResponseBytes; import software.amazon.awssdk.core.ResponseInputStream; import software.amazon.awssdk.core.sync.RequestBody; @@ -222,6 +223,11 @@ @TestInstance(TestInstance.Lifecycle.PER_CLASS) public abstract class AbstractS3SDKV2Tests extends OzoneTestBase implements NonHATests.TestCase { + private static final String READ_CONSISTENCY_HEADER = + "x-ozone-read-consistency"; + private static final String LOCAL_LEASE_LOG_LIMIT_HEADER = + "x-ozone-local-lease-log-limit"; + private MiniOzoneCluster cluster; private S3Client s3Client; private S3AsyncClient s3AsyncClient; @@ -300,6 +306,48 @@ public void testPutObject() { assertEquals("\"37b51d194a7513e45b56f6524f2d51f2\"", getObjectResponse.eTag()); } + @ParameterizedTest + @ValueSource(strings = {"follower-stale", "follower-linearizable", + "leader-only"}) + public void testGetObjectWithReadConsistencyHeader(String readConsistency) { + final String bucketName = getBucketName(); + final String keyName = getKeyName(); + final String content = "bar"; + s3Client.createBucket(b -> b.bucket(bucketName)); + s3Client.putObject(b -> b.bucket(bucketName).key(keyName), + RequestBody.fromString(content)); + + ResponseBytes objectBytes = s3Client.getObjectAsBytes( + b -> b.bucket(bucketName) + .key(keyName) + .overrideConfiguration(c -> { + c.putHeader(READ_CONSISTENCY_HEADER, readConsistency); + if ("follower-stale".equals(readConsistency)) { + c.putHeader(LOCAL_LEASE_LOG_LIMIT_HEADER, "10"); + } + })); + + assertEquals(content, objectBytes.asUtf8String()); + } + + @Test + public void testGetObjectWithInvalidReadConsistencyHeader() { + final String bucketName = getBucketName(); + final String keyName = getKeyName(); + s3Client.createBucket(b -> b.bucket(bucketName)); + s3Client.putObject(b -> b.bucket(bucketName).key(keyName), + RequestBody.fromString("bar")); + + S3Exception exception = assertThrows(S3Exception.class, + () -> s3Client.getObjectAsBytes(b -> b.bucket(bucketName) + .key(keyName) + .overrideConfiguration(c -> c.putHeader( + READ_CONSISTENCY_HEADER, "invalid")))); + + assertEquals(400, exception.statusCode()); + assertEquals("InvalidArgument", exception.awsErrorDetails().errorCode()); + } + @Test public void testPutObjectWithEmptyContentType() { final String bucketName = getBucketName(); @@ -408,6 +456,24 @@ public void testBucketTaggingPutGetDelete() { assertEquals("NoSuchTagSet", afterDelete.awsErrorDetails().errorCode()); } + @Test + public void testGetObjectWithMalformedReadConsistencyHeader() { + final String bucketName = getBucketName(); + final String keyName = getKeyName(); + s3Client.createBucket(b -> b.bucket(bucketName)); + s3Client.putObject(b -> b.bucket(bucketName).key(keyName), + RequestBody.fromString("bar")); + + S3Exception exception = assertThrows(S3Exception.class, + () -> s3Client.getObjectAsBytes(b -> b.bucket(bucketName) + .key(keyName) + .overrideConfiguration(c -> c.putHeader( + READ_CONSISTENCY_HEADER, "follower-stale;logLimit=10")))); + + assertEquals(400, exception.statusCode()); + assertEquals("InvalidArgument", exception.awsErrorDetails().errorCode()); + } + @Test public void testPutObjectIfNoneMatch() { final String bucketName = getBucketName(); diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAFollowerReadWithAllRunning.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAFollowerReadWithAllRunning.java index 0e97b2494209..2f67ee7af101 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAFollowerReadWithAllRunning.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAFollowerReadWithAllRunning.java @@ -598,7 +598,7 @@ void testClientWithLocalLeaseEnabled() throws Exception { HadoopRpcOMFollowerReadFailoverProxyProvider followerReadFailoverProxyProvider = OmTestUtil.getFollowerReadFailoverProxyProvider(objectStore); assertNotNull(followerReadFailoverProxyProvider); - assertTrue(followerReadFailoverProxyProvider.isUseFollowerRead()); + assertTrue(followerReadFailoverProxyProvider.isOmServiceSupportsFollowerRead()); String currentOMNodeId = followerReadFailoverProxyProvider.getCurrentProxy().getNodeId(); OzoneManager ozoneManager = getCluster().getOzoneManager(currentOMNodeId); diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAFollowerReadWithStoppedNodes.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAFollowerReadWithStoppedNodes.java index d5792eaecd72..7fd0e7f578a6 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAFollowerReadWithStoppedNodes.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAFollowerReadWithStoppedNodes.java @@ -256,7 +256,7 @@ void testFollowerReadOmProxyProviderFailoverOnConnectionFailure() throws Excepti // Verify that a failover occurred. the new proxy nodeId should be // different from the old proxy nodeId. assertNotEquals(firstProxyNodeId, newProxyNodeId); - assertTrue(followerReadFailoverProxyProvider.isUseFollowerRead()); + assertTrue(followerReadFailoverProxyProvider.isOmServiceSupportsFollowerRead()); } @Test diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAWithAllRunning.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAWithAllRunning.java index 746f99d64958..4e38f095a356 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAWithAllRunning.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHAWithAllRunning.java @@ -1134,7 +1134,7 @@ void testOMFollowerReadWithClusterDisabled() throws Exception { HadoopRpcOMFollowerReadFailoverProxyProvider followerReadFailoverProxyProvider = OmTestUtil.getFollowerReadFailoverProxyProvider(objectStore); assertNotNull(followerReadFailoverProxyProvider); - assertTrue(followerReadFailoverProxyProvider.isUseFollowerRead()); + assertTrue(followerReadFailoverProxyProvider.isOmServiceSupportsFollowerRead()); // Trigger write so that the leader failover proxy provider points to the leader @@ -1158,7 +1158,7 @@ void testOMFollowerReadWithClusterDisabled() throws Exception { // Client follower read is disabled since it detected that the cluster does not // support follower read - assertFalse(followerReadFailoverProxyProvider.isUseFollowerRead()); + assertFalse(followerReadFailoverProxyProvider.isOmServiceSupportsFollowerRead()); OMProxyInfo lastProxy = (OMProxyInfo) followerReadFailoverProxyProvider.getLastProxy(); // The last read will be done on the leader diff --git a/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/OzoneClientProducer.java b/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/OzoneClientProducer.java index 7404019d0035..70a578230aa7 100644 --- a/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/OzoneClientProducer.java +++ b/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/OzoneClientProducer.java @@ -48,7 +48,8 @@ public OzoneClient createClient() { @PreDestroy public void destroy() throws IOException { - LOG.debug("{}: Clearing thread-local auth", this); + LOG.debug("{}: Clearing thread-local S3 request context", this); client.getObjectStore().getClientProxy().clearThreadLocalS3Auth(); + client.getObjectStore().getClientProxy().clearThreadLocalReadConsistency(); } } diff --git a/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/endpoint/EndpointBase.java b/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/endpoint/EndpointBase.java index d237f577bbeb..2d309ca02c87 100644 --- a/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/endpoint/EndpointBase.java +++ b/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/endpoint/EndpointBase.java @@ -41,6 +41,8 @@ import static org.apache.hadoop.ozone.s3.exception.S3ErrorTable.newError; import static org.apache.hadoop.ozone.s3.util.S3Consts.AWS_TAG_PREFIX; import static org.apache.hadoop.ozone.s3.util.S3Consts.CUSTOM_METADATA_HEADER_PREFIX; +import static org.apache.hadoop.ozone.s3.util.S3Consts.LOCAL_LEASE_LOG_LIMIT_HEADER; +import static org.apache.hadoop.ozone.s3.util.S3Consts.READ_CONSISTENCY_HEADER; import static org.apache.hadoop.ozone.s3.util.S3Consts.RESERVED_USER_METADATA_KEY_PREFIX; import static org.apache.hadoop.ozone.s3.util.S3Consts.STORAGE_CLASS_HEADER; import static org.apache.hadoop.ozone.s3.util.S3Consts.STORAGE_CONFIG_HEADER; @@ -55,6 +57,7 @@ import static org.apache.hadoop.ozone.s3.util.S3Utils.urlDecode; import static org.apache.hadoop.ozone.s3.util.S3Utils.validateMultiChunksUpload; import static org.apache.hadoop.ozone.s3.util.S3Utils.validateSignatureHeader; +import static org.apache.hadoop.ozone.s3.util.S3Utils.wrapOS3Exception; import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.ImmutableMap; @@ -71,6 +74,7 @@ import java.util.HashSet; import java.util.Iterator; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Set; import java.util.function.Consumer; @@ -107,6 +111,7 @@ import org.apache.hadoop.ozone.client.protocol.ClientProtocol; import org.apache.hadoop.ozone.om.exceptions.OMException; import org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes; +import org.apache.hadoop.ozone.om.helpers.ReadConsistency; import org.apache.hadoop.ozone.om.protocol.S3Auth; import org.apache.hadoop.ozone.s3.MultiDigestInputStream; import org.apache.hadoop.ozone.s3.RequestIdentifier; @@ -256,6 +261,11 @@ public void initialization() { ClientProtocol clientProtocol = getClient().getObjectStore().getClientProxy(); clientProtocol.setThreadLocalS3Auth(s3Auth); + try { + setReadConsistencyFromHeader(clientProtocol); + } catch (OS3Exception ex) { + throw wrapOS3Exception(ex); + } bufferSize = (int) getOzoneConfiguration().getStorageSize( OZONE_S3G_CLIENT_BUFFER_SIZE_KEY, @@ -282,6 +292,73 @@ protected void init() { // hook method } + private void setReadConsistencyFromHeader(ClientProtocol clientProtocol) { + ReadConsistency readConsistency = parseReadConsistencyHeader(); + Long localLeaseLogLimit = parseLocalLeaseContextHeader( + LOCAL_LEASE_LOG_LIMIT_HEADER); + if (readConsistency == null) { + validateNoLocalLeaseContext(localLeaseLogLimit); + clientProtocol.clearThreadLocalReadConsistency(); + } else if (readConsistency == ReadConsistency.LOCAL_LEASE) { + clientProtocol.setThreadLocalReadConsistency(readConsistency, + localLeaseLogLimit, null); + } else { + validateNoLocalLeaseContext(localLeaseLogLimit); + clientProtocol.setThreadLocalReadConsistency(readConsistency); + } + } + + private ReadConsistency parseReadConsistencyHeader() { + String header = getHeaders().getHeaderString(READ_CONSISTENCY_HEADER); + if (StringUtils.isBlank(header)) { + return null; + } + switch (header.trim().toLowerCase(Locale.ROOT)) { + case "follower-stale": + return ReadConsistency.LOCAL_LEASE; + case "leader-only": + return ReadConsistency.LINEARIZABLE_LEADER_ONLY; + case "follower-linearizable": + return ReadConsistency.LINEARIZABLE_ALLOW_FOLLOWER; + default: + OS3Exception ex = newError(INVALID_ARGUMENT, READ_CONSISTENCY_HEADER); + ex.setErrorMessage("Unsupported read consistency: " + header); + throw ex; + } + } + + private Long parseLocalLeaseContextHeader(String headerName) { + String header = getHeaders().getHeaderString(headerName); + if (StringUtils.isBlank(header)) { + return null; + } + try { + long value = Long.parseLong(header.trim()); + if (value < -1) { + throw invalidLocalLeaseContext(headerName, header); + } + return value; + } catch (NumberFormatException e) { + throw invalidLocalLeaseContext(headerName, header); + } + } + + private void validateNoLocalLeaseContext(Long localLeaseLogLimit) { + if (localLeaseLogLimit != null) { + OS3Exception ex = newError(INVALID_ARGUMENT, READ_CONSISTENCY_HEADER); + ex.setErrorMessage("Local lease context requires read consistency: " + + "follower-stale"); + throw ex; + } + } + + private OS3Exception invalidLocalLeaseContext(String headerName, + String header) { + OS3Exception ex = newError(INVALID_ARGUMENT, headerName); + ex.setErrorMessage("Invalid local lease context: " + header); + return ex; + } + /** * Sets the IAM S3 action on thread-local {@link S3Auth} for fine-grained STS authorization. * Called when the handler resolves the {@link S3GAction}. diff --git a/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/util/S3Consts.java b/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/util/S3Consts.java index f75653ad098b..bb69cab8242d 100644 --- a/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/util/S3Consts.java +++ b/hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/util/S3Consts.java @@ -82,6 +82,10 @@ public final class S3Consts { public static final String RESERVED_USER_METADATA_KEY_PREFIX = "ozone-s3-internal-"; public static final String CUSTOM_METADATA_COPY_DIRECTIVE_HEADER = "x-amz-metadata-directive"; public static final String STORAGE_CONFIG_HEADER = "storage-config"; + public static final String READ_CONSISTENCY_HEADER = + "x-ozone-read-consistency"; + public static final String LOCAL_LEASE_LOG_LIMIT_HEADER = + "x-ozone-local-lease-log-limit"; public static final String DECODED_CONTENT_LENGTH_HEADER = "x-amz-decoded-content-length"; diff --git a/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/client/ClientProtocolStub.java b/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/client/ClientProtocolStub.java index 40e384d2c547..b4c7d9e056a8 100644 --- a/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/client/ClientProtocolStub.java +++ b/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/client/ClientProtocolStub.java @@ -53,6 +53,7 @@ import org.apache.hadoop.ozone.om.helpers.OmVolumeArgs; import org.apache.hadoop.ozone.om.helpers.OzoneFileStatus; import org.apache.hadoop.ozone.om.helpers.OzoneFileStatusLight; +import org.apache.hadoop.ozone.om.helpers.ReadConsistency; import org.apache.hadoop.ozone.om.helpers.S3SecretValue; import org.apache.hadoop.ozone.om.helpers.S3VolumeContext; import org.apache.hadoop.ozone.om.helpers.TenantStateList; @@ -77,6 +78,9 @@ public class ClientProtocolStub implements ClientProtocol { private static final String STUB_KERBEROS_ID = "stub_kerberos_id"; private static final String STUB_SECRET = "stub_secret"; private final ObjectStoreStub objectStoreStub; + private ReadConsistency readConsistency; + private Long localLeaseLogLimit; + private Long localLeaseTimeMs; public ClientProtocolStub(ObjectStoreStub objectStoreStub) { this.objectStoreStub = objectStoreStub; @@ -756,16 +760,52 @@ public void setThreadLocalS3Auth(S3Auth s3Auth) { } + @Override + public void setThreadLocalReadConsistency( + ReadConsistency newReadConsistency) { + this.readConsistency = newReadConsistency; + this.localLeaseLogLimit = null; + this.localLeaseTimeMs = null; + } + + @Override + public void setThreadLocalReadConsistency(ReadConsistency newReadConsistency, + Long newLocalLeaseLogLimit, Long newLocalLeaseTimeMs) { + this.readConsistency = newReadConsistency; + this.localLeaseLogLimit = newLocalLeaseLogLimit; + this.localLeaseTimeMs = newLocalLeaseTimeMs; + } + @Override public S3Auth getThreadLocalS3Auth() { return null; } + @Override + public ReadConsistency getThreadLocalReadConsistency() { + return readConsistency; + } + + public Long getThreadLocalLocalLeaseLogLimit() { + return localLeaseLogLimit; + } + + public Long getThreadLocalLocalLeaseTimeMs() { + return localLeaseTimeMs; + } + @Override public void clearThreadLocalS3Auth() { } + @Override + public void clearThreadLocalReadConsistency() { + readConsistency = null; + localLeaseLogLimit = null; + localLeaseTimeMs = null; + } + @Override public boolean setBucketOwner(String volumeName, String bucketName, String owner) throws IOException { diff --git a/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/TestOzoneClientProducer.java b/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/TestOzoneClientProducer.java new file mode 100644 index 000000000000..810a0832db9f --- /dev/null +++ b/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/TestOzoneClientProducer.java @@ -0,0 +1,58 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.ozone.s3; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.lang.reflect.Field; +import org.apache.hadoop.ozone.client.ObjectStore; +import org.apache.hadoop.ozone.client.OzoneClient; +import org.apache.hadoop.ozone.client.protocol.ClientProtocol; +import org.junit.jupiter.api.Test; + +/** + * Tests {@link OzoneClientProducer}. + */ +public class TestOzoneClientProducer { + + @Test + public void destroyClearsThreadLocalS3RequestContext() throws Exception { + ClientProtocol clientProtocol = mock(ClientProtocol.class); + ObjectStore objectStore = mock(ObjectStore.class); + when(objectStore.getClientProxy()).thenReturn(clientProtocol); + OzoneClient client = mock(OzoneClient.class); + when(client.getObjectStore()).thenReturn(objectStore); + + OzoneClientProducer producer = new OzoneClientProducer(); + setClient(producer, client); + + producer.destroy(); + + verify(clientProtocol).clearThreadLocalS3Auth(); + verify(clientProtocol).clearThreadLocalReadConsistency(); + } + + private static void setClient(OzoneClientProducer producer, OzoneClient client) + throws Exception { + Field field = OzoneClientProducer.class.getDeclaredField("client"); + field.setAccessible(true); + field.set(producer, client); + } +} diff --git a/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/endpoint/TestEndpointBase.java b/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/endpoint/TestEndpointBase.java index 9865345a9162..0450da350678 100644 --- a/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/endpoint/TestEndpointBase.java +++ b/hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/endpoint/TestEndpointBase.java @@ -20,6 +20,8 @@ import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes; import static org.apache.hadoop.ozone.s3.exception.S3ErrorTable.INVALID_ARGUMENT; import static org.apache.hadoop.ozone.s3.util.S3Consts.CUSTOM_METADATA_HEADER_PREFIX; +import static org.apache.hadoop.ozone.s3.util.S3Consts.LOCAL_LEASE_LOG_LIMIT_HEADER; +import static org.apache.hadoop.ozone.s3.util.S3Consts.READ_CONSISTENCY_HEADER; import static org.apache.hadoop.ozone.s3.util.S3Consts.RESERVED_USER_METADATA_KEY_PREFIX; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -34,11 +36,19 @@ import java.util.Locale; import java.util.Map; import java.util.stream.Stream; +import javax.ws.rs.WebApplicationException; +import javax.ws.rs.core.HttpHeaders; import javax.ws.rs.core.MultivaluedHashMap; import javax.ws.rs.core.MultivaluedMap; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.ozone.OzoneConsts; +import org.apache.hadoop.ozone.client.ClientProtocolStub; +import org.apache.hadoop.ozone.client.ObjectStoreStub; +import org.apache.hadoop.ozone.client.OzoneClientStub; import org.apache.hadoop.ozone.client.OzoneVolume; +import org.apache.hadoop.ozone.client.protocol.ClientProtocol; import org.apache.hadoop.ozone.om.exceptions.OMException; +import org.apache.hadoop.ozone.om.helpers.ReadConsistency; import org.apache.hadoop.ozone.s3.exception.OS3Exception; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; @@ -203,10 +213,154 @@ public void testRejectReservedInternalMetadataKeyPrefix(String metadataKey) { assertThat(e.getErrorMessage()).contains(RESERVED_USER_METADATA_KEY_PREFIX); } + @Test + public void testReadConsistencyHeaderLocalLease() { + HttpHeaders headers = mock(HttpHeaders.class); + when(headers.getHeaderString(READ_CONSISTENCY_HEADER)) + .thenReturn("follower-stale"); + + RootEndpoint endpoint = newRootEndpoint(headers); + + ClientProtocol clientProtocol = + endpoint.getClient().getObjectStore().getClientProxy(); + assertEquals(ReadConsistency.LOCAL_LEASE, + clientProtocol.getThreadLocalReadConsistency()); + } + + @Test + public void testReadConsistencyHeaderLocalLeaseContext() { + HttpHeaders headers = mock(HttpHeaders.class); + when(headers.getHeaderString(READ_CONSISTENCY_HEADER)) + .thenReturn("follower-stale"); + when(headers.getHeaderString(LOCAL_LEASE_LOG_LIMIT_HEADER)) + .thenReturn("10"); + + RootEndpoint endpoint = newRootEndpoint(headers); + + ClientProtocolStub clientProtocol = (ClientProtocolStub) endpoint + .getClient().getObjectStore().getClientProxy(); + assertEquals(ReadConsistency.LOCAL_LEASE, + clientProtocol.getThreadLocalReadConsistency()); + assertEquals(10L, clientProtocol.getThreadLocalLocalLeaseLogLimit()); + assertThat(clientProtocol.getThreadLocalLocalLeaseTimeMs()).isNull(); + } + + @Test + public void testReadConsistencyHeaderLocalLeaseContextAllowsMinusOne() { + HttpHeaders headers = mock(HttpHeaders.class); + when(headers.getHeaderString(READ_CONSISTENCY_HEADER)) + .thenReturn("follower-stale"); + when(headers.getHeaderString(LOCAL_LEASE_LOG_LIMIT_HEADER)) + .thenReturn("-1"); + + RootEndpoint endpoint = newRootEndpoint(headers); + + ClientProtocolStub clientProtocol = (ClientProtocolStub) endpoint + .getClient().getObjectStore().getClientProxy(); + assertEquals(-1L, clientProtocol.getThreadLocalLocalLeaseLogLimit()); + assertThat(clientProtocol.getThreadLocalLocalLeaseTimeMs()).isNull(); + } + + @Test + public void testReadConsistencyHeaderLinearizableFollower() { + HttpHeaders headers = mock(HttpHeaders.class); + when(headers.getHeaderString(READ_CONSISTENCY_HEADER)) + .thenReturn("follower-linearizable"); + + RootEndpoint endpoint = newRootEndpoint(headers); + + ClientProtocol clientProtocol = + endpoint.getClient().getObjectStore().getClientProxy(); + assertEquals(ReadConsistency.LINEARIZABLE_ALLOW_FOLLOWER, + clientProtocol.getThreadLocalReadConsistency()); + } + + @Test + public void testReadConsistencyHeaderLeaderOnly() { + HttpHeaders headers = mock(HttpHeaders.class); + when(headers.getHeaderString(READ_CONSISTENCY_HEADER)) + .thenReturn("leader-only"); + + RootEndpoint endpoint = newRootEndpoint(headers); + + ClientProtocol clientProtocol = + endpoint.getClient().getObjectStore().getClientProxy(); + assertEquals(ReadConsistency.LINEARIZABLE_LEADER_ONLY, + clientProtocol.getThreadLocalReadConsistency()); + } + + @Test + public void testReadConsistencyHeaderUnsetByDefault() { + RootEndpoint endpoint = newRootEndpoint(null); + + ClientProtocol clientProtocol = + endpoint.getClient().getObjectStore().getClientProxy(); + assertThat(clientProtocol.getThreadLocalReadConsistency()).isNull(); + } + + @Test + public void testInvalidReadConsistencyHeader() { + HttpHeaders headers = mock(HttpHeaders.class); + when(headers.getHeaderString(READ_CONSISTENCY_HEADER)) + .thenReturn("eventual"); + + WebApplicationException e = assertThrows(WebApplicationException.class, + () -> EndpointBuilder.newRootEndpointBuilder() + .setHeaders(headers) + .build()); + assertThat(e.getResponse().getStatus()).isEqualTo(400); + assertThat(e.getResponse().getEntity().toString()).contains("InvalidArgument"); + } + + @Test + public void testInvalidLocalLeaseContextWithoutLocalLease() { + HttpHeaders headers = mock(HttpHeaders.class); + when(headers.getHeaderString(READ_CONSISTENCY_HEADER)) + .thenReturn("follower-linearizable"); + when(headers.getHeaderString(LOCAL_LEASE_LOG_LIMIT_HEADER)) + .thenReturn("10"); + + WebApplicationException e = assertThrows(WebApplicationException.class, + () -> EndpointBuilder.newRootEndpointBuilder() + .setHeaders(headers) + .build()); + assertThat(e.getResponse().getStatus()).isEqualTo(400); + assertThat(e.getResponse().getEntity().toString()).contains("InvalidArgument"); + } + + @Test + public void testInvalidLocalLeaseContextValue() { + HttpHeaders headers = mock(HttpHeaders.class); + when(headers.getHeaderString(READ_CONSISTENCY_HEADER)) + .thenReturn("follower-stale"); + when(headers.getHeaderString(LOCAL_LEASE_LOG_LIMIT_HEADER)) + .thenReturn("abc"); + + WebApplicationException e = assertThrows(WebApplicationException.class, + () -> EndpointBuilder.newRootEndpointBuilder() + .setHeaders(headers) + .build()); + assertThat(e.getResponse().getStatus()).isEqualTo(400); + assertThat(e.getResponse().getEntity().toString()).contains("InvalidArgument"); + } + private static Stream reservedInternalMetadataKeyPrefixCases() { return Stream.of( RESERVED_USER_METADATA_KEY_PREFIX + "cache-control", RESERVED_USER_METADATA_KEY_PREFIX.toUpperCase(Locale.ROOT) + "cache-control"); } + private static RootEndpoint newRootEndpoint(HttpHeaders headers) { + ClientProtocol clientProtocol = new ClientProtocolStub(null); + ObjectStoreStub objectStore = + new ObjectStoreStub(new OzoneConfiguration(), clientProtocol); + EndpointBuilder builder = EndpointBuilder + .newRootEndpointBuilder() + .setClient(new OzoneClientStub(objectStore)); + if (headers != null) { + builder.setHeaders(headers); + } + return builder.build(); + } + }