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:

Closeable, 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.ConsumerPoolStrategy
static enum	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, Function<String,Boolean> isAAWCStoreFunc)

Method Summary

Modifier and Type	Method	Description
com.linkedin.davinci.kafka.consumer.SharedKafkaConsumer	assignConsumerFor(PubSubTopic versionTopic, PubSubTopicPartition pubSubTopicPartition)
void	batchUnsubscribe(PubSubTopic versionTopic, 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.
Map<PubSubTopicPartition,TopicPartitionIngestionInfo>	getIngestionInfoFor(PubSubTopic versionTopic, PubSubTopicPartition pubSubTopicPartition, boolean respectRedundantLoggingFilter)	This is for providing ingestion related information for a specific topic partition from the implementation of this class.
long	getLatestOffsetBasedOnMetrics(PubSubTopic versionTopic, PubSubTopicPartition pubSubTopicPartition)
long	getMaxElapsedTimeMSSinceLastPollInConsumerPool()
Map<PubSubTopicPartition,Long>	getStaleTopicPartitions(long thresholdTimestamp)
boolean	hasAnySubscriptionFor(PubSubTopic versionTopic)
void	startConsumptionIntoDataReceiver(PartitionReplicaIngestionContext partitionReplicaIngestionContext, PubSubPosition lastReadPosition, ConsumedDataReceiver<List<DefaultPubSubMessage>> 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 Details

KafkaConsumerServiceDelegator

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

Method Details

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,
 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,
 PubSubPosition lastReadPosition,
 ConsumedDataReceiver<List<DefaultPubSubMessage>> consumedDataReceiver)

Specified by:

startConsumptionIntoDataReceiver in class AbstractKafkaConsumerService

getLatestOffsetBasedOnMetrics

public long getLatestOffsetBasedOnMetrics(PubSubTopic versionTopic,
 PubSubTopicPartition pubSubTopicPartition)

Specified by:

getLatestOffsetBasedOnMetrics in class AbstractKafkaConsumerService

getIngestionInfoFor

public Map<PubSubTopicPartition,TopicPartitionIngestionInfo> getIngestionInfoFor(PubSubTopic versionTopic,
 PubSubTopicPartition pubSubTopicPartition,
 boolean respectRedundantLoggingFilter)

Description copied from class: AbstractKafkaConsumerService

This is for providing ingestion related information for a specific topic partition from the implementation of this class.

Specified by:

getIngestionInfoFor in class AbstractKafkaConsumerService

respectRedundantLoggingFilter - here is to guide if we need to prepare the info map, set to true when calling from heartbeat monitoring to enable rate-limiting; set to false for admin commands or tests where all info is needed.

getStaleTopicPartitions

public Map<PubSubTopicPartition,Long> getStaleTopicPartitions(long thresholdTimestamp)

Specified by:

getStaleTopicPartitions in class AbstractKafkaConsumerService

startInner

public boolean startInner()
                   throws 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:

Exception

stopInner

public void stopInner()
               throws Exception

Specified by:

stopInner in class AbstractVeniceService

Throws:

Exception