[SPARK-35011][CORE] Avoid Block Manager registrations when StopExecutor msg is in-flight

[sumeetgajjar]

What changes were proposed in this pull request?

This patch proposes a fix to prevent triggering BlockManager reregistration while StopExecutor msg is in-flight.
Here on receiving StopExecutor msg, we do not remove the corresponding BlockManagerInfo from blockManagerInfo map, instead we mark it as dead by updating the corresponding executorRemovalTs. There's a separate cleanup thread running to periodically remove the stale BlockManagerInfo from blockManangerInfo map.

Now if a recently removed BlockManager tries to register, the driver simply ignores it since the blockManagerInfo map already contains an entry for it. The same applies to BlockManagerHeartbeat, if the BlockManager belongs to a recently removed executor, the blockManagerInfo map would contain an entry and we shall not ask the corresponding BlockManager to re-register.

Why are the changes needed?

This changes are needed since BlockManager reregistration while executor is shutting down causes inconsistent bookkeeping of executors in Spark.
Consider the following scenario:

• CoarseGrainedSchedulerBackend issues async StopExecutor on executorEndpoint
• CoarseGrainedSchedulerBackend removes that executor from Driver's internal data structures and publishes SparkListenerExecutorRemoved on the listenerBus.
• Executor has still not processed StopExecutor from the Driver
• Driver receives heartbeat from the Executor, since it cannot find the executorId in its data structures, it responds with HeartbeatResponse(reregisterBlockManager = true)
• BlockManager on the Executor reregisters with the BlockManagerMaster and SparkListenerBlockManagerAdded is published on the listenerBus
• Executor starts processing the StopExecutor and exits
• AppStatusListener picks the SparkListenerBlockManagerAdded event and updates AppStatusStore
• statusTracker.getExecutorInfos refers AppStatusStore to get the list of executors which returns the dead executor as alive.

Does this PR introduce any user-facing change?

No

How was this patch tested?

• Modified the existing unittests.
• Ran a simple test application on minikube that asserts on number of executors are zero once the executor idle timeout is reached.

[sumeetgajjar]

It turns out "on-disk storage" (encryptionTest) under BlockManagerSuite fails when "encryption = on" on my mac. But Intellij does not report it as a failure since the jvm simply exited. It also does not run the next set of tests under BlockManagerSuite given such scenarios and hence I didn't notice NPE for two of those unrunned tests.

However, the same encryption test passes on Github. JVM exits while dynamically loading org.apache.commons.crypto.random.OpenSslCryptoRandom using commons-crypto on my machine.

Lesson learnt: always perform a final sanity test run using sbt/mvn before concluding the test passes 😋

[sumeetgajjar]

It turns out "on-disk storage" (encryptionTest) under BlockManagerSuite fails when "encryption = on" on my mac. But Intellij does not report it as a failure since the jvm simply exited. It also does not run the next set of tests under BlockManagerSuite given such scenarios and hence I didn't notice NPE for two of those unrunned tests.

However, the same encryption test passes on Github. JVM exits while dynamically loading org.apache.commons.crypto.random.OpenSslCryptoRandom using commons-crypto on my machine.

Lesson learned: always perform a final sanity test run using sbt/mvn before concluding the test passes 😋

I had [email protected] installed on my mac (Catalina 10.15.7), however, the corresponding shared libs libcrypto.1.1.dylib and libssl.1.1.dylib were missing from my /usr/local/lib dir.
Running the following commands solved the issue and I was able to run the encryptionTest.

cd /usr/local/Cellar/[email protected]/1.1.1k/lib
cp libssl.1.1.dylib libcrypto.1.1.dylib /usr/local/lib
cd /usr/local/lib
ln -s libcrypto.1.1.dylib libcrypto.dylib
ln -s libssl.1.1.dylib libssl.dylib 

You could check if the commons-crypto is using the correct openssl shared libs by running the following command

