Package com.linkedin.davinci.kafka.consumer

Class AggKafkaConsumerService

java.lang.Object

com.linkedin.venice.service.AbstractVeniceService

com.linkedin.davinci.kafka.consumer.AggKafkaConsumerService

All Implemented Interfaces:

Closeable, AutoCloseable

public class AggKafkaConsumerService
extends AbstractVeniceService

AggKafkaConsumerService supports Kafka consumer pool for multiple Kafka clusters from different data centers; for each Kafka bootstrap server url, AggKafkaConsumerService will create one KafkaConsumerService.

Nested Class Summary

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

AggKafkaConsumerService(PubSubConsumerAdapterFactory consumerFactory, TopicManagerContext.PubSubPropertiesSupplier pubSubPropertiesSupplier, VeniceServerConfig serverConfig, IngestionThrottler ingestionThrottler, KafkaClusterBasedRecordThrottler kafkaClusterBasedRecordThrottler, io.tehuti.metrics.MetricsRepository metricsRepository, StaleTopicChecker staleTopicChecker, PubSubMessageDeserializer pubSubDeserializer, Consumer<String> killIngestionTaskRunnable, Function<String,Boolean> isAAOrWCEnabledFunc, ReadOnlyStoreRepository metadataRepository)

Method Summary

Modifier and Type	Method	Description
AbstractKafkaConsumerService	createKafkaConsumerService(Properties consumerProperties)	Create a new KafkaConsumerService given consumerProperties which must contain a value for "bootstrap.servers".
String	getIngestionInfoFor(PubSubTopic versionTopic, PubSubTopicPartition pubSubTopicPartition, String regionName)
static int	getKeyLevelLockMaxPoolSizeBasedOnServerConfig(VeniceServerConfig serverConfig, int partitionCount)
long	getLatestOffsetBasedOnMetrics(String kafkaURL, PubSubTopic versionTopic, PubSubTopicPartition pubSubTopicPartition)
protected static Runnable	getStuckConsumerDetectionAndRepairRunnable(org.apache.logging.log4j.Logger logger, Time time, Map<String,AbstractKafkaConsumerService> kafkaServerToConsumerServiceMap, Map<String,StoreIngestionTask> versionTopicStoreIngestionTaskMapping, long stuckConsumerRepairThresholdMs, long nonExistingTopicIngestionTaskKillThresholdMs, long nonExistingTopicRetryIntervalMs, long consumerPollTrackerStaleThresholdMs, StuckConsumerRepairStats stuckConsumerRepairStats, Consumer<String> killIngestionTaskRunnable)
boolean	hasConsumerAssignedFor(String kafkaURL, PubSubTopic versionTopic, PubSubTopicPartition pubSubTopicPartition)
boolean	startInner()	IMPORTANT: All newly created KafkaConsumerService are already started in createKafkaConsumerService(Properties), if this is no longer the case in future, make sure to update the startInner logic here.
void	stopInner()
ConsumedDataReceiver<List<DefaultPubSubMessage>>	subscribeConsumerFor(String kafkaURL, StoreIngestionTask storeIngestionTask, PartitionReplicaIngestionContext partitionReplicaIngestionContext, PubSubPosition lastOffset)
void	unsubscribeConsumerFor(PubSubTopic versionTopic, PubSubTopicPartition pubSubTopicPartition)
void	unsubscribeConsumerFor(PubSubTopic versionTopic, PubSubTopicPartition pubSubTopicPartition, long timeoutMs)

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

AggKafkaConsumerService

public AggKafkaConsumerService(PubSubConsumerAdapterFactory consumerFactory,
 TopicManagerContext.PubSubPropertiesSupplier pubSubPropertiesSupplier,
 VeniceServerConfig serverConfig,
 IngestionThrottler ingestionThrottler,
 KafkaClusterBasedRecordThrottler kafkaClusterBasedRecordThrottler,
 io.tehuti.metrics.MetricsRepository metricsRepository,
 StaleTopicChecker staleTopicChecker,
 PubSubMessageDeserializer pubSubDeserializer,
 Consumer<String> killIngestionTaskRunnable,
 Function<String,Boolean> isAAOrWCEnabledFunc,
 ReadOnlyStoreRepository metadataRepository)

Method Details

startInner

public boolean startInner()

IMPORTANT: All newly created KafkaConsumerService are already started in createKafkaConsumerService(Properties), if this is no longer the case in future, make sure to update the startInner logic here.

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).

stopInner

public void stopInner()
               throws Exception

Specified by:

stopInner in class AbstractVeniceService

Throws:

Exception

getStuckConsumerDetectionAndRepairRunnable

protected static Runnable getStuckConsumerDetectionAndRepairRunnable(org.apache.logging.log4j.Logger logger,
 Time time,
 Map<String,AbstractKafkaConsumerService> kafkaServerToConsumerServiceMap,
 Map<String,StoreIngestionTask> versionTopicStoreIngestionTaskMapping,
 long stuckConsumerRepairThresholdMs,
 long nonExistingTopicIngestionTaskKillThresholdMs,
 long nonExistingTopicRetryIntervalMs,
 long consumerPollTrackerStaleThresholdMs,
 StuckConsumerRepairStats stuckConsumerRepairStats,
 Consumer<String> killIngestionTaskRunnable)

createKafkaConsumerService

public AbstractKafkaConsumerService createKafkaConsumerService(Properties consumerProperties)

Create a new KafkaConsumerService given consumerProperties which must contain a value for "bootstrap.servers". If a KafkaConsumerService for the given "bootstrap.servers" (Kafka URL) has already been created, this method returns the created KafkaConsumerService.

Parameters:

consumerProperties - consumer properties that are used to create KafkaConsumerService

hasConsumerAssignedFor

public boolean hasConsumerAssignedFor(String kafkaURL,
 PubSubTopic versionTopic,
 PubSubTopicPartition pubSubTopicPartition)

unsubscribeConsumerFor

public void unsubscribeConsumerFor(PubSubTopic versionTopic,
 PubSubTopicPartition pubSubTopicPartition)

unsubscribeConsumerFor

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

subscribeConsumerFor

public ConsumedDataReceiver<List<DefaultPubSubMessage>> subscribeConsumerFor(String kafkaURL,
 StoreIngestionTask storeIngestionTask,
 PartitionReplicaIngestionContext partitionReplicaIngestionContext,
 PubSubPosition lastOffset)

getLatestOffsetBasedOnMetrics

public long getLatestOffsetBasedOnMetrics(String kafkaURL,
 PubSubTopic versionTopic,
 PubSubTopicPartition pubSubTopicPartition)

getIngestionInfoFor

public String getIngestionInfoFor(PubSubTopic versionTopic,
 PubSubTopicPartition pubSubTopicPartition,
 String regionName)

getKeyLevelLockMaxPoolSizeBasedOnServerConfig

public static int getKeyLevelLockMaxPoolSizeBasedOnServerConfig(VeniceServerConfig serverConfig,
 int partitionCount)