StateStoreCoordinatorRef¶

StateStoreCoordinatorRef is used to (let the tasks on Spark executors to) send <> to the <> (that lives on the driver).

[[creating-instance]] [[rpcEndpointRef]] StateStoreCoordinatorRef is given the RpcEndpointRef to the StateStoreCoordinator RPC endpoint when created.

StateStoreCoordinatorRef is <> through StateStoreCoordinatorRef helper object when requested to create one for the <> (when StreamingQueryManager is created) or an <> (when StateStore helper object is requested for the RPC endpoint reference to StateStoreCoordinator for Executors).

[[messages]] .StateStoreCoordinatorRef's Methods and Underlying RPC Messages [width="100%",cols="1m,3",options="header"] |=== | Method | Description

| deactivateInstances a| [[deactivateInstances]]

[source, scala]¶

deactivateInstances(runId: UUID): Unit¶

Requests the RpcEndpointRef to send a DeactivateInstances synchronous message with the given runId and waits for a true / false response

Used exclusively when StreamingQueryManager is requested to handle termination of a streaming query (when StreamExecution is requested to run a streaming query and the query has finished (running streaming batches)).

| getLocation a| [[getLocation]]

[source, scala]¶

getLocation( stateStoreProviderId: StateStoreProviderId): Option[String]

Requests the RpcEndpointRef to send a GetLocation synchronous message with the given StateStoreProviderId and waits for the location

Used when:

• StateStoreAwareZipPartitionsRDD is requested for the preferred locations of a partition (when StreamingSymmetricHashJoinExec physical operator is executed

• StateStoreRDD is requested for preferred locations for a task for a partition

| reportActiveInstance a| [[reportActiveInstance]]

[source, scala]¶

reportActiveInstance( stateStoreProviderId: StateStoreProviderId, host: String, executorId: String): Unit

Requests the RpcEndpointRef to send a ReportActiveInstance one-way asynchronous (fire-and-forget) message with the given StateStoreProviderId, host and executorId

Used when StateStore utility is requested for reportActiveStoreInstance (when StateStore utility is requested to look up the StateStore by StateStoreProviderId)

| stop a| [[stop]]

[source, scala]¶

stop(): Unit¶

Requests the RpcEndpointRef to send a StopCoordinator synchronous message

Used exclusively for unit testing

| verifyIfInstanceActive a| [[verifyIfInstanceActive]]

[source, scala]¶

verifyIfInstanceActive( stateStoreProviderId: StateStoreProviderId, executorId: String): Boolean

Requests the RpcEndpointRef to send a VerifyIfInstanceActive synchronous message with the given StateStoreProviderId and executorId, and waits for a true / false response

Used when StateStore utility is requested for verifyIfStoreInstanceActive (when requested to doMaintenance from a running MaintenanceTask daemon thread)

|===

Creating StateStoreCoordinatorRef to StateStoreCoordinator RPC Endpoint for Driver¶

forDriver(
  env: SparkEnv): StateStoreCoordinatorRef

forDriver...FIXME

forDriver is used when StreamingQueryManager is created.

Creating StateStoreCoordinatorRef to StateStoreCoordinator RPC Endpoint for Executor¶

forExecutor(
  env: SparkEnv): StateStoreCoordinatorRef

forExecutor...FIXME

forExecutor is used when StateStore utility is requested for the RPC endpoint reference to StateStoreCoordinator for Executors.