[SUPPORT]hudi upsert data Caused by: org.apache.hadoop.fs.PathIsNotEmptyDirectoryException

upsert data code, but I check the temp dir is empty.


```java
private void upsertAllAction(Dataset jsonDataSet, long maxUseMemory, String tempPath) {
int dataKeepTime = 3 * 24 * 60 / config.getTriggerTime();
jsonDataSet.write()
.format("org.apache.hudi")
.option(HoodieTableConfig.TYPE.key(), HoodieTableType.MERGE_ON_READ.name())
.option(DataSourceWriteOptions.OPERATION().key(), WriteOperationType.UPSERT.value())
.option(DataSourceWriteOptions.TABLE_TYPE().key(), HoodieTableType.MERGE_ON_READ.name())
.option(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), config.getIdName())
.option(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key(), Constants.DT)
.option(KeyGeneratorOptions.HIVE_STYLE_PARTITIONING_ENABLE.key(), true)
.option(HoodieWriteConfig.PRECOMBINE_FIELD_NAME.key(), Constants.UPDATE_TIME)
.option(HoodieWriteConfig.COMBINE_BEFORE_UPSERT.key(), false)
.option(HoodieWriteConfig.UPSERT_PARALLELISM_VALUE.key(), 200)
.option(HoodieWriteConfig.FINALIZE_WRITE_PARALLELISM_VALUE.key(), 200)
.option(HoodieWriteConfig.WRITE_PAYLOAD_CLASS_NAME.key(), DefaultHoodieRecordPayload.class.getName())
.option(HoodieWriteConfig.AVRO_EXTERNAL_SCHEMA_TRANSFORMATION_ENABLE.key(), true)
.option(HoodieWriteConfig.AVRO_SCHEMA_VALIDATE_ENABLE.key(), true)
.option(HoodieWriteConfig.SCHEMA_ALLOW_AUTO_EVOLUTION_COLUMN_DROP.key(), true)
.option(HoodieCommonConfig.RECONCILE_SCHEMA.key(), true)
.option(HoodieCommonConfig.SCHEMA_EVOLUTION_ENABLE.key(), true)
.option(HoodieWriteConfig.MARKERS_TYPE.key(), MarkerType.DIRECT.toString())
.option(HoodieWriteConfig.COMBINE_BEFORE_INSERT.key(), true)
.option(HoodiePayloadConfig.PAYLOAD_CLASS_NAME.key(), DefaultHoodieRecordPayload.class.getName())
.option(HoodieCleanConfig.CLEANER_COMMITS_RETAINED.key(), dataKeepTime)
.option(HoodieCleanConfig.AUTO_CLEAN.key(), true)
.option(HoodieCleanConfig.CLEANER_INCREMENTAL_MODE_ENABLE.key(), true)
.option(HoodieArchivalConfig.MIN_COMMITS_TO_KEEP.key(), dataKeepTime + 1)
.option(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key(), dataKeepTime * 3)
.option(HoodieCompactionConfig.TARGET_IO_PER_COMPACTION_IN_MB.key(), 500 * 1024)
.option(HoodieCleanConfig.CLEANER_POLICY.key(), HoodieCleaningPolicy.KEEP_LATEST_BY_HOURS.name())
.option(HoodieCleanConfig.CLEANER_HOURS_RETAINED.key(), 72)
.option(HoodieCompactionConfig.PARQUET_SMALL_FILE_LIMIT.key(), 128 * 1024 * 1024)
.option(HoodieStorageConfig.PARQUET_MAX_FILE_SIZE.key(), 256 * 1024 * 1024)
.option(HoodieCompactionConfig.INLINE_COMPACT.key(), true)
.option(HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key(), 0)
.option(HoodieMemoryConfig.SPILLABLE_MAP_BASE_PATH.key(), tempPath)
.option(HoodieMetadataConfig.ENABLE.key(), true)
.option(HoodieMetadataConfig.MIN_COMMITS_TO_KEEP.key(), dataKeepTime + 1)
.option(HoodieMetadataConfig.MAX_COMMITS_TO_KEEP.key(), dataKeepTime + 2)
.option(HoodieMetadataConfig.CLEANER_COMMITS_RETAINED.key(), dataKeepTime)
.option(HoodieMemoryConfig.MAX_MEMORY_FOR_MERGE.key(), maxUseMemory)
.option(HoodieMemoryConfig.MAX_MEMORY_FOR_COMPACTION.key(), maxUseMemory)
.option(HoodiePayloadProps.PAYLOAD_ORDERING_FIELD_PROP_KEY, Constants.UPDATE_TIME)
.option(HoodiePayloadProps.PAYLOAD_EVENT_TIME_FIELD_PROP_KEY, Constants.UPDATE_TIME)
.option(HoodieTableConfig.NAME.key(), config.getName())
.option(HoodieTableConfig.KEY_GENERATOR_CLASS_NAME.key(), ComplexKeyGenerator.class.getName())
.option(HoodieIndexConfig.INDEX_TYPE.key(), HoodieIndex.IndexType.SIMPLE.name())
.option(DataSourceReadOptions.EXTRACT_PARTITION_VALUES_FROM_PARTITION_PATH().key(), true)
.option(HoodieIndexConfig.BUCKET_INDEX_NUM_BUCKETS.key(), 20)
.option(HoodieIndexConfig.BUCKET_INDEX_ENGINE_TYPE.key(), HoodieIndex.BucketIndexEngineType.SIMPLE.name())
.option(HoodieLayoutConfig.LAYOUT_TYPE.key(), HoodieStorageLayout.LayoutType.DEFAULT.name())
.option(HoodieLayoutConfig.LAYOUT_PARTITIONER_CLASS_NAME.key(), SparkBucketIndexPartitioner.class.getName())
.option(HoodieWriteConfig.WRITE_CONCURRENCY_MODE.key(), WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL.name())
.option(HoodieCleanConfig.FAILED_WRITES_CLEANER_POLICY.key(), HoodieFailedWritesCleaningPolicy.LAZY.name())
.option("hoodie.write.lock.provider", HiveMetastoreBasedLockProvider.class.getName())
.option("hoodie.write.lock.hivemetastore.database", "bi_ods_real")
.option("hoodie.write.lock.hivemetastore.table", getTableName())
.mode(SaveMode.Append)
.save(config.getSinkPath());
}
```

