Package com.linkedin.davinci.kafka.consumer

Class KafkaConsumerServiceDelegator

java.lang.Object

com.linkedin.venice.service.AbstractVeniceService

com.linkedin.davinci.kafka.consumer.AbstractKafkaConsumerService

com.linkedin.davinci.kafka.consumer.KafkaConsumerServiceDelegator

All Implemented Interfaces:

java.io.Closeable, java.lang.AutoCloseable

public class KafkaConsumerServiceDelegator
extends AbstractKafkaConsumerService

This delegator impl is used to distribute different partition requests into different consumer service. When {#link ConfigKeys#SERVER_DEDICATED_CONSUMER_POOL_FOR_AA_WC_LEADER_ENABLED} is off, this class will always return the default consumer service. When the option is on, it will return the dedicated consumer service when the topic partition belongs to a Real-time topic and the corresponding store has active/active or write compute enabled. The reason to use dedicated consumer pool for leader replicas of active/active or write compute stores is that handling the writes before putting into the drainer queue is too expensive comparing to others.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
class	KafkaConsumerServiceDelegator.AAOrWCLeaderConsumerPoolStrategy
class	KafkaConsumerServiceDelegator.ConsumerPoolStrategy
static class	KafkaConsumerServiceDelegator.ConsumerPoolStrategyType
class	KafkaConsumerServiceDelegator.CurrentVersionConsumerPoolStrategy
class	KafkaConsumerServiceDelegator.DefaultConsumerPoolStrategy

Nested classes/interfaces inherited from class com.linkedin.venice.service.AbstractVeniceService

AbstractVeniceService.ServiceState

Field Summary

Fields inherited from class com.linkedin.venice.service.AbstractVeniceService

logger, serviceState

Constructor Summary

Constructors

Constructor	Description
KafkaConsumerServiceDelegator(VeniceServerConfig serverConfig, com.linkedin.davinci.kafka.consumer.KafkaConsumerServiceDelegator.KafkaConsumerServiceBuilder consumerServiceConstructor, java.util.function.Function<java.lang.String,java.lang.Boolean> isAAWCStoreFunc)

Method Summary

Modifier and Type	Method	Description
com.linkedin.davinci.kafka.consumer.SharedKafkaConsumer	assignConsumerFor(PubSubTopic versionTopic, PubSubTopicPartition pubSubTopicPartition)
void	batchUnsubscribe(PubSubTopic versionTopic, java.util.Set<PubSubTopicPartition> topicPartitionsToUnSub)
com.linkedin.davinci.kafka.consumer.SharedKafkaConsumer	getConsumerAssignedToVersionTopicPartition(PubSubTopic versionTopic, PubSubTopicPartition topicPartition)	There might be no consumer service found for the version topic and topic partition, which means the topic partition is not subscribed.
java.util.Map<PubSubTopicPartition,TopicPartitionIngestionInfo>	getIngestionInfoFromConsumer(PubSubTopic versionTopic, PubSubTopicPartition pubSubTopicPartition)
long	getLatestOffsetBasedOnMetrics(PubSubTopic versionTopic, PubSubTopicPartition pubSubTopicPartition)
long	getMaxElapsedTimeMSSinceLastPollInConsumerPool()
long	getOffsetLagBasedOnMetrics(PubSubTopic versionTopic, PubSubTopicPartition pubSubTopicPartition)
boolean	hasAnySubscriptionFor(PubSubTopic versionTopic)
void	startConsumptionIntoDataReceiver(PartitionReplicaIngestionContext partitionReplicaIngestionContext, long lastReadOffset, ConsumedDataReceiver<java.util.List<PubSubMessage<KafkaKey,KafkaMessageEnvelope,java.lang.Long>>> consumedDataReceiver)
boolean	startInner()
void	stopInner()
void	unSubscribe(PubSubTopic versionTopic, PubSubTopicPartition pubSubTopicPartition, long timeoutMs)
void	unsubscribeAll(PubSubTopic versionTopic)

Methods inherited from class com.linkedin.davinci.kafka.consumer.AbstractKafkaConsumerService

unSubscribe

Methods inherited from class com.linkedin.venice.service.AbstractVeniceService

close, getName, isRunning, start, stop

Methods inherited from class java.lang.Object

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

Constructor Detail

KafkaConsumerServiceDelegator

public KafkaConsumerServiceDelegator(VeniceServerConfig serverConfig,
                                     com.linkedin.davinci.kafka.consumer.KafkaConsumerServiceDelegator.KafkaConsumerServiceBuilder consumerServiceConstructor,
                                     java.util.function.Function<java.lang.String,java.lang.Boolean> isAAWCStoreFunc)

Method Detail

getConsumerAssignedToVersionTopicPartition

public com.linkedin.davinci.kafka.consumer.SharedKafkaConsumer getConsumerAssignedToVersionTopicPartition(PubSubTopic versionTopic,
                                                                                                          PubSubTopicPartition topicPartition)

There might be no consumer service found for the version topic and topic partition, which means the topic partition is not subscribed. In this case, we should return null. Caller should check the return value and handle it properly.

Specified by:

getConsumerAssignedToVersionTopicPartition in class AbstractKafkaConsumerService

assignConsumerFor

public com.linkedin.davinci.kafka.consumer.SharedKafkaConsumer assignConsumerFor(PubSubTopic versionTopic,
                                                                                 PubSubTopicPartition pubSubTopicPartition)

Specified by:

assignConsumerFor in class AbstractKafkaConsumerService

unsubscribeAll

public void unsubscribeAll(PubSubTopic versionTopic)

Specified by:

unsubscribeAll in class AbstractKafkaConsumerService

unSubscribe

public void unSubscribe(PubSubTopic versionTopic,
                        PubSubTopicPartition pubSubTopicPartition,
                        long timeoutMs)

Specified by:

unSubscribe in class AbstractKafkaConsumerService

batchUnsubscribe

public void batchUnsubscribe(PubSubTopic versionTopic,
                             java.util.Set<PubSubTopicPartition> topicPartitionsToUnSub)

Specified by:

batchUnsubscribe in class AbstractKafkaConsumerService

hasAnySubscriptionFor

public boolean hasAnySubscriptionFor(PubSubTopic versionTopic)

Specified by:

hasAnySubscriptionFor in class AbstractKafkaConsumerService

getMaxElapsedTimeMSSinceLastPollInConsumerPool

public long getMaxElapsedTimeMSSinceLastPollInConsumerPool()

Specified by:

getMaxElapsedTimeMSSinceLastPollInConsumerPool in class AbstractKafkaConsumerService

startConsumptionIntoDataReceiver

public void startConsumptionIntoDataReceiver(PartitionReplicaIngestionContext partitionReplicaIngestionContext,
                                             long lastReadOffset,
                                             ConsumedDataReceiver<java.util.List<PubSubMessage<KafkaKey,KafkaMessageEnvelope,java.lang.Long>>> consumedDataReceiver)

Specified by:

startConsumptionIntoDataReceiver in class AbstractKafkaConsumerService

getOffsetLagBasedOnMetrics

public long getOffsetLagBasedOnMetrics(PubSubTopic versionTopic,
                                       PubSubTopicPartition pubSubTopicPartition)

Specified by:

getOffsetLagBasedOnMetrics in class AbstractKafkaConsumerService

getLatestOffsetBasedOnMetrics

public long getLatestOffsetBasedOnMetrics(PubSubTopic versionTopic,
                                          PubSubTopicPartition pubSubTopicPartition)

Specified by:

getLatestOffsetBasedOnMetrics in class AbstractKafkaConsumerService

getIngestionInfoFromConsumer

public java.util.Map<PubSubTopicPartition,TopicPartitionIngestionInfo> getIngestionInfoFromConsumer(PubSubTopic versionTopic,
                                                                                                          PubSubTopicPartition pubSubTopicPartition)

Specified by:

getIngestionInfoFromConsumer in class AbstractKafkaConsumerService

startInner

public boolean startInner()
                   throws java.lang.Exception

Specified by:

startInner in class AbstractVeniceService

Returns:

true if the service is completely started, false if it is still starting asynchronously (in this case, it is the implementer's responsibility to set AbstractVeniceService.serviceState to AbstractVeniceService.ServiceState.STARTED upon completion of the async work).

Throws:

java.lang.Exception

stopInner

public void stopInner()
               throws java.lang.Exception

Specified by:

stopInner in class AbstractVeniceService

Throws:

java.lang.Exception