Package com.linkedin.davinci.kafka.consumer

Class KafkaStoreIngestionService

java.lang.Object

com.linkedin.venice.service.AbstractVeniceService

com.linkedin.davinci.kafka.consumer.KafkaStoreIngestionService

All Implemented Interfaces:

StoreIngestionService, Closeable, AutoCloseable

public class KafkaStoreIngestionService
extends AbstractVeniceService
implements StoreIngestionService

Assumes: One to One mapping between a Venice Store and Kafka Topic. Manages Kafka topics and partitions that need to be consumed for the stores on this node. Launches StoreIngestionTask for each store version to consume and process messages. Uses the "new" Kafka Consumer.

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

KafkaStoreIngestionService(StorageService storageService, VeniceConfigLoader veniceConfigLoader, StorageMetadataService storageMetadataService, ClusterInfoProvider clusterInfoProvider, ReadOnlyStoreRepository metadataRepo, ReadOnlySchemaRepository schemaRepo, ReadOnlyLiveClusterConfigRepository liveClusterConfigRepository, io.tehuti.metrics.MetricsRepository metricsRepository, Optional<SchemaReader> kafkaMessageEnvelopeSchemaReader, Optional<ClientConfig> clientConfig, InternalAvroSpecificSerializer<PartitionState> partitionStateSerializer, Optional<HelixReadOnlyZKSharedSchemaRepository> zkSharedSchemaRepository, ICProvider icProvider, boolean isIsolatedIngestion, StorageEngineBackedCompressorFactory compressorFactory, Optional<ObjectCacheBackend> cacheBackend, boolean isDaVinciClient, RemoteIngestionRepairService remoteIngestionRepairService, PubSubClientsFactory pubSubClientsFactory, Optional<SSLFactory> sslFactory, HeartbeatMonitoringService heartbeatMonitoringService, Lazy<org.apache.helix.manager.zk.ZKHelixAdmin> zkHelixAdmin, AdaptiveThrottlerSignalService adaptiveThrottlerSignalService, Optional<com.linkedin.d2.balancer.D2Client> d2Client)

Method Summary