**Expected behavior**

A clear and concise description of what you expected to happen.

**Environment Description**

* Hudi version :0.13.1

* Spark version :3.2.1

* Hive version :3.2.1

* Hadoop version :3.2.2

* Storage (HDFS/S3/GCS..) :s3

* Running on Docker? (yes/no) :no

**Additional context**

Add any other context about the problem here.

**Stacktrace**

```Add the stacktrace of the error.```
```
3/06/21 07:52:04 WARN HiveConf: HiveConf of name hive.server2.thrift.url does not exist
23/06/21 07:52:04 WARN BaseTableMetadata: Metadata record for dt=202306 encountered some files to be deleted which was not added before. Ignoring the spurious deletes as the `_hoodie.metadata.ignore.spurious.deletes` config is set to true
23/06/21 07:52:06 WARN HiveConf: HiveConf of name hive.server2.thrift.url does not exist
23/06/21 07:52:13 ERROR HoodieTimelineArchiver: Failed to archive commits, .commit file: 20220429184323.compaction.requested
org.apache.hudi.exception.HoodieIOException: `s3a://xxdata/hudi/tables/ods_api_test_case/.hoodie/.temp/20220429184323': Directory is not empty
at org.apache.hudi.common.fs.FSUtils.deleteDir(FSUtils.java:750)
at org.apache.hudi.table.marker.DirectWriteMarkers.deleteMarkerDir(DirectWriteMarkers.java:82)
at org.apache.hudi.client.HoodieTimelineArchiver.deleteAnyLeftOverMarkers(HoodieTimelineArchiver.java:680)
at org.apache.hudi.client.HoodieTimelineArchiver.archive(HoodieTimelineArchiver.java:654)
at org.apache.hudi.client.HoodieTimelineArchiver.archiveIfRequired(HoodieTimelineArchiver.java:175)
at org.apache.hudi.client.BaseHoodieTableServiceClient.archive(BaseHoodieTableServiceClient.java:580)
at org.apache.hudi.client.BaseHoodieWriteClient.archive(BaseHoodieWriteClient.java:783)
at org.apache.hudi.client.BaseHoodieWriteClient.autoArchiveOnCommit(BaseHoodieWriteClient.java:573)
at org.apache.hudi.client.BaseHoodieWriteClient.mayBeCleanAndArchive(BaseHoodieWriteClient.java:539)
at org.apache.hudi.client.BaseHoodieWriteClient.commitStats(BaseHoodieWriteClient.java:245)
at org.apache.hudi.client.SparkRDDWriteClient.commit(SparkRDDWriteClient.java:102)
at org.apache.hudi.HoodieSparkSqlWriter$.commitAndPerformPostOperations(HoodieSparkSqlWriter.scala:954)
at org.apache.hudi.HoodieSparkSqlWriter$.write(HoodieSparkSqlWriter.scala:381)
at org.apache.hudi.DefaultSource.createRelation(DefaultSource.scala:150)
at org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:45)
at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:75)
at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:73)
at org.apache.spark.sql.execution.command.ExecutedCommandExec.executeCollect(commands.scala:84)
at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.$anonfun$applyOrElse$1(QueryExecution.scala:115)
at org.apache.spark.sql.catalyst.QueryPlanningTracker$.withTracker(QueryPlanningTracker.scala:107)
at org.apache.spark.sql.execution.SQLExecution$.withTracker(SQLExecution.scala:232)
at org.apache.spark.sql.execution.SQLExecution$.executeQuery$1(SQLExecution.scala:110)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$6(SQLExecution.scala:135)
at org.apache.spark.sql.catalyst.QueryPlanningTracker$.withTracker(QueryPlanningTracker.scala:107)
at org.apache.spark.sql.execution.SQLExecution$.withTracker(SQLExecution.scala:232)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:135)
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:253)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:134)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:775)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:68)
at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:112)
at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:108)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:775)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:68)
at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:112)
at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:108)
at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:519)
at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:83)
at org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:519)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:30)
at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:267)
at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:263)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:30)
at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:30)
at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:495)
at org.apache.spark.sql.execution.QueryExecution.eagerlyExecuteCommands(QueryExecution.scala:108)
at org.apache.spark.sql.execution.QueryExecution.commandExecuted$lzycompute(QueryExecution.scala:95)
at org.apache.spark.sql.execution.QueryExecution.commandExecuted(QueryExecution.scala:93)
at org.apache.spark.sql.execution.QueryExecution.assertCommandExecuted(QueryExecution.scala:136)
at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:848)
at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:382)
at org.apache.spark.sql.DataFrameWriter.saveInternal(DataFrameWriter.scala:355)
at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:239)
at com.tuya.HuDiMergeFunction.upsertAllAction(HuDiMergeFunction.java:234)
at com.tuya.HuDiMergeFunction.call(HuDiMergeFunction.java:115)
at com.tuya.HuDiMergeFunction.call(HuDiMergeFunction.java:46)
at org.apache.spark.sql.streaming.DataStreamWriter.$anonfun$foreachBatch$1(DataStreamWriter.scala:489)
at org.apache.spark.sql.streaming.DataStreamWriter.$anonfun$foreachBatch$1$adapted(DataStreamWriter.scala:489)
at org.apache.spark.sql.execution.streaming.sources.ForeachBatchSink.addBatch(ForeachBatchSink.scala:35)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runBatch$17(MicroBatchExecution.scala:600)
at org.apache.spark.sql.catalyst.QueryPlanningTracker$.withTracker(QueryPlanningTracker.scala:107)
at org.apache.spark.sql.execution.SQLExecution$.withTracker(SQLExecution.scala:232)
at org.apache.spark.sql.execution.SQLExecution$.executeQuery$1(SQLExecution.scala:110)
at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$6(SQLExecution.scala:135)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:187)
at org.apache.spark.sql.execution.streaming.StreamExecution.$anonfun$runStream$1(StreamExecution.scala:303)
at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:775)
at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:286)
at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:209)
Caused by: org.apache.hadoop.fs.PathIsNotEmptyDirectoryException: `s3a://xxx/hudi/tables/ods_api_test_case/.hoodie/.temp/20220429184323': Directory is not empty
at org.apache.hadoop.fs.s3a.S3AFileSystem.innerDelete(S3AFileSystem.java:1806)
at org.apache.hadoop.fs.s3a.S3AFileSystem.delete(S3AFileSystem.java:1751)
at org.apache.hudi.common.fs.HoodieWrapperFileSystem.lambda$delete$14(HoodieWrapperFileSystem.java:357)
at org.apache.hudi.common.fs.HoodieWrapperFileSystem.executeFuncWithTimeMetrics(HoodieWrapperFileSystem.java:114)
at org.apache.hudi.common.fs.HoodieWrapperFileSystem.delete(HoodieWrapperFileSystem.java:356)
at org.apache.hudi.common.fs.FSUtils.deleteDir(FSUtils.java:745)
... 84 more
```

Start with FSUtils.java, DirectWriteMarkers.java, and HoodieTimelineArchiver.java, following the stack trace from archive cleanup through deletion of the marker directory. Reproduce the Hudi 0.13.1 Spark write on S3 and determine why the directory remains non-empty; done requires identifying the cleanup failure and confirming the behavior with a focused regression test.

Written by the indexing model from the issue text.