Package org.apache.storm.cluster
Class StormClusterStateImpl
java.lang.Object
org.apache.storm.cluster.StormClusterStateImpl
- All Implemented Interfaces:
IStormClusterState
-
Constructor Summary
ConstructorDescriptionStormClusterStateImpl
(IStateStorage stateStorage, ILocalAssignmentsBackend assignmentsassignmentsBackend, ClusterStateContext context, boolean shouldCloseStateStorageOnDisconnect) -
Method Summary
Modifier and TypeMethodDescriptionvoid
void
addNimbusHost
(String nimbusId, NimbusSummary nimbusSummary) void
addPrivateWorkerKey
(WorkerTokenServiceType type, String topologyId, long keyVersion, PrivateWorkerKey key) Store a new version of a private key.assignmentInfo
(String stormId, Runnable callback) Get the assignment based on storm id from local backend.assignmentInfoWithVersion
(String stormId, Runnable callback) assignments
(Runnable callback) Get all the topologies assignments mapping stormId -> Assignment from local backend.assignmentVersion
(String stormId, Runnable callback) Get backpressure topologies.blobstoreInfo
(String blobKey) credentials
(String stormId, Runnable callback) void
deleteTopologyProfileRequests
(String stormId, ProfileRequest profileRequest) void
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.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.getPrivateWorkerKey
(WorkerTokenServiceType type, String topologyId, long keyVersion) Get a private key used to validate a token is correct.getTopologyProfileRequests
(String stormId) getWorkerHeartbeat
(String stormId, String node, Long port) getWorkerProfileRequests
(String stormId, NodeInfo nodeInfo) Get a list of all topologyIds that currently have private worker keys stored, of any kind.boolean
Flag to indicate if the assignments synced successfully, seeIStormClusterState.syncRemoteAssignments(Map)
.boolean
Flag to indicate if the Pacameker is backend store.protected void
protected void
issueMapCallback
(ConcurrentHashMap<String, Runnable> callbackConcurrentHashMap, String key) nimbuses()
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
void
setAssignment
(String stormId, Assignment info, Map<String, Object> topoConf) void
Mark the assignments as synced successfully, seeIStormClusterState.isAssignmentsBackendSynchronized()
.void
setCredentials
(String stormId, Credentials creds, Map<String, Object> topoConf) void
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) Get a storm base for a topology.Get storm id from passed name, null if the name doesn't exist on cluster.void
supervisorHeartbeat
(String supervisorId, SupervisorInfo info) supervisorInfo
(String supervisorId) supervisors
(Runnable callback) void
syncRemoteAssignments
(Map<String, byte[]> remote) Sync the remote state store assignments to local backend, used when master gains leadership, seeorg.apache.storm.nimbus.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.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 Details
-
StormClusterStateImpl
public StormClusterStateImpl(IStateStorage stateStorage, ILocalAssignmentsBackend assignmentsassignmentsBackend, ClusterStateContext context, boolean shouldCloseStateStorageOnDisconnect) throws Exception - Throws:
Exception
-
-
Method Details
-
issueCallback
-
issueMapCallback
protected void issueMapCallback(ConcurrentHashMap<String, Runnable> callbackConcurrentHashMap, String key) -
assignments
- Specified by:
assignments
in interfaceIStormClusterState
-
assignmentInfo
Description copied from interface:IStormClusterState
Get the assignment based on storm id from local backend.- Specified by:
assignmentInfo
in interfaceIStormClusterState
- Parameters:
stormId
- topology idcallback
- callback function- Returns:
Assignment
-
remoteAssignmentInfo
Description copied from interface:IStormClusterState
Get the assignment based on storm id from remote state store, eg: ZK.- Specified by:
remoteAssignmentInfo
in interfaceIStormClusterState
- Parameters:
stormId
- topology idcallback
- callback function- Returns:
Assignment
-
assignmentsInfo
Description copied from interface:IStormClusterState
Get all the topologies assignments mapping stormId -> Assignment from local backend.- Specified by:
assignmentsInfo
in interfaceIStormClusterState
- Returns:
- stormId -> Assignment mapping
-
syncRemoteAssignments
Description copied from interface:IStormClusterState
Sync the remote state store assignments to local backend, used when master gains leadership, seeorg.apache.storm.nimbus.LeaderListenerCallback
.- Specified by:
syncRemoteAssignments
in interfaceIStormClusterState
- Parameters:
remote
- assigned assignments for a specificIStormClusterState
instance, usually a supervisor/node.
-
isAssignmentsBackendSynchronized
public boolean isAssignmentsBackendSynchronized()Description copied from interface:IStormClusterState
Flag to indicate if the assignments synced successfully, seeIStormClusterState.syncRemoteAssignments(Map)
.- Specified by:
isAssignmentsBackendSynchronized
in interfaceIStormClusterState
- Returns:
- true if is synced successfully
-
isPacemakerStateStore
public boolean isPacemakerStateStore()Description copied from interface:IStormClusterState
Flag to indicate if the Pacameker is backend store.- Specified by:
isPacemakerStateStore
in interfaceIStormClusterState
- Returns:
- true if Pacemaker is being used as StateStore
-
setAssignmentsBackendSynchronized
public void setAssignmentsBackendSynchronized()Description copied from interface:IStormClusterState
Mark the assignments as synced successfully, seeIStormClusterState.isAssignmentsBackendSynchronized()
.- Specified by:
setAssignmentsBackendSynchronized
in interfaceIStormClusterState
-
assignmentInfoWithVersion
- Specified by:
assignmentInfoWithVersion
in interfaceIStormClusterState
-
assignmentVersion
- Specified by:
assignmentVersion
in interfaceIStormClusterState
- Throws:
Exception
-
blobstoreInfo
- Specified by:
blobstoreInfo
in interfaceIStormClusterState
-
nimbuses
- Specified by:
nimbuses
in interfaceIStormClusterState
-
addNimbusHost
- Specified by:
addNimbusHost
in interfaceIStormClusterState
-
activeStorms
- Specified by:
activeStorms
in interfaceIStormClusterState
-
stormBase
Description copied from interface:IStormClusterState
Get a storm base for a topology.- Specified by:
stormBase
in interfaceIStormClusterState
- Parameters:
stormId
- the id of the topologycallback
- something to call if the data changes (best effort)- Returns:
- the StormBase or null if it is not alive.
-
stormId
Description copied from interface:IStormClusterState
Get storm id from passed name, null if the name doesn't exist on cluster.- Specified by:
stormId
in interfaceIStormClusterState
- Parameters:
stormName
- storm name- Returns:
- storm id
-
syncRemoteIds
Description copied from interface:IStormClusterState
Sync all the active storm ids of the cluster, used now when master gains leadership.- Specified by:
syncRemoteIds
in interfaceIStormClusterState
- Parameters:
remote
- stormName -> stormId mapping
-
getWorkerHeartbeat
- Specified by:
getWorkerHeartbeat
in interfaceIStormClusterState
-
getWorkerProfileRequests
- Specified by:
getWorkerProfileRequests
in interfaceIStormClusterState
-
getTopologyProfileRequests
- Specified by:
getTopologyProfileRequests
in interfaceIStormClusterState
-
setWorkerProfileRequest
- Specified by:
setWorkerProfileRequest
in interfaceIStormClusterState
-
deleteTopologyProfileRequests
- Specified by:
deleteTopologyProfileRequests
in interfaceIStormClusterState
-
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 interfaceIStormClusterState
- Parameters:
stormId
- topology idexecutorNodePort
- executor id -> node + port- Returns:
- mapping of executorInfo -> executor beat
-
supervisors
- Specified by:
supervisors
in interfaceIStormClusterState
-
supervisorInfo
- Specified by:
supervisorInfo
in interfaceIStormClusterState
-
setupHeatbeats
- Specified by:
setupHeatbeats
in interfaceIStormClusterState
-
teardownHeartbeats
- Specified by:
teardownHeartbeats
in interfaceIStormClusterState
-
teardownTopologyErrors
- Specified by:
teardownTopologyErrors
in interfaceIStormClusterState
-
getLeader
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 interfaceIStormClusterState
- Parameters:
callback
- callback func- Returns:
NimbusInfo
-
backpressureTopologies
Description copied from interface:IStormClusterState
Get backpressure topologies. Note: In Storm 2.0. Retained for enabling transition from 1.x. Will be removed soon.- Specified by:
backpressureTopologies
in interfaceIStormClusterState
-
heartbeatStorms
- Specified by:
heartbeatStorms
in interfaceIStormClusterState
-
errorTopologies
- Specified by:
errorTopologies
in interfaceIStormClusterState
-
setTopologyLogConfig
- Specified by:
setTopologyLogConfig
in interfaceIStormClusterState
-
topologyLogConfig
- Specified by:
topologyLogConfig
in interfaceIStormClusterState
-
workerHeartbeat
- Specified by:
workerHeartbeat
in interfaceIStormClusterState
-
removeWorkerHeartbeat
- Specified by:
removeWorkerHeartbeat
in interfaceIStormClusterState
-
supervisorHeartbeat
- Specified by:
supervisorHeartbeat
in interfaceIStormClusterState
-
topologyBackpressure
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 interfaceIStormClusterState
- Parameters:
stormId
- The topology IdtimeoutMs
- 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
Description copied from interface:IStormClusterState
Setup backpressure. Note: In Storm 2.0. Retained for enabling transition from 1.x. Will be removed soon.- Specified by:
setupBackpressure
in interfaceIStormClusterState
-
removeBackpressure
Description copied from interface:IStormClusterState
Remove backpressure. Note: In Storm 2.0. Retained for enabling transition from 1.x. Will be removed soon.- Specified by:
removeBackpressure
in interfaceIStormClusterState
-
removeWorkerBackpressure
Description copied from interface:IStormClusterState
Remove worker backpressure. Note: In Storm 2.0. Retained for enabling transition from 1.x. Will be removed soon.- Specified by:
removeWorkerBackpressure
in interfaceIStormClusterState
-
activateStorm
- Specified by:
activateStorm
in interfaceIStormClusterState
-
updateStorm
To update this function due to APersistentMap/APersistentSet is clojure's structure.- Specified by:
updateStorm
in interfaceIStormClusterState
-
removeStormBase
- Specified by:
removeStormBase
in interfaceIStormClusterState
-
setAssignment
- Specified by:
setAssignment
in interfaceIStormClusterState
-
setupBlob
- Specified by:
setupBlob
in interfaceIStormClusterState
-
activeKeys
- Specified by:
activeKeys
in interfaceIStormClusterState
-
blobstore
- Specified by:
blobstore
in interfaceIStormClusterState
-
removeStorm
- Specified by:
removeStorm
in interfaceIStormClusterState
-
removeBlobstoreKey
- Specified by:
removeBlobstoreKey
in interfaceIStormClusterState
-
removeKeyVersion
- Specified by:
removeKeyVersion
in interfaceIStormClusterState
-
setupErrors
- Specified by:
setupErrors
in interfaceIStormClusterState
-
reportError
public void reportError(String stormId, String componentId, String node, Long port, Throwable error) - Specified by:
reportError
in interfaceIStormClusterState
-
errors
- Specified by:
errors
in interfaceIStormClusterState
-
lastError
- Specified by:
lastError
in interfaceIStormClusterState
-
setCredentials
- Specified by:
setCredentials
in interfaceIStormClusterState
-
credentials
- Specified by:
credentials
in interfaceIStormClusterState
-
disconnect
public void disconnect()- Specified by:
disconnect
in interfaceIStormClusterState
-
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 interfaceIStormClusterState
- 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 interfaceIStormClusterState
- Parameters:
type
- the type of service this key is for.topologyId
- the topology this key is forkeyVersion
- the version of the key this is for.key
- the key to store.
-
getNextPrivateWorkerKeyVersion
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 interfaceIStormClusterState
- 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
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 interfaceIStormClusterState
- Parameters:
topologyId
- the id of the topology to scan.
-
removeAllPrivateWorkerKeys
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 interfaceIStormClusterState
- Parameters:
topologyId
- the topology to clean up after.
-
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 interfaceIStormClusterState
- Returns:
- the list of topology ids with any kind of private worker key stored.
-