Package com.linkedin.venice.pushmonitor

Class PushMonitorDelegator

java.lang.Object

com.linkedin.venice.pushmonitor.PushMonitorDelegator

All Implemented Interfaces:

PushMonitor

public class PushMonitorDelegator
extends Object
implements PushMonitor

This class maintains the mapping of Kafka topic to each PushMonitor instance and delegates calls to the correct instance.

Constructor Summary

Constructors

Constructor

Description

PushMonitorDelegator(String clusterName, RoutingDataRepository routingDataRepository, OfflinePushAccessor offlinePushAccessor, StoreCleaner storeCleaner, ReadWriteStoreRepository metadataRepository, AggPushHealthStats aggPushHealthStats, RealTimeTopicSwitcher leaderFollowerTopicReplicator, ClusterLockManager clusterLockManager, String aggregateRealTimeSourceKafkaUrl, List<String> activeActiveRealTimeSourceKafkaURLs, HelixAdminClient helixAdminClient, VeniceControllerClusterConfig controllerConfig, PushStatusStoreReader pushStatusStoreReader, DisabledPartitionStats disabledPartitionStats, VeniceWriterFactory veniceWriterFactory, AbstractPushMonitor.CurrentVersionChangeNotifier currentVersionChangeNotifier)

Method Summary

Modifier and Type	Method	Description
void	cleanupStoreStatus(String storeName)	Clean up all push statuses related to a store including all error pushes.
ExecutionStatusWithDetails	getIncrementalPushStatusAndDetails(String kafkaTopic, String incrementalPushVersion, HelixCustomizedViewOfflinePushRepository customizedViewRepo)	Returns incremental push's status read from (ZooKeeper backed) OfflinePushStatus
ExecutionStatusWithDetails	getIncrementalPushStatusFromPushStatusStore(String kafkaTopic, String incrementalPushVersion, HelixCustomizedViewOfflinePushRepository customizedViewRepo, PushStatusStoreReader pushStatusStoreReader)	Returns incremental push's status read from push status store
OfflinePushStatus	getOfflinePushOrThrow(String topic)	Return a push's status (push status contains the history statuses and current status) and throw exception if it doesn't exist
List<OfflinePushStatus>	getOfflinePushStatusForStore(String storeName)
Set<String>	getOngoingIncrementalPushVersions(String kafkaTopic)	Check if there are any ongoing incremental push for the given version topic.
Set<String>	getOngoingIncrementalPushVersions(String kafkaTopic, PushStatusStoreReader pushStatusStoreReader)
ExecutionStatusWithDetails	getPushStatusAndDetails(String topic)
List<Instance>	getReadyToServeInstances(PartitionAssignment partitionAssignment, int partitionId)
List<String>	getTopicsOfOngoingOfflinePushes()	Find all ongoing pushes then return the topics associated to those pushes.
List<UncompletedPartition>	getUncompletedPartitions(String topic)
boolean	isOfflinePushMonitorDaVinciPushStatusEnabled()
void	loadAllPushes()	load previous push statuses when push monitor restarts
void	markOfflinePushAsError(String topic, String statusDetails)	Mark a push to be as error.
void	recordPushPreparationDuration(String topic, long offlinePushWaitTimeInSecond)	stats related operation.
OfflinePushStatus	refreshAndUpdatePushStatus(String kafkaTopic, ExecutionStatus newStatus, Optional<String> newStatusDetails)	Here, we refresh the push status, in order to avoid a race condition where a small job could already be completed.
void	startMonitorOfflinePush(String kafkaTopic, int numberOfPartition, int replicaFactor, OfflinePushStrategy strategy)
void	stopAllMonitoring()	Stop monitoring all offline pushes.
void	stopMonitorOfflinePush(String kafkaTopic, boolean deletePushStatus, boolean isForcedDelete)	Stop monitoring a push.

Methods inherited from class java.lang.Object

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

Constructor Details

PushMonitorDelegator

public PushMonitorDelegator(String clusterName,
 RoutingDataRepository routingDataRepository,
 OfflinePushAccessor offlinePushAccessor,
 StoreCleaner storeCleaner,
 ReadWriteStoreRepository metadataRepository,
 AggPushHealthStats aggPushHealthStats,
 RealTimeTopicSwitcher leaderFollowerTopicReplicator,
 ClusterLockManager clusterLockManager,
 String aggregateRealTimeSourceKafkaUrl,
 List<String> activeActiveRealTimeSourceKafkaURLs,
 HelixAdminClient helixAdminClient,
 VeniceControllerClusterConfig controllerConfig,
 PushStatusStoreReader pushStatusStoreReader,
 DisabledPartitionStats disabledPartitionStats,
 VeniceWriterFactory veniceWriterFactory,
 AbstractPushMonitor.CurrentVersionChangeNotifier currentVersionChangeNotifier)

Method Details

loadAllPushes

public void loadAllPushes()

Description copied from interface: PushMonitor

load previous push statuses when push monitor restarts

Specified by:

loadAllPushes in interface PushMonitor

startMonitorOfflinePush

public void startMonitorOfflinePush(String kafkaTopic,
 int numberOfPartition,
 int replicaFactor,
 OfflinePushStrategy strategy)

Specified by:

startMonitorOfflinePush in interface PushMonitor

strategy - on which criteria a push is considered to be completed. check PushStatusDecider for more details Start monitoring a new push

stopMonitorOfflinePush