java -jar Desktop/commons-crypto-1.1.0/commons-crypto-1.1.0.jar Crypto
Apache Commons Crypto 1.1.0
Native code loaded OK: 1.1.0
Native name: Apache Commons Crypto
Native built: Aug 28 2020
OpenSSL library loaded OK, version: 0x101010bf
OpenSSL library info: OpenSSL 1.1.1k  25 Mar 2021
Random instance created OK: org.apache.commons.crypto.random.OpenSslCryptoRandom@2a84aee7
Cipher AES/CTR/NoPadding instance created OK: org.apache.commons.crypto.cipher.OpenSslCipher@1fb3ebeb
Additional OpenSSL_version(n) details:
1: compiler: clang -fPIC -arch x86_64 -O3 -Wall -DL_ENDIAN -DOPENSSL_PIC -DOPENSSL_CPUID_OBJ -DOPENSSL_IA32_SSE2 -DOPENSSL_BN_ASM_MONT -DOPENSSL_BN_ASM_MONT5 -DOPENSSL_BN_ASM_GF2m -DSHA1_ASM -DSHA256_ASM -DSHA512_ASM -DKECCAK1600_ASM -DRC4_ASM -DMD5_ASM -DAESNI_ASM -DVPAES_ASM -DGHASH_ASM -DECP_NISTZ256_ASM -DX25519_ASM -DPOLY1305_ASM -D_REENTRANT -DNDEBUG
2: built on: Thu Mar 25 21:01:02 2021 UTC
3: platform: darwin64-x86_64-cc
4: OPENSSLDIR: "/usr/local/etc/[email protected]"
5: ENGINESDIR: "/usr/local/Cellar/[email protected]/1.1.1k/lib/engines-1.1"

[sumeetgajjar]

I just realized I can now re-run the checks in my personal fork, instead of pushing empty commits.

[HyukjinKwon]

cc @Ngone51 FYI

[attilapiros]

I fixed a typo in the title and description: registerations => registrations.
But I will review this properly only on next week.

[mridulm]

Any particular reason for such a high cache size ?
Also, expire after some time ?

[sumeetgajjar]

The high cache size is to ensure the fix works for a large enough job with 30000 executors.

Sure, does an expiry of 10min (or larger) sounds good?
This should give the executor long enough to process StopExecutor (in-flight) message and complete the shutdown.

[mridulm]

The timeout should be modeled based on what is the max expected delay for heartbeat to come in from executor.

[attilapiros]

I think another possible solution is to extend the BlockManagerInfo with the timestamp of the removing. So modelling the removing as a new state and this way we could avoid using this separate cache completely and all the bm related data would be in the same place.

Of course in this case you should implement the cleanup.

For example it could be just a simple Long var which is 0 by default which means the BlockManager is alive/active (this special value can be hidden behind a method of BlockMangerInfo like isAlive(currentTs)). The cleanup would triggered for delay plus some extra time to avoid too frequent iteration on the blockManagerInfo collection.

WDYT?

[attilapiros]

I thought about this a bit more but haven't checked the code yet: Is it possible to separate driver commanded intentionally removed executors from unintentional executor loss?

Intentionally removed executors shouldn't be re-registered.

cc @Ngone51

[sumeetgajjar]

When CoarseGrainedSchedulerBackend receives RemoveExecutor, it has the ExecutorLossReason. However, after processing the message, when it publishes SparkListenerExecutorRemoved, the reason is passed in form of a string.

[Ngone51]

Currently, on RemoveExecutor, we remove the corresponding BlockManagerInfo from blockManagerInfo map...

As per #32114 (comment), I don't think there would be a BlockManagerMessages.RemoveExecutor raised in this PR case.

Could you point out on which code path that the BlockManagerMessages.RemoveExecutor is raised?

If there's no more code path raises the BlockManagerMessages.RemoveExecutor in this PR case, then @attilapiros definitely works. But, I'd also suggest another idea in #32114 (comment).

[Ngone51]

Is it possible to separate driver commanded intentionally removed executors from unintentional executor loss?

@attilapiros It's possible to know that an executor is removed intentionally by the driver or not. The problem is, currently, the executor info is stored in many different places. So you have to update many methods or messages to add the isIntentional filed (for example), to let all the components know and make certain decisions on it, which could be miscellaneous.

[sumeetgajjar]

Yes, @Ngone51 as pointed out in #32114, BlockManagerMessages.RemoveExecutor is raised in limited cases. I could not find any more code path apart from the one you pointed.

