HyukjinKwon commented on code in PR #59144:
URL: https://github.com/apache/spark/pull/59144#discussion_r4176604179
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStore.scala:
##########
@@ -1625,13 +1628,69 @@ object StateStore extends Logging {
if (version < 0) {
throw QueryExecutionErrors.unexpectedStateStoreVersion(version)
}
- val (storeProvider, shouldForceSnapshotUpload) =
getStateStoreProvider(storeProviderId,
- keySchema, valueSchema, keyStateEncoderSpec, useColumnFamilies,
storeConf, hadoopConf,
- useMultipleValuesPerKey, stateSchemaBroadcast)
- storeProvider.getStore(version, stateStoreCkptId,
shouldForceSnapshotUpload)
+ withClosedProviderRetry(storeProviderId) { createUnpublishedProvider =>
+ getStateStoreProvider(storeProviderId,
+ keySchema, valueSchema, keyStateEncoderSpec, useColumnFamilies,
storeConf, hadoopConf,
+ useMultipleValuesPerKey, stateSchemaBroadcast,
createUnpublishedProvider)
+ } { (storeProvider, shouldForceSnapshotUpload) =>
+ storeProvider.getStore(version, stateStoreCkptId,
shouldForceSnapshotUpload)
+ }
}
// scalastyle:on
+ private def withClosedProviderRetry[T](
+ storeProviderId: StateStoreProviderId)(
+ getProvider: Boolean => (StateStoreProvider, Boolean))(
+ loadStore: (StateStoreProvider, Boolean) => T): T = {
+ val (firstProvider, firstShouldForceSnapshotUpload) = getProvider(false)
+ try {
+ loadStore(firstProvider, firstShouldForceSnapshotUpload)
+ } catch {
+ case error: StateStoreInvalidStateMachineTransition
+ if isClosedProviderLoadTransition(error) =>
+ loadedProviders.synchronized {
+ if (loadedProviders.get(storeProviderId).exists(_ eq firstProvider))
{
+ loadedProviders.remove(storeProviderId)
+ } else {
+ None
+ }
+ }.foreach { provider =>
+ unloadedProvidersToClose.add(
+ (storeProviderId, provider, MaintenanceOpRequest.All))
+ }
+ // Create the replacement without publishing it in loadedProviders yet.
+ val (replacementProvider, replacementShouldForceSnapshotUpload) =
getProvider(true)
Review Comment:
Agreed, this is fine as is -- thanks for the detailed explanation. You're
right that loadedProviders is not the sole holder of a provider for an id even
today: the ownership-move path removes providers into unloadedProvidersToClose
while tasks may still hold their stores, and close() blocks until the active
store is released, so the transient second provider the retry can create is
consistent with the existing lifecycle rather than a new invariant. With
independent native RocksDB directories and first-writer-wins (or checkpoint-v2
unique-id reporting) on the shared state-file commit, two instances committing
the same version don't corrupt state. Publishing before acquisition would
reopen the symmetric race you describe, so routing through the cache isn't
clearly safer. Withdrawing this concern.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4140900374","thread_id":"inline:4140900374","verdict_sha256":"1c0bc543708058bf9bf8df9fbcfa699f5f085bc7a5c67bf3ef70d1f98bbd9457"}
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