Modifier and Type	Method	Description
void	addIngestionNotifier(VeniceNotifier notifier)	Adds Notifier to get Notifications for get various status of the consumption tasks like start, completed, progress and error states.
void	attemptToPrintIngestionInfoFor(String storeName, Integer version, Integer partition, String regionName)
boolean	containsRunningConsumption(VeniceStoreVersionConfig veniceStore)	Check whether there is a running consumption task for given store.
boolean	containsRunningConsumption(String topic)	Check whether there is a running consumption task for given store version topic.
void	demoteToStandby(VeniceStoreVersionConfig veniceStoreVersionConfig, int partitionId, LeaderFollowerPartitionStateModel.LeaderSessionIdChecker checker)
CompletableFuture<Void>	dropStoragePartitionGracefully(VeniceStoreVersionConfig veniceStore, int partitionId)	Drops the corresponding Venice Partition gracefully.
AggVersionedIngestionStats	getAggVersionedIngestionStats()	Get AggVersionedStorageIngestionStats
AdminResponse	getConsumptionSnapshots(String topicName, ComplementSet<Integer> partitions)
Set<String>	getIngestingTopicsWithVersionStatusNotOnline()	Get topic names that are currently maintained by the ingestion service with corresponding version status not in an online state.
InternalDaVinciRecordTransformerConfig	getInternalRecordTransformerConfig(String storeName)
KafkaValueSerializer	getKafkaValueSerializer()
final ReadOnlyStoreRepository	getMetadataRepo()
ByteBuffer	getPartitionOffsetRecords(String topicName, int partition)	This method should only be called when the forked ingestion process is handing over ingestion task to main process.
PubSubContext	getPubSubContext()
Optional<PubSubPosition>	getPubSubPosition(VeniceStoreVersionConfig veniceStore, int partitionId, Long timestamp, PubSubPosition pubSubPosition)
StoreIngestionTask	getStoreIngestionTask(String topicName)
ByteBuffer	getStoreVersionCompressionDictionary(String topicName)
protected Map<String,StoreIngestionTask>	getTopicNameToIngestionTaskMap()
ReplicaIngestionResponse	getTopicPartitionIngestionContext(String versionTopic, String topicName, int partitionId)
VeniceConfigLoader	getVeniceConfigLoader()
VeniceWriterFactory	getVeniceWriterFactory()
boolean	hasCurrentVersionBootstrapping()
static boolean	hasCurrentVersionBootstrapping(Map<String,StoreIngestionTask> ingestionTaskMap)
boolean	isDaVinciClient()
boolean	isLiveUpdateSuppressionEnabled()
boolean	isPartitionConsuming(String topic, int partitionId)	Check whether the specified partition is still being consumed
boolean	killConsumptionTask(String topicName)	Kill all of running consumptions of given store.
void	promoteToLeader(VeniceStoreVersionConfig veniceStoreVersionConfig, int partitionId, LeaderFollowerPartitionStateModel.LeaderSessionIdChecker checker)
void	recordIngestionFailure(String storeName)
void	registerRecordTransformerConfig(String storeName, DaVinciRecordTransformerConfig recordTransformerConfig)
void	replaceAndAddTestNotifier(VeniceNotifier notifier)
void	shutdownStoreIngestionTask(String topicName)	This method closes the specified StoreIngestionTask and wait for up to 10 seconds for fully shutdown.
void	startConsumption(VeniceStoreVersionConfig veniceStore, int partitionId, Optional<PubSubPosition> pubSubPosition)	Starts consuming messages from Kafka Partition corresponding to Venice Partition.
boolean	startInner()	Starts the Kafka consumption tasks for already subscribed partitions.
CompletableFuture<Void>	stopConsumption(VeniceStoreVersionConfig veniceStore, int partitionId)	Stops consuming messages from Kafka Partition corresponding to Venice Partition.
void	stopConsumptionAndWait(VeniceStoreVersionConfig veniceStore, int partitionId, int sleepSeconds, int numRetries, boolean whetherToResetOffset)	Stops consuming messages from Kafka Partition corresponding to Venice Partition and wait up to (sleepSeconds * numRetires) to make sure partition consumption is stopped.
void	stopInner()	Stops all the Kafka consumption tasks.
void	syncTopicPartitionOffset(String topicName, int partition)	Updates offset metadata and sync to storage for specified topic partition.
boolean	topicPartitionHasAnyPendingActions(String topic, int partition)
void	updatePartitionOffsetRecords(String topicName, int partition, ByteBuffer offsetRecordByteBuffer)	This method updates all sub-partitions' latest offset records fetched from isolated ingestion process in main process, so main process's in-memory storage metadata service could be aware of the latest updates and will not re-start the ingestion from scratch.
protected void	updateStatsEmission(NavigableMap<String,StoreIngestionTask> taskMap, String storeName)	This function will go through all known ingestion task in this server node, find the task that matches the storeName and has the largest version number; if the task doesn't enable metric emission, enable it and update store ingestion stats.
protected void	updateStatsEmission(NavigableMap<String,StoreIngestionTask> taskMap, String storeName, int maximumVersion)	Find the task that matches both the storeName and maximumVersion number, enable metrics emission for this task and update ingestion stats with this task; disable metric emission for all the task that doesn't max version.
void	waitIngestionTaskToCompleteAllPartitionPendingActions(String topicName, int partition, long retryIntervalInMs, int numRetries)

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

KafkaStoreIngestionService

public KafkaStoreIngestionService(StorageService storageService,
 VeniceConfigLoader veniceConfigLoader,
 StorageMetadataService storageMetadataService,
 ClusterInfoProvider clusterInfoProvider,
 ReadOnlyStoreRepository metadataRepo,
 ReadOnlySchemaRepository schemaRepo,
 ReadOnlyLiveClusterConfigRepository liveClusterConfigRepository,
 io.tehuti.metrics.MetricsRepository metricsRepository,
 Optional<SchemaReader> kafkaMessageEnvelopeSchemaReader,
 Optional<ClientConfig> clientConfig,
 InternalAvroSpecificSerializer<PartitionState> partitionStateSerializer,
 Optional<HelixReadOnlyZKSharedSchemaRepository> zkSharedSchemaRepository,
 ICProvider icProvider,
 boolean isIsolatedIngestion,
 StorageEngineBackedCompressorFactory compressorFactory,
 Optional<ObjectCacheBackend> cacheBackend,
 boolean isDaVinciClient,
 RemoteIngestionRepairService remoteIngestionRepairService,
 PubSubClientsFactory pubSubClientsFactory,
 Optional<SSLFactory> sslFactory,
 HeartbeatMonitoringService heartbeatMonitoringService,
 Lazy<org.apache.helix.manager.zk.ZKHelixAdmin> zkHelixAdmin,
 AdaptiveThrottlerSignalService adaptiveThrottlerSignalService,
 Optional<com.linkedin.d2.balancer.D2Client> d2Client)

Method Details

startInner

public boolean startInner()

Starts the Kafka consumption tasks for already subscribed partitions.

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

hasCurrentVersionBootstrapping

public boolean hasCurrentVersionBootstrapping()

hasCurrentVersionBootstrapping

public static boolean hasCurrentVersionBootstrapping(Map<String,StoreIngestionTask> ingestionTaskMap)

stopInner

public void stopInner()

Stops all the Kafka consumption tasks. Closes all the Kafka clients.

