Package com.linkedin.davinci.consumer

Class VeniceChangelogConsumerImpl<K,V>

java.lang.Object

com.linkedin.davinci.consumer.VeniceChangelogConsumerImpl<K,V>

All Implemented Interfaces:

VeniceChangelogConsumer<K,V>, AutoCloseable

Direct Known Subclasses:

VeniceAfterImageConsumerImpl

public class VeniceChangelogConsumerImpl<K,V>
extends Object
implements VeniceChangelogConsumer<K,V>

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
protected class	VeniceChangelogConsumerImpl.HeartbeatReporterThread

Field Summary

Fields

Modifier and Type	Field	Description
protected final BasicConsumerStats	changeCaptureStats
protected final ChangelogClientConfig	changelogClientConfig
protected final ChunkAssembler	chunkAssembler
protected final String	clientRegionName
protected final CompressorFactory	compressorFactory
protected final VeniceConcurrentHashMap<Integer,AtomicLong>	consumerSequenceIdGeneratorMap
protected final long	consumerSequenceIdStartingValue
protected final Map<Integer,Long>	currentVersionLastHeartbeat
protected final VeniceChangelogConsumerImpl<K,V>.HeartbeatReporterThread	heartbeatReporterThread
protected final AtomicBoolean	isSubscribed
protected final RecordDeserializer<K>	keyDeserializer
protected final AtomicReference<CountDownLatch>	onGoingVersionSwapSignal
protected final Map<Integer,Boolean>	partitionToBootstrapState
protected final PubSubConsumerAdapter	pubSubConsumer
protected final PubSubMessageDeserializer	pubSubMessageDeserializer
protected final VeniceConcurrentHashMap<String,VeniceCompressor>	pubSubTopicNameToCompressorMap
protected final PubSubTopicRepository	pubSubTopicRepository
protected final ExecutorService	seekExecutorService
protected final long	startTimestamp
protected StoreDeserializerCache<V>	storeDeserializerCache
protected final String	storeName
protected NativeMetadataRepositoryViewAdapter	storeRepository
protected long	subscribeTime
protected final ReadWriteLock	subscriptionLock
protected Time	time
protected final int	totalRegionCount
protected final boolean	versionSwapByControlMessage
protected VersionSwapMessageState	versionSwapMessageState	Interaction of this field should acquire the subscriptionLock.writeLock()
protected final long	versionSwapTimeoutInMs

Constructor Summary

Constructors

Constructor	Description
VeniceChangelogConsumerImpl(ChangelogClientConfig changelogClientConfig, PubSubConsumerAdapter pubSubConsumer, PubSubMessageDeserializer pubSubMessageDeserializer, VeniceChangelogConsumerClientFactory veniceChangelogConsumerClientFactory)

Method Summary

