org.apache.storm.cluster

Class StormClusterStateImpl

java.lang.Object

org.apache.storm.cluster.StormClusterStateImpl

All Implemented Interfaces:

IStormClusterState

public class StormClusterStateImpl
extends Object
implements IStormClusterState

Constructor Summary

Constructors

Constructor and Description
StormClusterStateImpl(IStateStorage stateStorage, ILocalAssignmentsBackend assignmentsassignmentsBackend, ClusterStateContext context, boolean shouldCloseStateStorageOnDisconnect)

Method Summary

Modifier and Type	Method and Description
void	activateStorm(String stormId, StormBase stormBase, Map<String,Object> topoConf)
List<String>	activeKeys()
List<String>	activeStorms()
void	addNimbusHost(String nimbusId, NimbusSummary nimbusSummary)
void	addPrivateWorkerKey(WorkerTokenServiceType type, String topologyId, long keyVersion, PrivateWorkerKey key) Store a new version of a private key.
Assignment	assignmentInfo(String stormId, Runnable callback) Get the assignment based on storm id from local backend.
VersionedData<Assignment>	assignmentInfoWithVersion(String stormId, Runnable callback)
List<String>	assignments(Runnable callback)
Map<String,Assignment>	assignmentsInfo() Get all the topologies assignments mapping stormId -> Assignment from local backend.
Integer	assignmentVersion(String stormId, Runnable callback)
List<String>	backpressureTopologies() Get backpressure topologies.
List<String>	blobstore(Runnable callback)
List<String>	blobstoreInfo(String blobKey)
Credentials	credentials(String stormId, Runnable callback)
void	deleteTopologyProfileRequests(String stormId, ProfileRequest profileRequest)
void	disconnect()
List<ErrorInfo>	errors(String stormId, String componentId)
List<String>	errorTopologies()
Map<ExecutorInfo,ExecutorBeat>	executorBeats(String stormId, Map<List<Long>,NodeInfo> executorNodePort) need to take executor->node+port in explicitly so that we don’t run into a situation where a long dead worker with a skewed clock overrides all the timestamps.
NimbusInfo	getLeader(Runnable callback) Get leader info from state store, which was written when a master gains leadership.
long	getNextPrivateWorkerKeyVersion(WorkerTokenServiceType type, String topologyId) Get the next key version number that should be used for this topology id.
PrivateWorkerKey	getPrivateWorkerKey(WorkerTokenServiceType type, String topologyId, long keyVersion) Get a private key used to validate a token is correct.
List<ProfileRequest>	getTopologyProfileRequests(String stormId)
ClusterWorkerHeartbeat	getWorkerHeartbeat(String stormId, String node, Long port)
List<ProfileRequest>	getWorkerProfileRequests(String stormId, NodeInfo nodeInfo)
List<String>	heartbeatStorms()
Set<String>	idsOfTopologiesWithPrivateWorkerKeys() Get a list of all topologyIds that currently have private worker keys stored, of any kind.
boolean	isAssignmentsBackendSynchronized() Flag to indicate if the assignments synced successfully, see IStormClusterState.syncRemoteAssignments(Map).
protected void	issueCallback(AtomicReference<Runnable> cb)
protected void	issueMapCallback(ConcurrentHashMap<String,Runnable> callbackConcurrentHashMap, String key)
ErrorInfo	lastError(String stormId, String componentId)
List<NimbusSummary>	nimbuses()
Assignment	remoteAssignmentInfo(String stormId, Runnable callback) Get the assignment based on storm id from remote state store, eg: ZK.
void	removeAllPrivateWorkerKeys(String topologyId) Remove all of the worker keys for a given topology.
void	removeBackpressure(String stormId) Remove backpressure.
void	removeBlobstoreKey(String blobKey)
void	removeExpiredPrivateWorkerKeys(String topologyId) Remove all keys for the given topology that have expired.
void	removeKeyVersion(String blobKey)
void	removeStorm(String stormId)
void	removeStormBase(String stormId)
void	removeWorkerBackpressure(String stormId, String node, Long port) Remove worker backpressure.
void	removeWorkerHeartbeat(String stormId, String node, Long port)
void	reportError(String stormId, String componentId, String node, Long port, Throwable error)
void	setAssignment(String stormId, Assignment info, Map<String,Object> topoConf)
void	setAssignmentsBackendSynchronized() Mark the assignments as synced successfully, see IStormClusterState.isAssignmentsBackendSynchronized().
void	setCredentials(String stormId, Credentials creds, Map<String,Object> topoConf)
void	setTopologyLogConfig(String stormId, LogConfig logConfig, Map<String,Object> topoConf)
void	setupBackpressure(String stormId, Map<String,Object> topoConf) Setup backpressure.
void	setupBlob(String key, NimbusInfo nimbusInfo, Integer versionInfo)
void	setupErrors(String stormId, Map<String,Object> topoConf)
void	setupHeatbeats(String stormId, Map<String,Object> topoConf)
void	setWorkerProfileRequest(String stormId, ProfileRequest profileRequest)
StormBase	stormBase(String stormId, Runnable callback) Get a storm base for a topology.
String	stormId(String stormName) Get storm id from passed name, null if the name doesn’t exist on cluster.
void	supervisorHeartbeat(String supervisorId, SupervisorInfo info)
SupervisorInfo	supervisorInfo(String supervisorId)
List<String>	supervisors(Runnable callback)
void	syncRemoteAssignments(Map<String,byte[]> remote) Sync the remote state store assignments to local backend, used when master gains leadership, see LeaderListenerCallback.
void	syncRemoteIds(Map<String,String> remote) Sync all the active storm ids of the cluster, used now when master gains leadership.
void	teardownHeartbeats(String stormId)
void	teardownTopologyErrors(String stormId)
boolean	topologyBackpressure(String stormId, long timeoutMs, Runnable callback) Check whether a topology is in throttle-on status or not: if the backpresure/storm-id dir is not empty, this topology has throttle-on, otherwise throttle-off.
LogConfig	topologyLogConfig(String stormId, Runnable cb)
void	updateStorm(String stormId, StormBase newElems) To update this function due to APersistentMap/APersistentSet is clojure’s structure.
void	workerHeartbeat(String stormId, String node, Long port, ClusterWorkerHeartbeat info)

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.apache.storm.cluster.IStormClusterState