Specified by:

stopInner in class AbstractVeniceService

getPubSubPosition

public Optional<PubSubPosition> getPubSubPosition(VeniceStoreVersionConfig veniceStore,
 int partitionId,
 Long timestamp,
 PubSubPosition pubSubPosition)

Specified by:

getPubSubPosition in interface StoreIngestionService

startConsumption

public void startConsumption(VeniceStoreVersionConfig veniceStore,
 int partitionId,
 Optional<PubSubPosition> pubSubPosition)

Starts consuming messages from Kafka Partition corresponding to Venice Partition. Subscribes to partition if required.

Specified by:

startConsumption in interface StoreIngestionService

Parameters:

veniceStore - Venice Store for the partition.

partitionId - Venice partition's id.

pubSubPosition -

shutdownStoreIngestionTask

public void shutdownStoreIngestionTask(String topicName)

This method closes the specified StoreIngestionTask and wait for up to 10 seconds for fully shutdown.

Parameters:

topicName - Topic name of the ingestion task to be shutdown.

promoteToLeader

public void promoteToLeader(VeniceStoreVersionConfig veniceStoreVersionConfig,
 int partitionId,
 LeaderFollowerPartitionStateModel.LeaderSessionIdChecker checker)

Specified by:

promoteToLeader in interface StoreIngestionService

demoteToStandby

public void demoteToStandby(VeniceStoreVersionConfig veniceStoreVersionConfig,
 int partitionId,
 LeaderFollowerPartitionStateModel.LeaderSessionIdChecker checker)

Specified by:

demoteToStandby in interface StoreIngestionService

waitIngestionTaskToCompleteAllPartitionPendingActions

public void waitIngestionTaskToCompleteAllPartitionPendingActions(String topicName,
 int partition,
 long retryIntervalInMs,
 int numRetries)

topicPartitionHasAnyPendingActions

public boolean topicPartitionHasAnyPendingActions(String topic,
 int partition)

isLiveUpdateSuppressionEnabled

public boolean isLiveUpdateSuppressionEnabled()

getVeniceConfigLoader

public VeniceConfigLoader getVeniceConfigLoader()

Specified by:

getVeniceConfigLoader in interface StoreIngestionService

getVeniceWriterFactory

public VeniceWriterFactory getVeniceWriterFactory()

Specified by:

getVeniceWriterFactory in interface StoreIngestionService

updateStatsEmission

protected void updateStatsEmission(NavigableMap<String,StoreIngestionTask> taskMap,
 String storeName,
 int maximumVersion)

Find the task that matches both the storeName and maximumVersion number, enable metrics emission for this task and update ingestion stats with this task; disable metric emission for all the task that doesn't max version.

updateStatsEmission

protected void updateStatsEmission(NavigableMap<String,StoreIngestionTask> taskMap,
 String storeName)

This function will go through all known ingestion task in this server node, find the task that matches the storeName and has the largest version number; if the task doesn't enable metric emission, enable it and update store ingestion stats.

stopConsumption

public CompletableFuture<Void> stopConsumption(VeniceStoreVersionConfig veniceStore,
 int partitionId)

Stops consuming messages from Kafka Partition corresponding to Venice Partition.

Specified by:

stopConsumption in interface StoreIngestionService

Parameters:

veniceStore - Venice Store for the partition.

partitionId - Venice partition's id.

stopConsumptionAndWait

public void stopConsumptionAndWait(VeniceStoreVersionConfig veniceStore,
 int partitionId,
 int sleepSeconds,
 int numRetries,
 boolean whetherToResetOffset)

Stops consuming messages from Kafka Partition corresponding to Venice Partition and wait up to (sleepSeconds * numRetires) to make sure partition consumption is stopped.

Specified by:

stopConsumptionAndWait in interface StoreIngestionService

Parameters:

veniceStore - Venice Store for the partition.

partitionId - Venice partition's id.

sleepSeconds -

numRetries -

dropStoragePartitionGracefully

public CompletableFuture<Void> dropStoragePartitionGracefully(VeniceStoreVersionConfig veniceStore,
 int partitionId)

Drops the corresponding Venice Partition gracefully. This should only be called after stopConsumptionAndWait(com.linkedin.davinci.config.VeniceStoreVersionConfig, int, int, int, boolean) has been called

Parameters:

veniceStore - Venice Store for the partition.

partitionId - Venice partition's id.

Returns:

a future for the drop partition action.

killConsumptionTask

public boolean killConsumptionTask(String topicName)

Description copied from interface: StoreIngestionService

Kill all of running consumptions of given store.

Specified by:

killConsumptionTask in interface StoreIngestionService

Parameters:

topicName - Venice topic (store and version number) for the corresponding consumer task that needs to be killed. No action is taken for invocations of killConsumptionTask on topics that are not in the map. This includes logging.

