[SUPPORT]Hudi java client throws Error upserting bucketType UPDATE for partition :0

1. Hudi version: 0.12.0, aliyun oss file system,flink 1.13.6
2. Java client config
```java
org.apache.hadoop.conf.Configuration hadoopConf = new org.apache.hadoop.conf.Configuration();
String tablePath = warehousePath + databasePrefix + "/" + tableName;
// initialize the table, if not done already
Path path = new Path(tablePath);
FileSystem fs = FSUtils.getFs(tablePath, hadoopConf);

String schema = Util.getStringFromResource("/" + databasePrefix + "."
+ tableName + ".schema");
if (!fs.exists(path)) {
HoodieTableMetaClient.withPropertyBuilder()
.setBaseFileFormat(baseFileFormat)
.setPartitionFields(partitionFields)
//.setPreCombineField(preCombineField)
.setHiveStylePartitioningEnable(false)
.setTableCreateSchema(schema)
.setTableType(HoodieTableType.MERGE_ON_READ)
.setRecordKeyFields(recordKeyFields)
.setPayloadClassName(DefaultHoodieRecordPayload.class.getName())
.setTableName(tableName)
.initTable(hadoopConf, tablePath);
}

// Create the write client to write some records in
HoodieWriteConfig cfg = HoodieWriteConfig.newBuilder()
.withPath(tablePath)
.withAutoCommit(true)
.withEmbeddedTimelineServerEnabled(false)
.withRollbackUsingMarkers(false)
.withBulkInsertParallelism(parallelism)
.withSchema(schema)
.withSchemaEvolutionEnable(enableSchemaEvolution)
.withParallelism(parallelism, parallelism)
.withDeleteParallelism(1)
//.withEngineType(EngineType.SPARK)
.forTable(tableName)
.withMergeAllowDuplicateOnInserts(true)
.withCleanConfig(HoodieCleanConfig.newBuilder()
.withAutoClean(true)
.withAsyncClean(false)
.build())
.withStorageConfig(
HoodieStorageConfig.newBuilder()
.parquetWriteLegacyFormat("false")
.build())
.withMetadataConfig(
HoodieMetadataConfig.newBuilder()
.withAsyncClean(false)
.withAsyncIndex(true)
.enable(true)
.build())
.withConsistencyGuardConfig(
ConsistencyGuardConfig.newBuilder()
.withEnableOptimisticConsistencyGuard(false)
.build())
.withIndexConfig(
HoodieIndexConfig.newBuilder()
.withIndexType(IndexType.BLOOM)
.build())
.withCompactionConfig(
HoodieCompactionConfig.newBuilder()
.withCompactionLazyBlockReadEnabled(true)
.build())
.withMarkersType(MarkerType.DIRECT.name())
.build();
client = new HoodieJavaWriteClient<>(new HoodieJavaEngineContext(hadoopConf), cfg);
```
3. error message
```log
2022-09-01 09:27:23,262 ERROR shadow.gs.org.apache.hudi.common.util.queue.BoundedInMemoryExecutor [] - error producing records
java.lang.IllegalStateException: Queue closed for enqueueing new entries
at shadow.gs.org.apache.hudi.common.util.queue.BoundedInMemoryQueue.insertRecord(BoundedInMemoryQueue.java:180) ~[robot-stream.jar:?]
at shadow.gs.org.apache.hudi.common.util.queue.IteratorBasedQueueProducer.produce(IteratorBasedQueueProducer.java:46) ~[robot-stream.jar:?]
at shadow.gs.org.apache.hudi.common.util.queue.BoundedInMemoryExecutor.lambda$null$0(BoundedInMemoryExecutor.java:106) ~[robot-stream.jar:?]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_332]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_332]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_332]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_332]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_332]
at java.lang.Thread.run(Thread.java:750) [?:1.8.0_332]
2022-09-01 09:27:23,276 INFO com.aliyun.jindodata.jnative.NativeLogger [] - JdlsFileOutputStreamImpl.cpp:400] locateNextBlock: JfsLocatedBlock {1094210804279118:cloud-emr-prod:1656939867:blk_8051097586785600991_401847197 len 0; getBlockSize()=0; corrupt=0; offset=0; locs=[]}
2022-09-01 09:27:23,312 INFO com.aliyun.jindodata.common.FsStats [] - cmd=upload, src=oss://***, dst=null, size=445676, parameter=null, time-in-ms=39, version=4.3.1
2022-09-01 09:27:23,316 INFO com.aliyun.jindodata.common.FsStats [] - cmd=getFileStatus, src=oss://***, dst=null, size=0, parameter=null, time-in-ms=2, version=4.3.1
2022-09-01 09:27:23,316 INFO shadow.gs.org.apache.hudi.io.HoodieMergeHandle [] - MergeHandle for partitionPath 2022-08-24 fileID 24b999ef-92d2-46f3-8d36-96db76fc02fd-0, took 139 ms.
2022-09-01 09:27:23,316 ERROR shadow.gs.org.apache.hudi.table.action.commit.BaseJavaCommitActionExecutor [] - Error upserting bucketType UPDATE for partition :0
shadow.gs.org.apache.hudi.exception.HoodieException: shadow.gs.org.apache.hudi.exception.HoodieException: java.lang.InterruptedException
at shadow.gs.org.apache.hudi.table.action.commit.JavaMergeHelper.runMerge(JavaMergeHelper.java:103) ~[robot-stream.jar:?]
at shadow.gs.org.apache.hudi.table.action.commit.BaseJavaCommitActionExecutor.handleUpdateInternal(BaseJavaCommitActionExecutor.java:283) ~[robot-stream.jar:?]
at shadow.gs.org.apache.hudi.table.action.commit.BaseJavaCommitActionExecutor.handleUpdate(BaseJavaCommitActionExecutor.java:274) ~[robot-stream.jar:?]
at shadow.gs.org.apache.hudi.table.action.commit.BaseJavaCommitActionExecutor.handleUpsertPartition(BaseJavaCommitActionExecutor.java:247) ~[robot-stream.jar:?]
at shadow.gs.org.apache.hudi.table.action.commit.BaseJavaCommitActionExecutor.handleInsertPartition(BaseJavaCommitActionExecutor.java:260) ~[robot-stream.jar:?]
at shadow.gs.org.apache.hudi.table.action.commit.BaseJavaCommitActionExecutor.lambda$execute$0(BaseJavaCommitActionExecutor.java:122) ~[robot-stream.jar:?]
at java.util.LinkedHashMap.forEach(LinkedHashMap.java:684) ~[?:1.8.0_332]
at shadow.gs.org.apache.hudi.table.action.commit.BaseJavaCommitActionExecutor.execute(BaseJavaCommitActionExecutor.java:118) ~[robot-stream.jar:?]
at shadow.gs.org.apache.hudi.table.action.commit.BaseJavaCommitActionExecutor.execute(BaseJavaCommitActionExecutor.java:68) ~[robot-stream.jar:?]
at shadow.gs.org.apache.hudi.table.action.commit.BaseWriteHelper.write(BaseWriteHelper.java:57) ~[robot-stream.jar:?]
at shadow.gs.org.apache.hudi.table.action.commit.JavaInsertCommitActionExecutor.execute(JavaInsertCommitActionExecutor.java:47) ~[robot-stream.jar:?]
at shadow.gs.org.apache.hudi.table.HoodieJavaCopyOnWriteTable.insert(HoodieJavaCopyOnWriteTable.java:111) ~[robot-stream.jar:?]
at shadow.gs.org.apache.hudi.table.HoodieJavaCopyOnWriteTable.insert(HoodieJavaCopyOnWriteTable.java:82) ~[robot-stream.jar:?]
at shadow.gs.org.apache.hudi.client.HoodieJavaWriteClient.insert(HoodieJavaWriteClient.java:126) ~[robot-stream.jar:?]
at com.robot.gs.sink.HudiSink.invoke(HudiSink.java:113) ~[robot-stream.jar:?]
at org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:54) ~[robot-stream.jar:?]
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:71) ~[robot-stream.jar:?]
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:46) ~[robot-stream.jar:?]
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:26) ~[robot-stream.jar:?]
at org.apache.flink.streaming.runtime.tasks.BroadcastingOutputCollector.collect(BroadcastingOutputCollector.java:75) ~[robot-stream.jar:?]
at org.apache.flink.streaming.runtime.tasks.BroadcastingOutputCollector.collect(BroadcastingOutputCollector.java:32) ~[robot-stream.jar:?]
at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:50) ~[robot-stream.jar:?]
at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:28) ~[robot-stream.jar:?]
at org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:50) ~[robot-stream.jar:?]
at com.robot.gs.process.DuplicatingRccTaskReportStream.processElement(DuplicatingRccTaskReportStream.java:29) ~[robot-stream.jar:?]
at com.robot.gs.process.DuplicatingRccTaskReportStream.processElement(DuplicatingRccTaskReportStream.java:10) ~[robot-stream.jar:?]
at org.apache.flink.streaming.api.operators.ProcessOperator.processElement(ProcessOperator.java:66) ~[robot-stream.jar:?]
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:71) ~[robot-stream.jar:?]
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:46) ~[robot-stream.jar:?]
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:26) ~[robot-stream.jar:?]
at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:50) ~[robot-stream.jar:?]
at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:28) ~[robot-stream.jar:?]
at org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:50) ~[robot-stream.jar:?]
at com.robot.gs.process.BatchingDataInWindow.process(BatchingDataInWindow.java:18) ~[robot-stream.jar:?]
at org.apache.flink.streaming.runtime.operators.windowing.functions.InternalIterableProcessAllWindowFunction.process(InternalIterableProcessAllWindowFunction.java:64) ~[robot-stream.jar:?]
at org.apache.flink.streaming.runtime.operators.windowing.functions.InternalIterableProcessAllWindowFunction.process(InternalIterableProcessAllWindowFunction.java:33) ~[robot-stream.jar:?]
at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator.emitWindowContents(WindowOperator.java:577) ~[robot-stream.jar:?]
at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator.processElement(WindowOperator.java:434) ~[robot-stream.jar:?]
at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:205) ~[robot-stream.jar:?]
at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:134) ~[robot-stream.jar:?]
at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:105) ~[robot-stream.jar:?]
at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:66) ~[robot-stream.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:423) ~[robot-stream.jar:?]
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:204) ~[robot-stream.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:684) ~[robot-stream.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTask.executeInvoke(StreamTask.java:639) ~[robot-stream.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:650) [robot-stream.jar:?]
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:623) [robot-stream.jar:?]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:779) [robot-stream.jar:?]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566) [robot-stream.jar:?]
at java.lang.Thread.run(Thread.java:750) [?:1.8.0_332]
Caused by: shadow.gs.org.apache.hudi.exception.HoodieException: java.lang.InterruptedException
at shadow.gs.org.apache.hudi.common.util.queue.BoundedInMemoryExecutor.execute(BoundedInMemoryExecutor.java:159) ~[robot-stream.jar:?]
at shadow.gs.org.apache.hudi.table.action.commit.JavaMergeHelper.runMerge(JavaMergeHelper.java:101) ~[robot-stream.jar:?]
... 50 more
Caused by: java.lang.InterruptedException
at java.util.concurrent.FutureTask.awaitDone(FutureTask.java:404) ~[?:1.8.0_332]
at java.util.concurrent.FutureTask.get(FutureTask.java:191) ~[?:1.8.0_332]
at shadow.gs.org.apache.hudi.common.util.queue.BoundedInMemoryExecutor.execute(BoundedInMemoryExecutor.java:155) ~[robot-stream.jar:?]
at shadow.gs.org.apache.hudi.table.action.commit.JavaMergeHelper.runMerge(JavaMergeHelper.java:101) ~[robot-stream.jar:?]
... 50 more
```

Start with JavaMergeHelper.java:103, BaseJavaCommitActionExecutor.java:283, and BoundedInMemoryQueue.java:180, then trace the provided HoodieJavaWriteClient configuration and stack trace. Reproduce the upsert failure with the reported Hudi 0.12.0, Flink 1.13.6, and Aliyun OSS setup; done means the update completes without the queue-closed or interrupted exception.

Written by the indexing model from the issue text.