Redacting this due to new findings.

[sumeetgajjar]

I found a code path which will break @attilapiros's proposed solution since the following holds true.

However, we will have to abstract blockManagerInfo: mutable.Map[BlockManagerId, BlockManagerInfo].
Currently, ...

Let's consider the following set of events:

• A CoarseGrainedClusterMessage.RemoveExecutor is issued
• CoarseGrainedSchedulerBackend issues async StopExecutor on executorEndpoint and then invokes executorLost on TaskSchedulerImpl
  

  spark/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala

  Line 430 in e609395

  scheduler.executorLost(executorId, lossReason)

• TaskSchedulerImpl in its executorLost invokes dagScheduler.executorLost
  

  spark/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala

  Lines 998 to 1001 in e609395

  	if (failedExecutor.isDefined) {
  	dagScheduler.executorLost(failedExecutor.get, reason)
  	backend.reviveOffers()
  	}

• DAGScheduler while handling executorLost invokes removeExecutorAndUnregisterOutputs which internally invokes blockManagerMaster.removeExecutor(execId) (as you pointed out in your comment below) which further clears blockManagerId from blockManagerInfo in BlockManagerMasterEndpoint
  

  spark/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

  Line 2043 in e609395

  removeExecutorAndUnregisterOutputs(

• The Executor has not yet processed StopExecutor
• Executor reports its Heartbeat
• HeartbeatReceiver invokes scheduler.executorHeartbeatReceived to check if the BlockManager on the executor requires re-registration
  

  spark/core/src/main/scala/org/apache/spark/HeartbeatReceiver.scala

  Line 137 in e609395

  val unknownExecutor = !scheduler.executorHeartbeatReceived(

• TashSchedulerImpl delegates this to DAGScheduler
  

  spark/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala

  Line 858 in e609395

  dagScheduler.executorHeartbeatReceived(execId, accumUpdatesWithTaskIds, blockManagerId,

• DAGScheduler asks BlockManagerMasterHeartbeatEndpoint if it knows the BlockManager
  

  spark/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

  Line 300 in e609395

  blockManagerMaster.driverHeartbeatEndPoint.askSync[Boolean](

• BlockManagerMasterHeartbeatEndpoint returns false since it cannot find blockManagerId in BlockManagerInfo indicating the blockManager should re-register
  

  spark/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterHeartbeatEndpoint.scala

  Line 51 in e609395

  if (!blockManagerInfo.contains(blockManagerId)) {

• BlockManager re-registers which publishes the SparkListenerBlockManagerAdded causing the inconsistent book-keeping in AppStatusStore
• Executor processes StopExecutor and exits.

[mridulm]

Move the cache into BlockManagerMasterEndpoint and hide the impl detail ?
We can have a cleaner interface here ... BlockManagerMasterHeartbeatEndpoint simply needs a way to validate if an executor was recently removed - does not need to know if it was a Cache/Set/etc.

[sumeetgajjar]

Sure.
The same comment applies to already present blockManagerInfo map as well.
I could refactor that as well. I was thinking of creating a BlockMangerEndpointSharedState class which contains blockManagerInfo and recentlyRemovedExecutors. The class would expose corresponding methods for lookup and updates.

[attilapiros]

Nit: for just checking whether an object is null you do not need to build an Option instance (Option is good when when you pass the object to another method to emphasize it can be null).

[attilapiros]

Nit: same here (Option is not needed)

[Ngone51]

Hey guys, I'd like to propose a simpler (but might be a little bit ticky) fix if I understand the issue correctly. The idea is,

Instead of removing the executor directly, we set executorLastSeen(executorId) = -1L when we receives ExecutorRemoved in HeartbeatReceiver. And then,

1. if Heartbeat comes first before ExpireDeadHosts, we remove the executor from executorLastSeen by checking the value "< 0" and avoid the re-register.

2. if ExpireDeadHosts comes first before Heartbeat, we set executorLastSeen(executorId) = -2L. We can't remove it this time in ExpireDeadHosts because if Heartbeat comes later we'd have the same issue again.

2.1 if Heartbeat comes later, we remove the executor from executorLastSeen by checking the value "< 0" too and also avoid the re-register.

2.2 if Heartbeat doesn't come (that means the executor stopped before sending the heartbeat), we remove the executor from executorLastSeen by checking the value = -2L in next ExpireDeadHosts.

In this way, we can avoid the extra cache and all changes should be limited to HeartbeatReceiver.

Any thoughts?

[mridulm]

In essence, if I understood correctly, we are adding a lostExecutorCandidates:Map[String, ExpirationState] ?

• If we detect a request to expire an executor comes in - then expire based on (some) policy : timeout since initial expiry/number of expirations/other reasons : else add/update expiration state of candidate.
• If heartbeat comes in, then remove from candidate set.
• If explicit remove, then remove from both executorLastSeen and lostExecutorCandidates:Set.

Did I miss anything ? I am fine with this approach.
(I explicitly pulled out magic values out for explanation clarity)

[sumeetgajjar]

Hey guys, I'd like to propose a simpler (but might be a little bit ticky) fix if I understand the issue correctly. The idea is,

Instead of removing the executor directly, we set executorLastSeen(executorId) = -1L when we receives ExecutorRemoved in HeartbeatReceiver. And then,

1. if `Heartbeat` comes first before `ExpireDeadHosts`, we remove the executor from `executorLastSeen` by checking the value "< 0" and avoid the re-register.

2. if `ExpireDeadHosts` comes first before `Heartbeat`,  we set `executorLastSeen(executorId) = -2L`. We can't remove it this time in `ExpireDeadHosts` because if `Heartbeat` comes later we'd have the same issue again.

2.1 if Heartbeat comes later, we remove the executor from executorLastSeen by checking the value "< 0" too and also avoid the re-register.

2.2 if Heartbeat doesn't come (that means the executor stopped before sending the heartbeat), we remove the executor from executorLastSeen by checking the value = -2L in next ExpireDeadHosts.

In this way, we can avoid the extra cache and all changes should be limited to HeartbeatReceiver.

Any thoughts?

Thanks for the comment @Ngone51 .
There are two places from where the re-registration can be triggered

• From HeartbeatReceiver - by responding HeartbeatResponse(reregisterBlockManager = true).
  Your solution will take care of this.
• From BlockManager - e.g. reportBlockStatus.
  Just modifying HeartbeatReceiver won't solve the re-registration issue here. We will also have to implement a similar kind of tracking inside BlockManagerMasterEndpoint. And now since both tracking are independent of each other, it might introduce some race condition (please correct me if I am wrong).

[sumeetgajjar]

In essence, if I understood correctly, we are adding a lostExecutorCandidates:Map[String, ExpirationState] ?

* If we detect a request to expire an executor comes in - then expire based on (some) policy : timeout since initial expiry/number of expirations/other reasons : else add/update expiration state of candidate.

* If heartbeat comes in, then remove from candidate set.

* If explicit remove, then remove from both `executorLastSeen` and `lostExecutorCandidates:Set`.

Did I miss anything ? I am fine with this approach.
(I explicitly pulled out magic values out for explanation clarity)

Thanks for the comment @mridulm .
I believe this comment applies here as well.

I believe @attilapiros suggestion would take care of both cases where re-registration is trigger without introducing another Cache of recentlyRemovedExecutors.

[mridulm]

I am getting a little confused between PR description and the subsequent discussion.
What exactly is the behavior we are trying to converge towards/address ?

An expiration of executor from heartbeat master not only sends a StopExecutor to voluntarily get executor to exit, but also gets the cluster manager to force termination (in case of MIA/hung executor). So in steady state, once transitionary/overlapping updates are done, the executor should be gone according to driver.

My understanding was, there is a race here between cluster manager notifying application (after killing executor) and the executor heartbeat/blockmanager re-registration : which ends up causing a dead executor to be marked live indefinitely.

Is this the only case we are addressing ? Or are there any other paths that are impacted ?

(@Ngone51 Not sure if standalone has nuances that I am missing here).

[Ngone51]

Standalone should be the same. @mridulm

[Ngone51]

From BlockManager - e.g. reportBlockStatus.
Just modifying HeartbeatReceiver won't solve the re-registration issue here. We will also have to implement a similar kind of tracking inside BlockManagerMasterEndpoint. And now since both tracking are independent of each other, it might introduce some race condition (please correct me if I am wrong).

You're right. I followed the PR description only so I thought HeartbeatReceiver is the only problematic place.

I checked the code and surprisingly find that we don't remove BlockManager when we remove an executor. And removing BlockManager happens in few cases only,

• the corresponding executor of the BlockManager caused the shuffle fetch failure
  

  spark/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

  Line 2081 in ee7d838

  blockManagerMaster.removeExecutor(execId)

• an executor is removed redundantly
  

  spark/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala

  Line 434 in ee7d838

  scheduler.sc.env.blockManager.master.removeExecutorAsync(executorId)

• a new registered BlockManager evicts an old one (if any)
  

  spark/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala

  Lines 534 to 537 in ee7d838

  	// A block manager of the same executor already exists, so remove it (assumed dead)
  	logError("Got two different block manager registrations on same executor - "
  	+ s" will replace old one $oldId with new one $id")
  	removeExecutor(id.executorId)

If that's the case (it seems not correct but exits for a long time already), I think posting the SparkListenerBlockManagerAdded inside the if (!blockManagerInfo.contains(id)) would be enough for the whole fix?

spark/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala

Lines 531 to 562 in ee7d838

	if (!blockManagerInfo.contains(id)) {
	blockManagerIdByExecutor.get(id.executorId) match {
	case Some(oldId) =>
	// A block manager of the same executor already exists, so remove it (assumed dead)
	logError("Got two different block manager registrations on same executor - "
	+ s" will replace old one $oldId with new one $id")
	removeExecutor(id.executorId)
	case None =>
	}
	logInfo("Registering block manager %s with %s RAM, %s".format(
	id.hostPort, Utils.bytesToString(maxOnHeapMemSize + maxOffHeapMemSize), id))
	blockManagerIdByExecutor(id.executorId) = id
	val externalShuffleServiceBlockStatus =
	if (externalShuffleServiceRddFetchEnabled) {
	val externalShuffleServiceBlocks = blockStatusByShuffleService
	.getOrElseUpdate(externalShuffleServiceIdOnHost(id), new JHashMap[BlockId, BlockStatus])
	Some(externalShuffleServiceBlocks)
	} else {
	None
	}
	blockManagerInfo(id) = new BlockManagerInfo(id, System.currentTimeMillis(),
	maxOnHeapMemSize, maxOffHeapMemSize, storageEndpoint, externalShuffleServiceBlockStatus)
	if (pushBasedShuffleEnabled) {
	addMergerLocation(id)
	}
	}
	listenerBus.post(SparkListenerBlockManagerAdded(time, id, maxOnHeapMemSize + maxOffHeapMemSize,
	Some(maxOnHeapMemSize), Some(maxOffHeapMemSize)))

[sumeetgajjar]

If that's the case (it seems not correct but exits for a long time already), I think posting the SparkListenerBlockManagerAdded inside the if (!blockManagerInfo.contains(id)) would be enough for the whole fix?

@Ngone51 I believe moving SparkListenerBlockManagerAdded inside the if-loop should be enough.
I will give it a try and check if any other cases surface.

Thanks!

[sumeetgajjar]

I was wondering what is policy for merging the latest changes to a upstream-dev-branch?
For e.g. I have a branch SPARK-35011 which contains my changes, it is behind upstream/master

• Should I be rebasing my commits on the new changes from upstream/master ?

  • current branch is SPARK-35011 -- git pull upstream master --rebase
  • In this case when I push my changes to PR, I will require a --force push since the commit ordering is now changed.

• Or should I merge the upstream/master into SPARK-35011?

  • current branch is SPARK-35011 -- git merge upstream/master
  • No force push required, however my commits are too far behind in the history.

[sumeetgajjar]

If that's the case (it seems not correct but exits for a long time already), I think posting the SparkListenerBlockManagerAdded inside the if (!blockManagerInfo.contains(id)) would be enough for the whole fix?

@Ngone51 I believe moving SparkListenerBlockManagerAdded inside the if-loop should be enough.
I will give it a try and check if any other cases surface.

Thanks!

I tried this and the issue still exists. When we apply the sequence of events mentioned in #32114, the issue surfaces again.

[Ngone51]

Should I be rebasing my commits on the new changes from upstream/master ?

Rebase is recommended whenever it's possible. @sumeetgajjar

[Ngone51]

I tried this and the issue still exists. When we apply the sequence of events mentioned in #32114, the issue surfaces again.

I see. I think I missed the code path of scheduler.executorLost(executorId, lossReason).

Thanks for the experiment.

[Ngone51]

So I think the solution now would be:

For the heartbeat, using #32114 (review).

For theBlokmanager, adding a BlockManagerEndpointSharedState (as mentioned by @sumeetgajjar in #32114 (comment)) for both BlockManagerMasterEndpoint and BlockManagerMasterHeartbeatEndpoint. It's true that if we only adding a removal state to the blockManagerInfo, we have to filter the removed blockmanagers first before traversing it (e.g., we won't expect to return a removed blockmanager in getPeers).

In BlockManagerEndpointSharedState, we'd have both activeBlockManagerInfo and the removedBlockManagerInfo. We don't have to set up a new cleaner to clear the removedBlockManagerInfo. Instead, we can reuse the fix of HeartbeatReceiver as whenever there's a sure removal in HeartbeatReceiver, we can send a removal request to BlockManagerEndpointSharedState as well by following the code path of !scheduler.executorHeartbeatReceived(e.g., we could have scheduler.clearBlockManagerInfo similarly).

WDYT?

[sumeetgajjar]

So I think the solution now would be:

For the heartbeat, using #32114 (review).

For theBlokmanager, adding a BlockManagerEndpointSharedState (as mentioned by @sumeetgajjar in #32114 (comment)) for both BlockManagerMasterEndpoint and BlockManagerMasterHeartbeatEndpoint. It's true that if we only adding a removal state to the blockManagerInfo, we have to filter the removed blockmanagers first before traversing it (e.g., we won't expect to return a removed blockmanager in getPeers).

In BlockManagerEndpointSharedState, we'd have both activeBlockManagerInfo and the removedBlockManagerInfo. We don't have to set up a new cleaner to clear the removedBlockManagerInfo. Instead, we can reuse the fix of HeartbeatReceiver as whenever there's a sure removal in HeartbeatReceiver, we can send a removal request to BlockManagerEndpointSharedState as well by following the code path of !scheduler.executorHeartbeatReceived(e.g., we could have scheduler.clearBlockManagerInfo similarly).

WDYT?

@Ngone51 Thank you for this suggestion, I understand the solution, however, I believe this might be slight complex to keep track of things since the cleanup/removal is triggered from a different component i.e. HeartbeatReceiver.

I spoke to @attilapiros offline regarding his solution and it seems he missed to mention one thing that BlockManagerInfo will not be removed here:

spark/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala

Line 344 in 1b609c7

blockManagerInfo.remove(blockManagerId)

Insterad, the cleanup thread would take care of removal. This will keep the whole logic in the same component i.e. BlockManagerMasterEndpoint and would be easy to track from a code understanding point of view.

I will proceed with @attilapiros proposal where we model BlockManager removal as a new state, run some tests and update more.

[Ngone51]

I just realized this bug does cause the real problem when working in conjunction with #24533. Basically, the re-registration issue leads to the driver thinks an executor is alive while it's actually dead, which in turn causes the client to retry the block on the dead executor, while it shouldn't. Could you @sumeetgajjar backport this fix to 3.1/3.0 as well?
cc @mridulm @attilapiros

[sumeetgajjar]

I just realized this bug does cause the real problem when working in conjunction with #24533. Basically, the re-registration issue leads to the driver thinks an executor is alive while it's actually dead, which in turn causes the client to retry the block on the dead executor, while it shouldn't. Could you @sumeetgajjar backport this fix to 3.1/3.0 as well?
cc @mridulm @attilapiros

@Ngone51, sure I will backport it to 3.1 and 3.0 as well.

[sumeetgajjar]

SPARK-34949 should also be backported to close any gaps.

P.S. It is already in 3.1 we just need to backport it to 3.0