allSupervisorInfo, allSupervisorInfo, getTopoId, topologyBases

Constructor Detail

StormClusterStateImpl

public StormClusterStateImpl(IStateStorage stateStorage,
                             ILocalAssignmentsBackend assignmentsassignmentsBackend,
                             ClusterStateContext context,
                             boolean shouldCloseStateStorageOnDisconnect)
                      throws Exception

Throws:

Exception

Method Detail

issueCallback

protected void issueCallback(AtomicReference<Runnable> cb)

issueMapCallback

protected void issueMapCallback(ConcurrentHashMap<String,Runnable> callbackConcurrentHashMap,
                                String key)

assignments

public List<String> assignments(Runnable callback)

Specified by:

assignments in interface IStormClusterState

assignmentInfo

public Assignment assignmentInfo(String stormId,
                                 Runnable callback)

Description copied from interface: IStormClusterState

Get the assignment based on storm id from local backend.

Specified by:

assignmentInfo in interface IStormClusterState

Parameters:

stormId - topology id

callback - callback function

Returns:

Assignment

remoteAssignmentInfo

public Assignment remoteAssignmentInfo(String stormId,
                                       Runnable callback)

Description copied from interface: IStormClusterState

Get the assignment based on storm id from remote state store, eg: ZK.

Specified by:

remoteAssignmentInfo in interface IStormClusterState

Parameters:

stormId - topology id

callback - callback function

Returns:

Assignment

assignmentsInfo

public Map<String,Assignment> assignmentsInfo()

Description copied from interface: IStormClusterState

Get all the topologies assignments mapping stormId -> Assignment from local backend.

Specified by:

assignmentsInfo in interface IStormClusterState

Returns:

stormId -> Assignment mapping

syncRemoteAssignments

public void syncRemoteAssignments(Map<String,byte[]> remote)

Description copied from interface: IStormClusterState

Sync the remote state store assignments to local backend, used when master gains leadership, see LeaderListenerCallback.

Specified by:

syncRemoteAssignments in interface IStormClusterState

Parameters:

remote - assigned assignments for a specific IStormClusterState instance, usually a supervisor/node.

isAssignmentsBackendSynchronized

public boolean isAssignmentsBackendSynchronized()

Description copied from interface: IStormClusterState

Flag to indicate if the assignments synced successfully, see IStormClusterState.syncRemoteAssignments(Map).

Specified by:

isAssignmentsBackendSynchronized in interface IStormClusterState

Returns:

true if is synced successfully