Modifier and Type	Method	Description
void	close()	Release the internal resources.
protected Optional<PubSubMessage<K,ChangeEvent<V>,VeniceChangeCoordinate>>	convertPubSubMessageToChangeEvent(DefaultPubSubMessage message, PubSubTopicPartition pubSubTopicPartition)	Converts a raw PubSub message into a change event message suitable for returning to the consumer.
protected BasicConsumerStats	getChangeCaptureStats()
protected PubSubTopic	getCurrentServingVersionTopic()
protected VeniceChangelogConsumerImpl<K,V>.HeartbeatReporterThread	getHeartbeatReporterThread()
Map<Integer,Long>	getLastHeartbeatPerPartition()	Returns the timestamp of the last heartbeat received for each subscribed partition.
protected long	getNextConsumerSequenceId(int partition)
int	getPartitionCount()
protected List<PubSubTopicPartition>	getPartitionListToSubscribe(Set<Integer> partitions, Set<PubSubTopicPartition> topicPartitionSet, PubSubTopic topic)
protected Long	getSubscribeTime()
protected Set<PubSubTopicPartition>	getTopicAssignment()
protected VeniceCompressor	getVersionCompressor(PubSubTopic pubSubTopic)
protected boolean	handleControlMessage(ControlMessage controlMessage, PubSubTopicPartition pubSubTopicPartition, byte[] key, long timestamp, PubSubPosition position)	Handle control message from the given topic.
protected boolean	handleVersionSwapMessageInVT(ControlMessage controlMessage, PubSubTopicPartition pubSubTopicPartition, PubSubPosition position)
protected CompletableFuture<Void>	internalFindNewVersionCheckpoints(String oldVersionTopic, String newVersionTopic, long generationId, Set<Integer> partitions)
protected Collection<PubSubMessage<K,ChangeEvent<V>,VeniceChangeCoordinate>>	internalPoll(long timeoutInMs)
protected Collection<PubSubMessage<K,ChangeEvent<V>,VeniceChangeCoordinate>>	internalPoll(long timeoutInMs, boolean includeControlMessage)
protected CompletableFuture<Void>	internalSeek(Set<Integer> partitions, PubSubTopic targetTopic, com.linkedin.davinci.consumer.VeniceChangelogConsumerImpl.SeekFunction seekAction)
CompletableFuture<Void>	internalSeekToTimestamps(Map<Integer,Long> timestamps)
CompletableFuture<Void>	internalSeekToTimestamps(Map<Integer,Long> timestamps, org.apache.logging.log4j.Logger logger)
protected CompletableFuture<Void>	internalSubscribe(Set<Integer> partitions, PubSubTopic topic)
protected void	internalUnsubscribe(Set<Integer> partitions, boolean isForVersionSwap)
boolean	isCaughtUp()	Checks whether all subscribed partitions are caught up during bootstrap.
void	pause()	Pause the client on all subscriptions.
void	pause(Set<Integer> partitions)	Pause the client on all or subset of partitions this client is subscribed to.
Collection<PubSubMessage<K,ChangeEvent<V>,VeniceChangeCoordinate>>	poll(long timeoutInMs)	Polling function to get any available messages from the underlying system for all partitions subscribed.
void	resume()	Pause the client on all subscriptions.
void	resume(Set<Integer> partitions)	Resume the client on all or a subset of partitions this client is subscribed to and has paused.
CompletableFuture<Void>	seekToBeginningOfPush()	Seek to the beginning of the push for subscribed partitions.
CompletableFuture<Void>	seekToBeginningOfPush(Set<Integer> partitions)	Seek to the beginning of the push for a set of partitions.
CompletableFuture<Void>	seekToCheckpoint(Set<VeniceChangeCoordinate> checkpoints)	Seek the provided checkpoints for the specified partitions.
CompletableFuture<Void>	seekToEndOfPush()	Seek to the end of the push for all subscribed partitions.
CompletableFuture<Void>	seekToEndOfPush(Set<Integer> partitions)	Seek to the end of the last push for a given set of partitions.
CompletableFuture<Void>	seekToTail()	Seek to the end of events which have been transmitted to Venice for all subscribed partitions.
CompletableFuture<Void>	seekToTail(Set<Integer> partitions)	Seek to the end of events which have been transmitted to Venice and start consuming new events.
CompletableFuture<Void>	seekToTimestamp(Long timestamp)	Seek to the specified timestamp for all subscribed partitions.
CompletableFuture<Void>	seekToTimestamps(Map<Integer,Long> timestamps)	Seek to the provided timestamps for the specified partitions based on wall clock time for when this message was processed by Venice and produced to change capture.
protected void	setStoreRepository(NativeMetadataRepositoryViewAdapter repository)
CompletableFuture<Void>	subscribe(Set<Integer> partitions)	Subscribe a set of partitions for a store to this VeniceChangelogConsumer, with the earliest position in the topic for each partition.
CompletableFuture<Void>	subscribeAll()	Subscribe all partitions belonging to a specific store, with the earliest position in the topic for each partition.
protected boolean	switchToNewTopic(PubSubTopicPartition newTopicPartition)
protected void	synchronousSeek(Set<Integer> partitions, PubSubTopic targetTopic, com.linkedin.davinci.consumer.VeniceChangelogConsumerImpl.SeekFunction seekAction)
protected void	synchronousSeekToCheckpoint(Set<VeniceChangeCoordinate> checkpoints)
void	throwIfReadsDisabled()
void	unsubscribe(Set<Integer> partitions)	Stop ingesting messages from a set of partitions for a specific store.
void	unsubscribeAll()	Stop ingesting messages from all partitions.

Methods inherited from class java.lang.Object

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

Field Details

