Package com.linkedin.venice.listener

Class ReadQuotaEnforcementHandler

java.lang.Object

io.netty.channel.ChannelHandlerAdapter

io.netty.channel.ChannelInboundHandlerAdapter

io.netty.channel.SimpleChannelInboundHandler<RouterRequest>

com.linkedin.venice.listener.ReadQuotaEnforcementHandler

All Implemented Interfaces:

RoutingDataRepository.RoutingDataChangedListener, StoreDataChangedListener, io.netty.channel.ChannelHandler, io.netty.channel.ChannelInboundHandler

@Sharable
public class ReadQuotaEnforcementHandler
extends io.netty.channel.SimpleChannelInboundHandler<RouterRequest>
implements RoutingDataRepository.RoutingDataChangedListener, StoreDataChangedListener

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static class	ReadQuotaEnforcementHandler.QuotaEnforcementResult

Nested classes/interfaces inherited from interface io.netty.channel.ChannelHandler

io.netty.channel.ChannelHandler.Sharable

Field Summary

Fields

Modifier and Type	Field	Description
static java.lang.String	INVALID_REQUEST_RESOURCE_MSG
static java.lang.String	SERVER_OVER_CAPACITY_MSG

Constructor Summary

Constructors

Constructor	Description
ReadQuotaEnforcementHandler(VeniceServerConfig serverConfig, ReadOnlyStoreRepository storeRepository, java.util.concurrent.CompletableFuture<HelixCustomizedViewOfflinePushRepository> customizedViewRepository, java.lang.String nodeId, AggServerQuotaUsageStats stats, io.tehuti.metrics.MetricsRepository metricsRepository)
ReadQuotaEnforcementHandler(VeniceServerConfig serverConfig, ReadOnlyStoreRepository storeRepository, java.util.concurrent.CompletableFuture<HelixCustomizedViewOfflinePushRepository> customizedViewRepository, java.lang.String nodeId, AggServerQuotaUsageStats stats, io.tehuti.metrics.MetricsRepository metricsRepository, java.time.Clock clock)

Method Summary

Modifier and Type	Method	Description
void	channelRead0(io.netty.channel.ChannelHandlerContext ctx, RouterRequest request)
ReadQuotaEnforcementHandler.QuotaEnforcementResult	enforceQuota(RouterRequest request)	Enforce quota for a given request.
protected java.util.Set<java.lang.String>	getActiveStoreVersions()	Helper methods for unit testing
protected static double	getNodeResponsibilityForQuota(PartitionAssignment partitionAssignment, java.lang.String nodeId)	Recalculates the amount of quota that this node should serve given the partition assignment.
static int	getRcu(RouterRequest request)	Initially, this is the key count.
AggServerQuotaUsageStats	getStats()
ReadOnlyStoreRepository	getStoreRepository()
void	handleStoreChanged(Store store)	This is where we add new TokenBucket for new store version and remove irrelevant ones.
void	handleStoreCreated(Store store)	Do NOT try to acquire the lock of store repository again in the implementation, otherwise a dead lock issue will happen.
void	handleStoreDeleted(java.lang.String storeName)
void	init()	Initialize token buckets for all resources in the customized view repository.
boolean	isInitialized()	We only ever expect the initialized value to flip from false to true, but that initialization might happen in a different thread.
void	onCustomizedViewAdded(PartitionAssignment partitionAssignment)
void	onCustomizedViewChange(PartitionAssignment partitionAssignment)
void	onExternalViewChange(PartitionAssignment partitionAssignment)	Handle routing data changed event.
void	onPartitionStatusChange(java.lang.String topic, ReadOnlyPartitionStatus partitionStatus)
void	onRoutingDataDeleted(java.lang.String kafkaTopic)

Methods inherited from class io.netty.channel.SimpleChannelInboundHandler

acceptInboundMessage, channelRead

Methods inherited from class io.netty.channel.ChannelInboundHandlerAdapter

channelActive, channelInactive, channelReadComplete, channelRegistered, channelUnregistered, channelWritabilityChanged, exceptionCaught, userEventTriggered

Methods inherited from class io.netty.channel.ChannelHandlerAdapter

ensureNotSharable, handlerAdded, handlerRemoved, isSharable

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface io.netty.channel.ChannelHandler

handlerAdded, handlerRemoved

Methods inherited from interface com.linkedin.venice.meta.StoreDataChangedListener

handleStoreDeleted

Field Detail

SERVER_OVER_CAPACITY_MSG

public static final java.lang.String SERVER_OVER_CAPACITY_MSG

See Also:

Constant Field Values

INVALID_REQUEST_RESOURCE_MSG

public static final java.lang.String INVALID_REQUEST_RESOURCE_MSG

See Also:

Constant Field Values

Constructor Detail

ReadQuotaEnforcementHandler