setAssignmentsBackendSynchronized

public void setAssignmentsBackendSynchronized()

Description copied from interface: IStormClusterState

Mark the assignments as synced successfully, see IStormClusterState.isAssignmentsBackendSynchronized().

Specified by:

setAssignmentsBackendSynchronized in interface IStormClusterState

assignmentInfoWithVersion

public VersionedData<Assignment> assignmentInfoWithVersion(String stormId,
                                                           Runnable callback)

Specified by:

assignmentInfoWithVersion in interface IStormClusterState

assignmentVersion

public Integer assignmentVersion(String stormId,
                                 Runnable callback)
                          throws Exception

Specified by:

assignmentVersion in interface IStormClusterState

Throws:

Exception

blobstoreInfo

public List<String> blobstoreInfo(String blobKey)

Specified by:

blobstoreInfo in interface IStormClusterState

nimbuses

public List<NimbusSummary> nimbuses()

Specified by:

nimbuses in interface IStormClusterState

addNimbusHost

public void addNimbusHost(String nimbusId,
                          NimbusSummary nimbusSummary)

Specified by:

addNimbusHost in interface IStormClusterState

activeStorms

public List<String> activeStorms()

Specified by:

activeStorms in interface IStormClusterState

stormBase

public StormBase stormBase(String stormId,
                           Runnable callback)

Description copied from interface: IStormClusterState

Get a storm base for a topology.

Specified by:

stormBase in interface IStormClusterState

Parameters:

stormId - the id of the topology

callback - something to call if the data changes (best effort)

Returns:

the StormBase or null if it is not alive.

stormId

public String stormId(String stormName)

Description copied from interface: IStormClusterState

Get storm id from passed name, null if the name doesn’t exist on cluster.

Specified by:

stormId in interface IStormClusterState

Parameters:

stormName - storm name

Returns:

storm id

syncRemoteIds

public void syncRemoteIds(Map<String,String> remote)

Description copied from interface: IStormClusterState

Sync all the active storm ids of the cluster, used now when master gains leadership.

Specified by:

syncRemoteIds in interface IStormClusterState

Parameters:

remote - stormName -> stormId mapping

getWorkerHeartbeat

public ClusterWorkerHeartbeat getWorkerHeartbeat(String stormId,
                                                 String node,
                                                 Long port)

Specified by:

getWorkerHeartbeat in interface IStormClusterState

getWorkerProfileRequests

public List<ProfileRequest> getWorkerProfileRequests(String stormId,
                                                     NodeInfo nodeInfo)

Specified by:

getWorkerProfileRequests in interface IStormClusterState

getTopologyProfileRequests

public List<ProfileRequest> getTopologyProfileRequests(String stormId)

Specified by:

getTopologyProfileRequests in interface IStormClusterState

setWorkerProfileRequest

public void setWorkerProfileRequest(String stormId,
                                    ProfileRequest profileRequest)

Specified by:

setWorkerProfileRequest in interface IStormClusterState

deleteTopologyProfileRequests

public void deleteTopologyProfileRequests(String stormId,
                                          ProfileRequest profileRequest)

Specified by:

deleteTopologyProfileRequests in interface IStormClusterState

executorBeats

public Map<ExecutorInfo,ExecutorBeat> executorBeats(String stormId,
                                                    Map<List<Long>,NodeInfo> executorNodePort)

need to take executor->node+port in explicitly so that we don’t run into a situation where a long dead worker with a skewed clock overrides all the timestamps. By only checking heartbeats with an assigned node+port, and only reading executors from that heartbeat that are actually assigned, we avoid situations like that.

Specified by:

executorBeats in interface IStormClusterState

Parameters:

stormId - topology id

executorNodePort - executor id -> node + port

Returns:

mapping of executorInfo -> executor beat

supervisors

public List<String> supervisors(Runnable callback)

Specified by:

supervisors in interface IStormClusterState

supervisorInfo

public SupervisorInfo supervisorInfo(String supervisorId)

Specified by:

supervisorInfo in interface IStormClusterState

setupHeatbeats

public void setupHeatbeats(String stormId,
                           Map<String,Object> topoConf)

Specified by:

setupHeatbeats in interface IStormClusterState

teardownHeartbeats

public void teardownHeartbeats(String stormId)

Specified by:

teardownHeartbeats in interface IStormClusterState

teardownTopologyErrors

public void teardownTopologyErrors(String stormId)

Specified by:

teardownTopologyErrors in interface IStormClusterState