subscribeTime

protected long subscribeTime

subscriptionLock

protected final ReadWriteLock subscriptionLock

compressorFactory

protected final CompressorFactory compressorFactory

pubSubTopicNameToCompressorMap

protected final VeniceConcurrentHashMap<String,VeniceCompressor> pubSubTopicNameToCompressorMap

storeDeserializerCache

protected StoreDeserializerCache<V> storeDeserializerCache

storeRepository

protected NativeMetadataRepositoryViewAdapter storeRepository

partitionToBootstrapState

protected final Map<Integer,Boolean> partitionToBootstrapState

startTimestamp

protected final long startTimestamp

isSubscribed

protected final AtomicBoolean isSubscribed

keyDeserializer

protected final RecordDeserializer<K> keyDeserializer

storeName

protected final String storeName

pubSubConsumer

protected final PubSubConsumerAdapter pubSubConsumer

pubSubTopicRepository

protected final PubSubTopicRepository pubSubTopicRepository

pubSubMessageDeserializer

protected final PubSubMessageDeserializer pubSubMessageDeserializer

seekExecutorService

protected final ExecutorService seekExecutorService

currentVersionLastHeartbeat

protected final Map<Integer,Long> currentVersionLastHeartbeat

changelogClientConfig

protected final ChangelogClientConfig changelogClientConfig

chunkAssembler

protected final ChunkAssembler chunkAssembler

changeCaptureStats

protected final BasicConsumerStats changeCaptureStats

heartbeatReporterThread

protected final VeniceChangelogConsumerImpl<K,V>.HeartbeatReporterThread heartbeatReporterThread

consumerSequenceIdGeneratorMap

protected final VeniceConcurrentHashMap<Integer,AtomicLong> consumerSequenceIdGeneratorMap

consumerSequenceIdStartingValue

protected final long consumerSequenceIdStartingValue

versionSwapByControlMessage

protected final boolean versionSwapByControlMessage

clientRegionName

protected final String clientRegionName

totalRegionCount

protected final int totalRegionCount

versionSwapTimeoutInMs

protected final long versionSwapTimeoutInMs

onGoingVersionSwapSignal

protected final AtomicReference<CountDownLatch> onGoingVersionSwapSignal

versionSwapMessageState

protected volatile VersionSwapMessageState versionSwapMessageState

Interaction of this field should acquire the subscriptionLock.writeLock()

time

protected Time time

Constructor Details

VeniceChangelogConsumerImpl

public VeniceChangelogConsumerImpl(ChangelogClientConfig changelogClientConfig,
 PubSubConsumerAdapter pubSubConsumer,
 PubSubMessageDeserializer pubSubMessageDeserializer,
 VeniceChangelogConsumerClientFactory veniceChangelogConsumerClientFactory)

Method Details

throwIfReadsDisabled

public void throwIfReadsDisabled()

getPartitionCount

public int getPartitionCount()

Specified by:

getPartitionCount in interface VeniceChangelogConsumer<K,V>

Returns:

total number of store partitions

subscribe

public CompletableFuture<Void> subscribe(Set<Integer> partitions)

Description copied from interface: VeniceChangelogConsumer

Subscribe a set of partitions for a store to this VeniceChangelogConsumer, with the earliest position in the topic for each partition. The VeniceChangelogConsumer should try to consume messages from all partitions that are subscribed to it.

Specified by:

subscribe in interface VeniceChangelogConsumer<K,V>

Parameters:

partitions - the set of partition to subscribe and consume

Returns:

a future which completes when data from the partitions are ready to be consumed

internalSubscribe

protected CompletableFuture<Void> internalSubscribe(Set<Integer> partitions,
 PubSubTopic topic)

getVersionCompressor

protected VeniceCompressor getVersionCompressor(PubSubTopic pubSubTopic)

seekToBeginningOfPush

public CompletableFuture<Void> seekToBeginningOfPush(Set<Integer> partitions)

Description copied from interface: VeniceChangelogConsumer

Seek to the beginning of the push for a set of partitions. This is analogous to doing a bootstrap of data for the consumer. This seek will ONLY seek to the beginning of the version which is currently serving data, and the consumer will switch to reading data from a new version (should one get created) once it has read up to the point in the change capture stream that indicates the version swap (which can only occur after consuming all the data in the last push). This instructs the consumer to consume data from the batch push.