public ReadQuotaEnforcementHandler(VeniceServerConfig serverConfig,
                                   ReadOnlyStoreRepository storeRepository,
                                   java.util.concurrent.CompletableFuture<HelixCustomizedViewOfflinePushRepository> customizedViewRepository,
                                   java.lang.String nodeId,
                                   AggServerQuotaUsageStats stats,
                                   io.tehuti.metrics.MetricsRepository metricsRepository)

ReadQuotaEnforcementHandler

public ReadQuotaEnforcementHandler(VeniceServerConfig serverConfig,
                                   ReadOnlyStoreRepository storeRepository,
                                   java.util.concurrent.CompletableFuture<HelixCustomizedViewOfflinePushRepository> customizedViewRepository,
                                   java.lang.String nodeId,
                                   AggServerQuotaUsageStats stats,
                                   io.tehuti.metrics.MetricsRepository metricsRepository,
                                   java.time.Clock clock)

Method Detail

getRcu

public static int getRcu(RouterRequest request)

Initially, this is the key count. As we develop a more accurate capacity model, this method can be refined.

Parameters:

request -

Returns:

getNodeResponsibilityForQuota

protected static double getNodeResponsibilityForQuota(PartitionAssignment partitionAssignment,
                                                      java.lang.String nodeId)

Recalculates the amount of quota that this node should serve given the partition assignment. Assumes each partition gets an even portion of quota, and for each partition divides the quota by the readyToServe instances.

Parameters:

partitionAssignment -

nodeId -

Returns:

init

public final void init()

Initialize token buckets for all resources in the customized view repository.

isInitialized

public boolean isInitialized()

We only ever expect the initialized value to flip from false to true, but that initialization might happen in a different thread. By only checking the volatile variable if it is false, we guarantee that we see the change as early as possible and also allow the thread to cache the value once it goes true.

Returns:

enforceQuota

public ReadQuotaEnforcementHandler.QuotaEnforcementResult enforceQuota(RouterRequest request)

Enforce quota for a given request. This is common to both HTTP and GRPC handlers. Respective handlers will take actions such as retaining the request and passing it to the next handler, or sending an error response.

Parameters:

request - RouterRequest

Returns:

QuotaEnforcementResult

channelRead0

public void channelRead0(io.netty.channel.ChannelHandlerContext ctx,
                         RouterRequest request)

Specified by:

channelRead0 in class io.netty.channel.SimpleChannelInboundHandler<RouterRequest>

onExternalViewChange

public void onExternalViewChange(PartitionAssignment partitionAssignment)

Description copied from interface: RoutingDataRepository.RoutingDataChangedListener

Handle routing data changed event.

Specified by:

onExternalViewChange in interface RoutingDataRepository.RoutingDataChangedListener

Parameters:

partitionAssignment - Newest partitions assignments information including resource name and all of instances assigned to this resource. If the number of partition is 0, it means the kafka topic is deleted.

onCustomizedViewChange

public void onCustomizedViewChange(PartitionAssignment partitionAssignment)

Specified by:

onCustomizedViewChange in interface RoutingDataRepository.RoutingDataChangedListener

onCustomizedViewAdded

public void onCustomizedViewAdded(PartitionAssignment partitionAssignment)

Specified by:

onCustomizedViewAdded in interface RoutingDataRepository.RoutingDataChangedListener

onPartitionStatusChange

public void onPartitionStatusChange(java.lang.String topic,
                                    ReadOnlyPartitionStatus partitionStatus)

Specified by:

onPartitionStatusChange in interface RoutingDataRepository.RoutingDataChangedListener

onRoutingDataDeleted

public void onRoutingDataDeleted(java.lang.String kafkaTopic)

Specified by:

onRoutingDataDeleted in interface RoutingDataRepository.RoutingDataChangedListener

handleStoreCreated

public void handleStoreCreated(Store store)

Description copied from interface: StoreDataChangedListener

Do NOT try to acquire the lock of store repository again in the implementation, otherwise a dead lock issue will happen.

Specified by:

handleStoreCreated in interface StoreDataChangedListener

handleStoreDeleted

public void handleStoreDeleted(java.lang.String storeName)

Specified by:

handleStoreDeleted in interface StoreDataChangedListener

handleStoreChanged

public void handleStoreChanged(Store store)

This is where we add new TokenBucket for new store version and remove irrelevant ones. We should keep the same number of token buckets as the number of active versions. This is because readers like FC might be lagging behind or vice versa. This way quota will still be enforced properly during the version swap or transition period.

Specified by:

handleStoreChanged in interface StoreDataChangedListener

getActiveStoreVersions

protected java.util.Set<java.lang.String> getActiveStoreVersions()

Helper methods for unit testing

getStoreRepository

public ReadOnlyStoreRepository getStoreRepository()

getStats

public AggServerQuotaUsageStats getStats()