Package com.linkedin.venice.helix

Class VeniceOfflinePushMonitorAccessor

java.lang.Object

com.linkedin.venice.helix.VeniceOfflinePushMonitorAccessor

All Implemented Interfaces:

OfflinePushAccessor

public class VeniceOfflinePushMonitorAccessor
extends Object
implements OfflinePushAccessor

Venice implementation of OfflinePushAccessor. All the statuses would be stored on Zookeeper and this class provides the ways to read/write/create/remove status from ZK.

As this class is only an accessor but not a repository so it will not cache anything in local memory. In other words it's stateless and Thread-Safe.

The data structure on ZK would be:

• /OfflinePushes/$topic -> push status for $topic
• /OfflinePushes/$topic/$partitionId -> partition status including all of replicas statuses for $topic and $partitionId.

Field Summary

Fields

Modifier and Type	Field	Description
static final String	OFFLINE_PUSH_SUB_PATH

Constructor Summary

Constructors

Constructor	Description
VeniceOfflinePushMonitorAccessor(String clusterName, org.apache.helix.manager.zk.ZkBaseDataAccessor<OfflinePushStatus> offlinePushStatusAccessor, org.apache.helix.manager.zk.ZkBaseDataAccessor<PartitionStatus> partitionStatusAccessor)	For testing purpose only.
VeniceOfflinePushMonitorAccessor(String clusterName, org.apache.helix.zookeeper.impl.client.ZkClient zkClient, HelixAdapterSerializer adapter)
VeniceOfflinePushMonitorAccessor(String clusterName, org.apache.helix.zookeeper.impl.client.ZkClient zkClient, HelixAdapterSerializer adapter, int refreshAttemptsForZkReconnect, long refreshIntervalForZkReconnectInMs)

Method Summary

Modifier and Type	Method	Description
void	batchUpdateReplicaIncPushStatus(String kafkaTopic, int partitionId, String instanceId, long progress, List<String> pendingReportIncPushVersionList)
void	createOfflinePushStatusAndItsPartitionStatuses(OfflinePushStatus pushStatus)	Create offline push and its partition statues on persistent storage.
void	deleteOfflinePushStatusAndItsPartitionStatuses(String kafkaTopic)	Delete offline push and its partition statuses from persistent storage.
final String	getOfflinePushStatuesParentPath()
org.apache.helix.manager.zk.ZkBaseDataAccessor<OfflinePushStatus>	getOfflinePushStatusAccessor()	For testing only.
OfflinePushStatus	getOfflinePushStatusAndItsPartitionStatuses(String kafkaTopic)	Read one offline push and its partitions status from persistent storage.
Optional<Long>	getOfflinePushStatusCreationTime(String kafkaTopic)	Get the creation time of the given offline push ZK path; the creation time is set by System.currentTimeMillis().
protected PartitionStatus	getPartitionStatus(String topic, int partitionId)	Get one partition status ZNode from ZK by given topic and partition.
protected List<PartitionStatus>	getPartitionStatuses(String topic, int partitionCount)	Get all partition status ZNodes under offline push of given topic from ZK.
List<OfflinePushStatus>	loadOfflinePushStatusesAndPartitionStatuses()	Load entire database of offline pushes including offline pushes statues and their partitions statuses from persistent storage.
List<String>	loadOfflinePushStatusPaths()	Load store versions of all existing push statuses on Zookeeper, without actually deserializing the contents
void	subscribePartitionStatusChange(OfflinePushStatus pushStatus, PartitionStatusListener listener)	Subscribe the data change of partition status.
void	subscribePushStatusCreationChange(org.apache.helix.zookeeper.zkclient.IZkChildListener childListener)	Subscribe a child listener that listens to OfflinePushStatus creation/deleted.
void	unsubscribePartitionsStatusChange(OfflinePushStatus pushStatus, PartitionStatusListener listener)	Unsubscribe the data change of partition status.
void	unsubscribePartitionsStatusChange(String topicName, int partitionCount, PartitionStatusListener listener)	Unsubscribe the data change of partition status.
void	unsubscribePushStatusCreationChange(org.apache.helix.zookeeper.zkclient.IZkChildListener childListener)	Unsubscribe a child listener
void	updateOfflinePushStatus(OfflinePushStatus pushStatus)	Update status of the given offline push to persistent storage.
void	updateReplicaStatus(String topic, int partitionId, String instanceId, ExecutionStatus status, long progress, String incrementalPushVersion)	Update one particular replica status and progress by given topic, partition and instanceId to the persistent storage.
void	updateReplicaStatus(String topic, int partitionId, String instanceId, ExecutionStatus status, String incrementalPushVersion)	Update one particular replica status only by given topic, partition and instanceId to the persistent storage.