Specified by:

seekToBeginningOfPush in interface VeniceChangelogConsumer<K,V>

Parameters:

partitions - the set of partitions to seek with

Returns:

a future which completes when the operation has succeeded for all partitions.

seekToBeginningOfPush

public CompletableFuture<Void> seekToBeginningOfPush()

Description copied from interface: VeniceChangelogConsumer

Seek to the beginning of the push for subscribed partitions. See VeniceChangelogConsumer.seekToBeginningOfPush(Set) for more information.

Specified by:

seekToBeginningOfPush in interface VeniceChangelogConsumer<K,V>

Returns:

a future which completes when the partitions are ready to be consumed data

seekToEndOfPush

public CompletableFuture<Void> seekToEndOfPush(Set<Integer> partitions)

Description copied from interface: VeniceChangelogConsumer

Seek to the end of the last push for a given set of partitions. This instructs the consumer to begin consuming events which are transmitted to Venice following the last batch push.

Specified by:

seekToEndOfPush in interface VeniceChangelogConsumer<K,V>

Parameters:

partitions - the set of partitions to seek with

Returns:

a future which completes when the operation has succeeded for all partitions.

pause

public void pause()

Description copied from interface: VeniceChangelogConsumer

Pause the client on all subscriptions. See VeniceChangelogConsumer.pause(Set) for more information.

Specified by:

pause in interface VeniceChangelogConsumer<K,V>

resume

public void resume(Set<Integer> partitions)

Description copied from interface: VeniceChangelogConsumer

Resume the client on all or a subset of partitions this client is subscribed to and has paused.

Specified by:

resume in interface VeniceChangelogConsumer<K,V>

resume

public void resume()

Description copied from interface: VeniceChangelogConsumer

Pause the client on all subscriptions. See VeniceChangelogConsumer.resume(Set) for more information.

Specified by:

resume in interface VeniceChangelogConsumer<K,V>

pause

public void pause(Set<Integer> partitions)

Description copied from interface: VeniceChangelogConsumer

Pause the client on all or subset of partitions this client is subscribed to. Calls to VeniceChangelogConsumer.poll(long) will not return results from paused partitions until VeniceChangelogConsumer.resume(Set) is called again later for those partitions.

Specified by:

pause in interface VeniceChangelogConsumer<K,V>

seekToEndOfPush

public CompletableFuture<Void> seekToEndOfPush()

Description copied from interface: VeniceChangelogConsumer

Seek to the end of the push for all subscribed partitions. See VeniceChangelogConsumer.seekToEndOfPush(Set) for more information.

Specified by:

seekToEndOfPush in interface VeniceChangelogConsumer<K,V>

Returns:

a future which completes when the operation has succeeded for all partitions.

seekToTail

public CompletableFuture<Void> seekToTail(Set<Integer> partitions)

Description copied from interface: VeniceChangelogConsumer

Seek to the end of events which have been transmitted to Venice and start consuming new events. This will ONLY consume events transmitted via nearline and incremental push. It will not read batch push data.

Specified by:

seekToTail in interface VeniceChangelogConsumer<K,V>

Parameters:

partitions - the set of partitions to seek with

Returns:

a future which completes when the operation has succeeded for all partitions.

getCurrentServingVersionTopic

protected PubSubTopic getCurrentServingVersionTopic()

seekToTail

public CompletableFuture<Void> seekToTail()

Description copied from interface: VeniceChangelogConsumer

Seek to the end of events which have been transmitted to Venice for all subscribed partitions. See VeniceChangelogConsumer.seekToTail(Set) for more information.

Specified by:

seekToTail in interface VeniceChangelogConsumer<K,V>

Returns:

a future which completes when the operation has succeeded for all partitions.

seekToCheckpoint

public CompletableFuture<Void> seekToCheckpoint(Set<VeniceChangeCoordinate> checkpoints)
                                         throws VeniceCoordinateOutOfRangeException

Description copied from interface: VeniceChangelogConsumer