Returns:

true if a kill is needed and called, otherwise false

addIngestionNotifier

public void addIngestionNotifier(VeniceNotifier notifier)

Description copied from interface: StoreIngestionService

Adds Notifier to get Notifications for get various status of the consumption tasks like start, completed, progress and error states. Multiple Notifiers can be added for the same consumption tasks and all of them will be notified in order.

Specified by:

addIngestionNotifier in interface StoreIngestionService

replaceAndAddTestNotifier

public void replaceAndAddTestNotifier(VeniceNotifier notifier)

Specified by:

replaceAndAddTestNotifier in interface StoreIngestionService

containsRunningConsumption

public boolean containsRunningConsumption(VeniceStoreVersionConfig veniceStore)

Description copied from interface: StoreIngestionService

Check whether there is a running consumption task for given store.

Specified by:

containsRunningConsumption in interface StoreIngestionService

containsRunningConsumption

public boolean containsRunningConsumption(String topic)

Description copied from interface: StoreIngestionService

Check whether there is a running consumption task for given store version topic.

Specified by:

containsRunningConsumption in interface StoreIngestionService

isPartitionConsuming

public boolean isPartitionConsuming(String topic,
 int partitionId)

Description copied from interface: StoreIngestionService

Check whether the specified partition is still being consumed

Specified by:

isPartitionConsuming in interface StoreIngestionService

getIngestingTopicsWithVersionStatusNotOnline

public Set<String> getIngestingTopicsWithVersionStatusNotOnline()

Description copied from interface: StoreIngestionService

Get topic names that are currently maintained by the ingestion service with corresponding version status not in an online state. Topics with invalid store or version number are also included in the returned list.

Specified by:

getIngestingTopicsWithVersionStatusNotOnline in interface StoreIngestionService

Returns:

a Set of topic names.

recordIngestionFailure

public void recordIngestionFailure(String storeName)

Specified by:

recordIngestionFailure in interface StoreIngestionService

getAggVersionedIngestionStats

public AggVersionedIngestionStats getAggVersionedIngestionStats()

Description copied from interface: StoreIngestionService

Get AggVersionedStorageIngestionStats

Specified by:

getAggVersionedIngestionStats in interface StoreIngestionService

Returns:

an instance of AggVersionedIngestionStats

getStoreVersionCompressionDictionary

public ByteBuffer getStoreVersionCompressionDictionary(String topicName)

getStoreIngestionTask

public StoreIngestionTask getStoreIngestionTask(String topicName)

Specified by:

getStoreIngestionTask in interface StoreIngestionService

isDaVinciClient

public boolean isDaVinciClient()

getConsumptionSnapshots

public AdminResponse getConsumptionSnapshots(String topicName,
 ComplementSet<Integer> partitions)

getTopicPartitionIngestionContext

public ReplicaIngestionResponse getTopicPartitionIngestionContext(String versionTopic,
 String topicName,
 int partitionId)

updatePartitionOffsetRecords

public void updatePartitionOffsetRecords(String topicName,
 int partition,
 ByteBuffer offsetRecordByteBuffer)

This method updates all sub-partitions' latest offset records fetched from isolated ingestion process in main process, so main process's in-memory storage metadata service could be aware of the latest updates and will not re-start the ingestion from scratch.

getPartitionOffsetRecords

public ByteBuffer getPartitionOffsetRecords(String topicName,
 int partition)

This method should only be called when the forked ingestion process is handing over ingestion task to main process. It collects the user partition's latest OffsetRecords from partition consumption states (PCS). In theory, PCS should be available in this situation as we haven't unsubscribed from topic. If it is not available, we will throw exception as this is not as expected.

syncTopicPartitionOffset

public void syncTopicPartitionOffset(String topicName,
 int partition)

Updates offset metadata and sync to storage for specified topic partition. This method is invoked only when isolated ingestion process is reporting topic partition completion to make sure ingestion process is persisted.

getMetadataRepo

public final ReadOnlyStoreRepository getMetadataRepo()

getPubSubContext

public PubSubContext getPubSubContext()

getKafkaValueSerializer

public KafkaValueSerializer getKafkaValueSerializer()

getTopicNameToIngestionTaskMap

protected Map<String,StoreIngestionTask> getTopicNameToIngestionTaskMap()

registerRecordTransformerConfig

public void registerRecordTransformerConfig(String storeName,
 DaVinciRecordTransformerConfig recordTransformerConfig)

getInternalRecordTransformerConfig

public InternalDaVinciRecordTransformerConfig getInternalRecordTransformerConfig(String storeName)

attemptToPrintIngestionInfoFor

public void attemptToPrintIngestionInfoFor(String storeName,
 Integer version,
 Integer partition,
 String regionName)