Methods inherited from class java.lang.Object

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

Field Details

OFFLINE_PUSH_SUB_PATH

public static final String OFFLINE_PUSH_SUB_PATH

See Also:

• Constant Field Values

Constructor Details

VeniceOfflinePushMonitorAccessor

public VeniceOfflinePushMonitorAccessor(String clusterName,
 org.apache.helix.zookeeper.impl.client.ZkClient zkClient,
 HelixAdapterSerializer adapter)

VeniceOfflinePushMonitorAccessor

public VeniceOfflinePushMonitorAccessor(String clusterName,
 org.apache.helix.zookeeper.impl.client.ZkClient zkClient,
 HelixAdapterSerializer adapter,
 int refreshAttemptsForZkReconnect,
 long refreshIntervalForZkReconnectInMs)

VeniceOfflinePushMonitorAccessor

public VeniceOfflinePushMonitorAccessor(String clusterName,
 org.apache.helix.manager.zk.ZkBaseDataAccessor<OfflinePushStatus> offlinePushStatusAccessor,
 org.apache.helix.manager.zk.ZkBaseDataAccessor<PartitionStatus> partitionStatusAccessor)

For testing purpose only.

Method Details

loadOfflinePushStatusesAndPartitionStatuses

public List<OfflinePushStatus> loadOfflinePushStatusesAndPartitionStatuses()

Description copied from interface: OfflinePushAccessor

Load entire database of offline pushes including offline pushes statues and their partitions statuses from persistent storage.

Specified by:

loadOfflinePushStatusesAndPartitionStatuses in interface OfflinePushAccessor

loadOfflinePushStatusPaths

public List<String> loadOfflinePushStatusPaths()

Description copied from interface: OfflinePushAccessor

Load store versions of all existing push statuses on Zookeeper, without actually deserializing the contents

Specified by:

loadOfflinePushStatusPaths in interface OfflinePushAccessor

getOfflinePushStatusAndItsPartitionStatuses

public OfflinePushStatus getOfflinePushStatusAndItsPartitionStatuses(String kafkaTopic)

Description copied from interface: OfflinePushAccessor

Read one offline push and its partitions status from persistent storage.

Specified by:

getOfflinePushStatusAndItsPartitionStatuses in interface OfflinePushAccessor

getOfflinePushStatusCreationTime

public Optional<Long> getOfflinePushStatusCreationTime(String kafkaTopic)

Description copied from interface: OfflinePushAccessor

Get the creation time of the given offline push ZK path; the creation time is set by System.currentTimeMillis(). If the path does not exist or error when trying to get creation time, return an empty Optional.

Specified by:

getOfflinePushStatusCreationTime in interface OfflinePushAccessor

updateOfflinePushStatus

public void updateOfflinePushStatus(OfflinePushStatus pushStatus)

Description copied from interface: OfflinePushAccessor

Update status of the given offline push to persistent storage.

Specified by:

updateOfflinePushStatus in interface OfflinePushAccessor

createOfflinePushStatusAndItsPartitionStatuses

public void createOfflinePushStatusAndItsPartitionStatuses(OfflinePushStatus pushStatus)

Description copied from interface: OfflinePushAccessor

Create offline push and its partition statues on persistent storage.

Specified by:

createOfflinePushStatusAndItsPartitionStatuses in interface OfflinePushAccessor