Seek the provided checkpoints for the specified partitions. Note about checkpoints: Checkpoints have the following properties and should be considered: - Checkpoints are NOT comparable or valid across partitions. - Checkpoints are NOT comparable or valid across regions - Checkpoints are NOT comparable across store versions - It is not possible to determine the number of events between two checkpoints - It is possible that a checkpoint is no longer on retention. In such case, we will return an exception to the caller.

Specified by:

seekToCheckpoint in interface VeniceChangelogConsumer<K,V>

Returns:

a future which completes when seek has completed for all partitions

Throws:

VeniceCoordinateOutOfRangeException - if passed checkpoint is no longer valid

synchronousSeekToCheckpoint

protected void synchronousSeekToCheckpoint(Set<VeniceChangeCoordinate> checkpoints)

subscribeAll

public CompletableFuture<Void> subscribeAll()

Description copied from interface: VeniceChangelogConsumer

Subscribe all partitions belonging to a specific store, with the earliest position in the topic for each partition.

Specified by:

subscribeAll in interface VeniceChangelogConsumer<K,V>

Returns:

a future which completes when all partitions are ready to be consumed data

seekToTimestamps

public CompletableFuture<Void> seekToTimestamps(Map<Integer,Long> timestamps)

Description copied from interface: VeniceChangelogConsumer

Seek to the provided timestamps for the specified partitions based on wall clock time for when this message was processed by Venice and produced to change capture. Note, this API can only be used to seek on nearline data applied to the current serving version in Venice. This will not seek on data transmitted via Batch Push. If the provided timestamp is lower than the earliest timestamp on a given stream, the earliest event will be returned. THIS WILL NOT SEEK TO DATA WHICH WAS APPLIED ON A PREVIOUS VERSION. You should never seek back in time to a timestamp which is smaller than the current time - rewindTimeInSeconds configured in the hybrid settings for this Venice store. The timestamp passed to this function should be associated to timestamps processed by this interface. The timestamp returned by PubSubMessage.getPubSubMessageTime() refers to the time when Venice processed the event, and calls to this method will seek based on that sequence of events. Note: it bears no relation to timestamps provided by upstream producers when writing to Venice where a user may optionally provide a timestamp at time of producing a record.

Specified by:

seekToTimestamps in interface VeniceChangelogConsumer<K,V>

Parameters:

timestamps - a map keyed by a partition ID, and the timestamp checkpoints to seek for each partition.

Returns:

internalSeekToTimestamps

public CompletableFuture<Void> internalSeekToTimestamps(Map<Integer,Long> timestamps)

internalSeekToTimestamps

public CompletableFuture<Void> internalSeekToTimestamps(Map<Integer,Long> timestamps,
 org.apache.logging.log4j.Logger logger)

seekToTimestamp

public CompletableFuture<Void> seekToTimestamp(Long timestamp)

Description copied from interface: VeniceChangelogConsumer

Seek to the specified timestamp for all subscribed partitions. See VeniceChangelogConsumer.seekToTimestamps(Map) for more information.

Specified by:

seekToTimestamp in interface VeniceChangelogConsumer<K,V>

Returns:

a future which completes when the operation has succeeded for all partitions.

isCaughtUp

public boolean isCaughtUp()

Description copied from interface: VeniceChangelogConsumer

Checks whether all subscribed partitions are caught up during bootstrap. If a partition's (currentTimestamp - latestMessageTimestamp) is smaller or equal to 1 min, we consider this partition is caught up.

Specified by:

isCaughtUp in interface VeniceChangelogConsumer<K,V>

Returns:

True if all subscribed partitions have caught up.

internalSeek

protected CompletableFuture<Void> internalSeek(Set<Integer> partitions,
 PubSubTopic targetTopic,
 com.linkedin.davinci.consumer.VeniceChangelogConsumerImpl.SeekFunction seekAction)

synchronousSeek

protected void synchronousSeek(Set<Integer> partitions,
 PubSubTopic targetTopic,
 com.linkedin.davinci.consumer.VeniceChangelogConsumerImpl.SeekFunction seekAction)

getPartitionListToSubscribe

protected List<PubSubTopicPartition> getPartitionListToSubscribe(Set<Integer> partitions,
 Set<PubSubTopicPartition> topicPartitionSet,
 PubSubTopic topic)

unsubscribe

public void unsubscribe(Set<Integer> partitions)

Description copied from interface: VeniceChangelogConsumer