getLeader

public NimbusInfo getLeader(Runnable callback)

Description copied from interface: IStormClusterState

Get leader info from state store, which was written when a master gains leadership.

Caution: it can not be used for fencing and is only for informational purposes because we use ZK as our backend now, which could have a overdue info of nodes.

Specified by:

getLeader in interface IStormClusterState

Parameters:

callback - callback func

Returns:

NimbusInfo

backpressureTopologies

public List<String> backpressureTopologies()

Description copied from interface: IStormClusterState

Get backpressure topologies.

Specified by:

backpressureTopologies in interface IStormClusterState

heartbeatStorms

public List<String> heartbeatStorms()

Specified by:

heartbeatStorms in interface IStormClusterState

errorTopologies

public List<String> errorTopologies()

Specified by:

errorTopologies in interface IStormClusterState

setTopologyLogConfig

public void setTopologyLogConfig(String stormId,
                                 LogConfig logConfig,
                                 Map<String,Object> topoConf)

Specified by:

setTopologyLogConfig in interface IStormClusterState

topologyLogConfig

public LogConfig topologyLogConfig(String stormId,
                                   Runnable cb)

Specified by:

topologyLogConfig in interface IStormClusterState

workerHeartbeat

public void workerHeartbeat(String stormId,
                            String node,
                            Long port,
                            ClusterWorkerHeartbeat info)

Specified by:

workerHeartbeat in interface IStormClusterState

removeWorkerHeartbeat

public void removeWorkerHeartbeat(String stormId,
                                  String node,
                                  Long port)

Specified by:

removeWorkerHeartbeat in interface IStormClusterState

supervisorHeartbeat

public void supervisorHeartbeat(String supervisorId,
                                SupervisorInfo info)

Specified by:

supervisorHeartbeat in interface IStormClusterState

topologyBackpressure

public boolean topologyBackpressure(String stormId,
                                    long timeoutMs,
                                    Runnable callback)

Check whether a topology is in throttle-on status or not: if the backpresure/storm-id dir is not empty, this topology has throttle-on, otherwise throttle-off. But if the backpresure/storm-id dir is not empty and has not been updated for more than timeoutMs, we treat it as throttle-off. This will prevent the spouts from getting stuck indefinitely if something wrong happens.

Specified by:

topologyBackpressure in interface IStormClusterState

Parameters:

stormId - The topology Id

timeoutMs - How long until the backpressure znode is invalid.

callback - The callback function

Returns:

True is backpresure/storm-id dir is not empty and at least one of the backpressure znodes has not timed out; false otherwise.

setupBackpressure

public void setupBackpressure(String stormId,
                              Map<String,Object> topoConf)

Description copied from interface: IStormClusterState

Setup backpressure.

Specified by:

setupBackpressure in interface IStormClusterState

removeBackpressure

public void removeBackpressure(String stormId)

Description copied from interface: IStormClusterState

Remove backpressure.

Specified by:

removeBackpressure in interface IStormClusterState

removeWorkerBackpressure

public void removeWorkerBackpressure(String stormId,
                                     String node,
                                     Long port)

Description copied from interface: IStormClusterState

Remove worker backpressure.

Specified by:

removeWorkerBackpressure in interface IStormClusterState

activateStorm

public void activateStorm(String stormId,
                          StormBase stormBase,
                          Map<String,Object> topoConf)

Specified by:

activateStorm in interface IStormClusterState

updateStorm

public void updateStorm(String stormId,
                        StormBase newElems)

To update this function due to APersistentMap/APersistentSet is clojure’s structure.

Specified by:

updateStorm in interface IStormClusterState

removeStormBase

public void removeStormBase(String stormId)

Specified by:

removeStormBase in interface IStormClusterState

setAssignment

public void setAssignment(String stormId,
                          Assignment info,
                          Map<String,Object> topoConf)

Specified by:

setAssignment in interface IStormClusterState

setupBlob

public void setupBlob(String key,
                      NimbusInfo nimbusInfo,
                      Integer versionInfo)

Specified by:

setupBlob in interface IStormClusterState

activeKeys

public List<String> activeKeys()

Specified by:

activeKeys in interface IStormClusterState

blobstore

public List<String> blobstore(Runnable callback)

Specified by:

blobstore in interface IStormClusterState

removeStorm

public void removeStorm(String stormId)

Specified by:

removeStorm in interface IStormClusterState

removeBlobstoreKey