public void stopMonitorOfflinePush(String kafkaTopic,
 boolean deletePushStatus,
 boolean isForcedDelete)

Description copied from interface: PushMonitor

Stop monitoring a push. This function should be called when: 1. Retire a version; 2. Leader controller transits to standby; in this case, controller shouldn't delete any push status.

Specified by:

stopMonitorOfflinePush in interface PushMonitor

stopAllMonitoring

public void stopAllMonitoring()

Description copied from interface: PushMonitor

Stop monitoring all offline pushes.

Specified by:

stopAllMonitoring in interface PushMonitor

cleanupStoreStatus

public void cleanupStoreStatus(String storeName)

Description copied from interface: PushMonitor

Clean up all push statuses related to a store including all error pushes. This is called when a store gets deleted.

Specified by:

cleanupStoreStatus in interface PushMonitor

getOfflinePushOrThrow

public OfflinePushStatus getOfflinePushOrThrow(String topic)

Description copied from interface: PushMonitor

Return a push's status (push status contains the history statuses and current status) and throw exception if it doesn't exist

Specified by:

getOfflinePushOrThrow in interface PushMonitor

getPushStatusAndDetails

public ExecutionStatusWithDetails getPushStatusAndDetails(String topic)

Specified by:

getPushStatusAndDetails in interface PushMonitor

Returns:

return the current status. If it's in error, some debugging info is also presented.

getUncompletedPartitions

public List<UncompletedPartition> getUncompletedPartitions(String topic)

Specified by:

getUncompletedPartitions in interface PushMonitor

getIncrementalPushStatusAndDetails

public ExecutionStatusWithDetails getIncrementalPushStatusAndDetails(String kafkaTopic,
 String incrementalPushVersion,
 HelixCustomizedViewOfflinePushRepository customizedViewRepo)

Description copied from interface: PushMonitor

Returns incremental push's status read from (ZooKeeper backed) OfflinePushStatus

Specified by:

getIncrementalPushStatusAndDetails in interface PushMonitor

getIncrementalPushStatusFromPushStatusStore

public ExecutionStatusWithDetails getIncrementalPushStatusFromPushStatusStore(String kafkaTopic,
 String incrementalPushVersion,
 HelixCustomizedViewOfflinePushRepository customizedViewRepo,
 PushStatusStoreReader pushStatusStoreReader)

Description copied from interface: PushMonitor

Returns incremental push's status read from push status store

Specified by:

getIncrementalPushStatusFromPushStatusStore in interface PushMonitor

getOngoingIncrementalPushVersions

public Set<String> getOngoingIncrementalPushVersions(String kafkaTopic)

Description copied from interface: PushMonitor

Check if there are any ongoing incremental push for the given version topic.

Specified by:

getOngoingIncrementalPushVersions in interface PushMonitor

Parameters:

kafkaTopic - to check for ongoing incremental push

Returns:

ongoing incremental push versions if there are any, otherwise an empty set is returned.

getOngoingIncrementalPushVersions

public Set<String> getOngoingIncrementalPushVersions(String kafkaTopic,
 PushStatusStoreReader pushStatusStoreReader)

Specified by:

getOngoingIncrementalPushVersions in interface PushMonitor

getTopicsOfOngoingOfflinePushes

public List<String> getTopicsOfOngoingOfflinePushes()

Description copied from interface: PushMonitor

Find all ongoing pushes then return the topics associated to those pushes.

Specified by:

getTopicsOfOngoingOfflinePushes in interface PushMonitor

getOfflinePushStatusForStore

public List<OfflinePushStatus> getOfflinePushStatusForStore(String storeName)

Specified by:

getOfflinePushStatusForStore in interface PushMonitor

markOfflinePushAsError

public void markOfflinePushAsError(String topic,
 String statusDetails)

Description copied from interface: PushMonitor

Mark a push to be as error. This is usually called when push is killed.

Specified by:

markOfflinePushAsError in interface PushMonitor

refreshAndUpdatePushStatus

public OfflinePushStatus refreshAndUpdatePushStatus(String kafkaTopic,
 ExecutionStatus newStatus,
 Optional<String> newStatusDetails)

Description copied from interface: PushMonitor

Here, we refresh the push status, in order to avoid a race condition where a small job could already be completed. Previously, we would clobber the COMPLETED status with STARTED, which would stall the job forever. Now, since we get the refreshed status, we can validate whether a transition to is valid, before making the change. If if wouldn't be valid (because the job already completed or already failed, for example), then we leave the status as is, rather than adding in the new details.

Specified by:

refreshAndUpdatePushStatus in interface PushMonitor

Returns:

the new OfflinePushStatus if it was updated, or null if the update was skipped.

recordPushPreparationDuration

public void recordPushPreparationDuration(String topic,
 long offlinePushWaitTimeInSecond)

Description copied from interface: PushMonitor

stats related operation. TODO: we may want to move it out of the interface

Specified by:

recordPushPreparationDuration in interface PushMonitor

getReadyToServeInstances

public List<Instance> getReadyToServeInstances(PartitionAssignment partitionAssignment,
 int partitionId)

Specified by:

getReadyToServeInstances in interface PushMonitor

isOfflinePushMonitorDaVinciPushStatusEnabled

public boolean isOfflinePushMonitorDaVinciPushStatusEnabled()

Specified by:

isOfflinePushMonitorDaVinciPushStatusEnabled in interface PushMonitor