Stop ingesting messages from a set of partitions for a specific store.

Specified by:

unsubscribe in interface VeniceChangelogConsumer<K,V>

Parameters:

partitions - The set of topic partitions to unsubscribe

internalUnsubscribe

protected void internalUnsubscribe(Set<Integer> partitions,
 boolean isForVersionSwap)

unsubscribeAll

public void unsubscribeAll()

Description copied from interface: VeniceChangelogConsumer

Stop ingesting messages from all partitions.

Specified by:

unsubscribeAll in interface VeniceChangelogConsumer<K,V>

poll

public Collection<PubSubMessage<K,ChangeEvent<V>,VeniceChangeCoordinate>> poll(long timeoutInMs)

Description copied from interface: VeniceChangelogConsumer

Polling function to get any available messages from the underlying system for all partitions subscribed.

Specified by:

poll in interface VeniceChangelogConsumer<K,V>

Parameters:

timeoutInMs - The maximum time to block/wait in between two polling requests (must not be greater than Long.MAX_VALUE milliseconds)

Returns:

a collection of messages since the last fetch for the subscribed list of topic partitions

internalPoll

protected Collection<PubSubMessage<K,ChangeEvent<V>,VeniceChangeCoordinate>> internalPoll(long timeoutInMs,
 boolean includeControlMessage)

internalPoll

protected Collection<PubSubMessage<K,ChangeEvent<V>,VeniceChangeCoordinate>> internalPoll(long timeoutInMs)

handleControlMessage

protected boolean handleControlMessage(ControlMessage controlMessage,
 PubSubTopicPartition pubSubTopicPartition,
 byte[] key,
 long timestamp,
 PubSubPosition position)

Handle control message from the given topic. Returns true if a topic switch should occur and records should be returned

Parameters:

controlMessage - the encountered control message

pubSubTopicPartition - the topic that this control message was concerned

Returns:

true if a version switch happened, false if otherwise

convertPubSubMessageToChangeEvent

protected Optional<PubSubMessage<K,ChangeEvent<V>,VeniceChangeCoordinate>> convertPubSubMessageToChangeEvent(DefaultPubSubMessage message,
 PubSubTopicPartition pubSubTopicPartition)

Converts a raw PubSub message into a change event message suitable for returning to the consumer. During a version swap, a low watermark approach is used for the VeniceChangeCoordinate returned to prevent users from seeking in between a sequence of related version swap messages within a partition.

handleVersionSwapMessageInVT

protected boolean handleVersionSwapMessageInVT(ControlMessage controlMessage,
 PubSubTopicPartition pubSubTopicPartition,
 PubSubPosition position)

internalFindNewVersionCheckpoints

protected CompletableFuture<Void> internalFindNewVersionCheckpoints(String oldVersionTopic,
 String newVersionTopic,
 long generationId,
 Set<Integer> partitions)

getTopicAssignment

protected Set<PubSubTopicPartition> getTopicAssignment()

switchToNewTopic

protected boolean switchToNewTopic(PubSubTopicPartition newTopicPartition)

close

public void close()

Description copied from interface: VeniceChangelogConsumer

Release the internal resources.

Specified by:

close in interface AutoCloseable

Specified by:

close in interface VeniceChangelogConsumer<K,V>

setStoreRepository

protected void setStoreRepository(NativeMetadataRepositoryViewAdapter repository)

getSubscribeTime

protected Long getSubscribeTime()

getHeartbeatReporterThread

protected VeniceChangelogConsumerImpl<K,V>.HeartbeatReporterThread getHeartbeatReporterThread()

getChangeCaptureStats

protected BasicConsumerStats getChangeCaptureStats()

getNextConsumerSequenceId

protected long getNextConsumerSequenceId(int partition)

getLastHeartbeatPerPartition

public Map<Integer,Long> getLastHeartbeatPerPartition()

Description copied from interface: VeniceChangelogConsumer

Returns the timestamp of the last heartbeat received for each subscribed partition. Heartbeats are messages sent periodically by Venice servers to measure lag.

Specified by:

getLastHeartbeatPerPartition in interface VeniceChangelogConsumer<K,V>

Returns:

a map of partition number to the timestamp, in milliseconds, of the last heartbeat received for that partition.