public void removeBlobstoreKey(String blobKey)

Specified by:

removeBlobstoreKey in interface IStormClusterState

removeKeyVersion

public void removeKeyVersion(String blobKey)

Specified by:

removeKeyVersion in interface IStormClusterState

setupErrors

public void setupErrors(String stormId,
                        Map<String,Object> topoConf)

Specified by:

setupErrors in interface IStormClusterState

reportError

public void reportError(String stormId,
                        String componentId,
                        String node,
                        Long port,
                        Throwable error)

Specified by:

reportError in interface IStormClusterState

errors

public List<ErrorInfo> errors(String stormId,
                              String componentId)

Specified by:

errors in interface IStormClusterState

lastError

public ErrorInfo lastError(String stormId,
                           String componentId)

Specified by:

lastError in interface IStormClusterState

setCredentials

public void setCredentials(String stormId,
                           Credentials creds,
                           Map<String,Object> topoConf)

Specified by:

setCredentials in interface IStormClusterState

credentials

public Credentials credentials(String stormId,
                               Runnable callback)

Specified by:

credentials in interface IStormClusterState

disconnect

public void disconnect()

Specified by:

disconnect in interface IStormClusterState

getPrivateWorkerKey

public PrivateWorkerKey getPrivateWorkerKey(WorkerTokenServiceType type,
                                            String topologyId,
                                            long keyVersion)

Description copied from interface: IStormClusterState

Get a private key used to validate a token is correct. This is expected to be called from a privileged daemon, and the ACLs should be set up to only allow nimbus and these privileged daemons access to these private keys.

Specified by:

getPrivateWorkerKey in interface IStormClusterState

Parameters:

type - the type of service the key is for.

topologyId - the topology id the key is for.

keyVersion - the version of the key this is for.

Returns:

the private key or null if it could not be found.

addPrivateWorkerKey

public void addPrivateWorkerKey(WorkerTokenServiceType type,
                                String topologyId,
                                long keyVersion,
                                PrivateWorkerKey key)

Description copied from interface: IStormClusterState

Store a new version of a private key. This is expected to only ever be called from nimbus. All ACLs however need to be setup to allow the given services access to the stored information.

Specified by:

addPrivateWorkerKey in interface IStormClusterState

Parameters:

type - the type of service this key is for.

topologyId - the topology this key is for

keyVersion - the version of the key this is for.

key - the key to store.

getNextPrivateWorkerKeyVersion

public long getNextPrivateWorkerKeyVersion(WorkerTokenServiceType type,
                                           String topologyId)

Description copied from interface: IStormClusterState

Get the next key version number that should be used for this topology id. This is expected to only ever be called from nimbus, but it is acceptable if the ACLs are setup so that it can work from a privileged daemon for the given service.

Specified by:

getNextPrivateWorkerKeyVersion in interface IStormClusterState

Parameters:

type - the type of service this is for.

topologyId - the topology id this is for.

Returns:

the next version number. It should be 0 for a new topology id/service combination.

removeExpiredPrivateWorkerKeys

public void removeExpiredPrivateWorkerKeys(String topologyId)

Description copied from interface: IStormClusterState

Remove all keys for the given topology that have expired. The number of keys should be small enough that doing an exhaustive scan of them all is acceptable as there is no guarantee that expiration time and version number are related. This should be for all service types. This is expected to only ever be called from nimbus and some ACLs may be setup so being called from other daemons will cause it to fail.

Specified by:

removeExpiredPrivateWorkerKeys in interface IStormClusterState

Parameters:

topologyId - the id of the topology to scan.

removeAllPrivateWorkerKeys

public void removeAllPrivateWorkerKeys(String topologyId)

Description copied from interface: IStormClusterState

Remove all of the worker keys for a given topology. Used to clean up after a topology finishes. This is expected to only ever be called from nimbus and ideally should only ever work from nimbus.

Specified by:

removeAllPrivateWorkerKeys in interface IStormClusterState

Parameters:

topologyId - the topology to clean up after.

idsOfTopologiesWithPrivateWorkerKeys

public Set<String> idsOfTopologiesWithPrivateWorkerKeys()

Description copied from interface: IStormClusterState

Get a list of all topologyIds that currently have private worker keys stored, of any kind. This is expected to only ever be called from nimbus.

Specified by:

idsOfTopologiesWithPrivateWorkerKeys in interface IStormClusterState

Returns:

the list of topology ids with any kind of private worker key stored.