deleteOfflinePushStatusAndItsPartitionStatuses

public void deleteOfflinePushStatusAndItsPartitionStatuses(String kafkaTopic)

Description copied from interface: OfflinePushAccessor

Delete offline push and its partition statuses from persistent storage.

Specified by:

deleteOfflinePushStatusAndItsPartitionStatuses in interface OfflinePushAccessor

Parameters:

kafkaTopic - version topic

updateReplicaStatus

public void updateReplicaStatus(String topic,
 int partitionId,
 String instanceId,
 ExecutionStatus status,
 long progress,
 String incrementalPushVersion)

Description copied from interface: OfflinePushAccessor

Update one particular replica status and progress by given topic, partition and instanceId to the persistent storage.

Specified by:

updateReplicaStatus in interface OfflinePushAccessor

updateReplicaStatus

public void updateReplicaStatus(String topic,
 int partitionId,
 String instanceId,
 ExecutionStatus status,
 String incrementalPushVersion)

Description copied from interface: OfflinePushAccessor

Update one particular replica status only by given topic, partition and instanceId to the persistent storage.

Specified by:

updateReplicaStatus in interface OfflinePushAccessor

batchUpdateReplicaIncPushStatus

public void batchUpdateReplicaIncPushStatus(String kafkaTopic,
 int partitionId,
 String instanceId,
 long progress,
 List<String> pendingReportIncPushVersionList)

Specified by:

batchUpdateReplicaIncPushStatus in interface OfflinePushAccessor

subscribePartitionStatusChange

public void subscribePartitionStatusChange(OfflinePushStatus pushStatus,
 PartitionStatusListener listener)

Description copied from interface: OfflinePushAccessor

Subscribe the data change of partition status.

Specified by:

subscribePartitionStatusChange in interface OfflinePushAccessor

unsubscribePartitionsStatusChange

public void unsubscribePartitionsStatusChange(OfflinePushStatus pushStatus,
 PartitionStatusListener listener)

Description copied from interface: OfflinePushAccessor

Unsubscribe the data change of partition status.

Specified by:

unsubscribePartitionsStatusChange in interface OfflinePushAccessor

unsubscribePartitionsStatusChange

public void unsubscribePartitionsStatusChange(String topicName,
 int partitionCount,
 PartitionStatusListener listener)

Description copied from interface: OfflinePushAccessor

Unsubscribe the data change of partition status.

Specified by:

unsubscribePartitionsStatusChange in interface OfflinePushAccessor

subscribePushStatusCreationChange

public void subscribePushStatusCreationChange(org.apache.helix.zookeeper.zkclient.IZkChildListener childListener)

Description copied from interface: OfflinePushAccessor

Subscribe a child listener that listens to OfflinePushStatus creation/deleted.

Specified by:

subscribePushStatusCreationChange in interface OfflinePushAccessor

unsubscribePushStatusCreationChange

public void unsubscribePushStatusCreationChange(org.apache.helix.zookeeper.zkclient.IZkChildListener childListener)

Description copied from interface: OfflinePushAccessor

Unsubscribe a child listener

Specified by:

unsubscribePushStatusCreationChange in interface OfflinePushAccessor

getPartitionStatus

protected PartitionStatus getPartitionStatus(String topic,
 int partitionId)

Get one partition status ZNode from ZK by given topic and partition.

getPartitionStatuses

protected List<PartitionStatus> getPartitionStatuses(String topic,
 int partitionCount)

Get all partition status ZNodes under offline push of given topic from ZK. The partition statuses paths for a topic are created in ZK one by one; Helix doesn't guarantee that all the paths are created atomically; therefore, it's possible that a partial list is returned. This function would take cover of all edges cases -- empty response as well as partial response by filling the missing partitions. The returned partition status list is ordered by partition Id.

getOfflinePushStatuesParentPath

public final String getOfflinePushStatuesParentPath()

getOfflinePushStatusAccessor

public org.apache.helix.manager.zk.ZkBaseDataAccessor<OfflinePushStatus> getOfflinePushStatusAccessor()

For testing only.