Package com.linkedin.venice.pushmonitor

Class AbstractPushMonitor

java.lang.Object

com.linkedin.venice.pushmonitor.AbstractPushMonitor

All Implemented Interfaces:

RoutingDataRepository.RoutingDataChangedListener, PartitionStatusListener, PushMonitor

Direct Known Subclasses:

PartitionStatusBasedPushMonitor

public abstract class AbstractPushMonitor
extends Object
implements PushMonitor, PartitionStatusListener, RoutingDataRepository.RoutingDataChangedListener

AbstractPushMonitor is a high level abstraction that manages OfflinePushStatus. Depending on the implementation, it collects info from different paths and updates push status accordingly. At present, push status has 1 initial state ExecutionStatus.STARTED and 2 end states ExecutionStatus.COMPLETED and ExecutionStatus.ERROR. State mutation is unidirectional and once it reaches to either end state, we stop mutating it. Check OfflinePushStatus for more details.

Field Summary

Fields

Modifier and Type	Field	Description
static final int	MAX_PUSH_TO_KEEP

Constructor Summary

Constructors

Constructor

Description

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

Method Summary

Modifier and Type	Method	Description
protected abstract ExecutionStatusWithDetails	checkPushStatus(OfflinePushStatus pushStatus, PartitionAssignment partitionAssignment, DisableReplicaCallback callback)
protected void	checkWhetherToStartEOPProcedures(OfflinePushStatus offlinePushStatus)
void	cleanupStoreStatus(String storeName)	Clean up all push statuses related to a store including all error pushes.
protected DisableReplicaCallback	getDisableReplicaCallback(String kafkaTopic)
protected long	getDurationInSec(OfflinePushStatus pushStatus)
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
protected OfflinePushStatus	getOfflinePush(String topic)
protected OfflinePushAccessor	getOfflinePushAccessor()
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)	Get ongoing incremental push versions from the push status store
ExecutionStatus	getPushStatus(String topic)
ExecutionStatusWithDetails	getPushStatusAndDetails(String topic)
protected ReadWriteStoreRepository	getReadWriteStoreRepository()
abstract List<Instance>	getReadyToServeInstances(PartitionAssignment partitionAssignment, int partitionId)
RealTimeTopicSwitcher	getRealTimeTopicSwitcher()
protected RoutingDataRepository	getRoutingDataRepository()
List<String>	getTopicsOfOngoingOfflinePushes()	Find all ongoing pushes then return the topics associated to those pushes.
List<UncompletedPartition>	getUncompletedPartitions(String topic)
protected void	handleCompletedPush(String topic)
protected void	handleErrorPush(String topic, ExecutionStatusWithDetails executionStatusWithDetails)
protected void	handleTerminalOfflinePushUpdate(OfflinePushStatus pushStatus, ExecutionStatusWithDetails statusWithDetails)	This method will unsubscribe external view changes and is intended to be called when the statues are terminable.
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	onCustomizedViewAdded(PartitionAssignment partitionAssignment)
void	onCustomizedViewChange(PartitionAssignment partitionAssignment)
void	onExternalViewChange(PartitionAssignment partitionAssignment)	Handle routing data changed event.
protected void	onPartitionStatusChange(OfflinePushStatus offlinePushStatus)
void	onPartitionStatusChange(String topic, ReadOnlyPartitionStatus partitionStatus)
void	onRoutingDataDeleted(String kafkaTopic)
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.
protected void	retireOldErrorPushes(String storeName)
void	setRealTimeTopicSwitcher(RealTimeTopicSwitcher realTimeTopicSwitcher)	For testing only; in order to override the topicReplicator with mocked Replicator.
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.
protected void	updateOfflinePush(String topic)
protected OfflinePushStatus	updatePushStatus(OfflinePushStatus expectedCurrPushStatus, ExecutionStatus newExecutionStatus, Optional<String> newExecutionStatusDetails)	Direct calls to updatePushStatus should be made carefully.

Methods inherited from class java.lang.Object

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

Field Details

MAX_PUSH_TO_KEEP

public static final int MAX_PUSH_TO_KEEP

See Also:

• Constant Field Values

Constructor Details

AbstractPushMonitor

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

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

getOfflinePush

protected OfflinePushStatus getOfflinePush(String topic)

updateOfflinePush

protected void updateOfflinePush(String topic)

getPushStatus

public ExecutionStatus getPushStatus(String topic)

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)

Get ongoing incremental push versions from the push status store

Specified by:

getOngoingIncrementalPushVersions in interface PushMonitor

Parameters:

kafkaTopic - kafka topic belonging to a store version for which we are fetching ongoing inc-pushe versions

pushStatusStoreReader - - push status system store reader

Returns:

set of (supposedly) ongoing incremental pushes

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.

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

getUncompletedPartitions

public List<UncompletedPartition> getUncompletedPartitions(String topic)

Specified by:

getUncompletedPartitions 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

retireOldErrorPushes

protected void retireOldErrorPushes(String storeName)

checkPushStatus

protected abstract ExecutionStatusWithDetails checkPushStatus(OfflinePushStatus pushStatus,
 PartitionAssignment partitionAssignment,
 DisableReplicaCallback callback)

getReadyToServeInstances

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

Specified by:

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

updatePushStatus

protected OfflinePushStatus updatePushStatus(OfflinePushStatus expectedCurrPushStatus,
 ExecutionStatus newExecutionStatus,
 Optional<String> newExecutionStatusDetails)

Direct calls to updatePushStatus should be made carefully. e.g. calling with ExecutionStatus.ERROR or other terminal status update should be made through handleOfflinePushUpdate. That method will then invoke handleErrorPush and perform relevant operations to handle the ERROR status update properly.

Returns:

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

getDurationInSec

protected long getDurationInSec(OfflinePushStatus pushStatus)

getOfflinePushAccessor

protected OfflinePushAccessor getOfflinePushAccessor()

getReadWriteStoreRepository

protected ReadWriteStoreRepository getReadWriteStoreRepository()

getRoutingDataRepository

protected RoutingDataRepository getRoutingDataRepository()

onPartitionStatusChange

public void onPartitionStatusChange(String topic,
 ReadOnlyPartitionStatus partitionStatus)

Specified by:

onPartitionStatusChange in interface PartitionStatusListener

Specified by:

onPartitionStatusChange in interface RoutingDataRepository.RoutingDataChangedListener

onPartitionStatusChange

protected void onPartitionStatusChange(OfflinePushStatus offlinePushStatus)

getDisableReplicaCallback

protected DisableReplicaCallback getDisableReplicaCallback(String kafkaTopic)

onExternalViewChange

public void onExternalViewChange(PartitionAssignment partitionAssignment)

Description copied from interface: RoutingDataRepository.RoutingDataChangedListener

Handle routing data changed event.

Specified by:

onExternalViewChange in interface RoutingDataRepository.RoutingDataChangedListener

Parameters:

partitionAssignment - Newest partitions assignments information including resource name and all of instances assigned to this resource. If the number of partition is 0, it means the kafka topic is deleted.

onCustomizedViewChange

public void onCustomizedViewChange(PartitionAssignment partitionAssignment)

Specified by:

onCustomizedViewChange in interface RoutingDataRepository.RoutingDataChangedListener

onCustomizedViewAdded

public void onCustomizedViewAdded(PartitionAssignment partitionAssignment)

Specified by:

onCustomizedViewAdded in interface RoutingDataRepository.RoutingDataChangedListener

onRoutingDataDeleted

public void onRoutingDataDeleted(String kafkaTopic)

Specified by:

onRoutingDataDeleted in interface RoutingDataRepository.RoutingDataChangedListener

checkWhetherToStartEOPProcedures

protected void checkWhetherToStartEOPProcedures(OfflinePushStatus offlinePushStatus)

handleTerminalOfflinePushUpdate

protected void handleTerminalOfflinePushUpdate(OfflinePushStatus pushStatus,
 ExecutionStatusWithDetails statusWithDetails)

This method will unsubscribe external view changes and is intended to be called when the statues are terminable.

handleCompletedPush

protected void handleCompletedPush(String topic)

handleErrorPush

protected void handleErrorPush(String topic,
 ExecutionStatusWithDetails executionStatusWithDetails)

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

setRealTimeTopicSwitcher

public void setRealTimeTopicSwitcher(RealTimeTopicSwitcher realTimeTopicSwitcher)

For testing only; in order to override the topicReplicator with mocked Replicator.

getRealTimeTopicSwitcher

public RealTimeTopicSwitcher getRealTimeTopicSwitcher()

isOfflinePushMonitorDaVinciPushStatusEnabled

public boolean isOfflinePushMonitorDaVinciPushStatusEnabled()

Specified by:

isOfflinePushMonitorDaVinciPushStatusEnabled in interface PushMonitor