[Bug]: After the token expires,the optimizer repeatedly commits tasks that have been reset to ams

[rfyu]

What happened?

After the token expires,,the optimizer repeatedly commits tasks that have been reset to ams, so that new tasks cannot be pulled.

Affects Versions

master

What engines are you seeing the problem on?

AMS

How to reproduce

1. Start an optimizer with only one thread
2. optimizer polledTask and ackedTask
3. token expire and reset task
4. completeTask
5. Then you will find that the optimizer performs completeTask repeatedly.

Relevant log output

2024-03-11 18:20:00,388 ERROR com.netease.arctic.optimizer.common.AbstractOptimizerOperator [] [flink-optimizer-executor-1] - Call ams got an error and will try again later
com.netease.arctic.ams.api.ArcticException: java.lang.IllegalStateException: Task not scheduled yet, taskId:OptimizingTaskId(processId:1710151650912, taskId:30)
    at com.netease.arctic.ams.api.OptimizingService$completeTask_result$completeTask_resultStandardScheme.read(OptimizingService.java:4937) ~[MuiseOptimizer.jar:?]
    at com.netease.arctic.ams.api.OptimizingService$completeTask_result$completeTask_resultStandardScheme.read(OptimizingService.java:4923) ~[MuiseOptimizer.jar:?]
    at com.netease.arctic.ams.api.OptimizingService$completeTask_result.read(OptimizingService.java:4873) ~[MuiseOptimizer.jar:?]
    at org.apache.thrift.TServiceClient.receiveBase(TServiceClient.java:88) ~[MuiseOptimizer.jar:?]
    at com.netease.arctic.ams.api.OptimizingService$Client.recv_completeTask(OptimizingService.java:176) ~[MuiseOptimizer.jar:?]
    at com.netease.arctic.ams.api.OptimizingService$Client.completeTask(OptimizingService.java:162) ~[MuiseOptimizer.jar:?]
    at sun.reflect.GeneratedMethodAccessor58.invoke(Unknown Source) ~[?:?]
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_121]
    at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_121]
    at com.netease.arctic.ams.api.client.ThriftClientPool.lambda$iface$1(ThriftClientPool.java:261) ~[MuiseOptimizer.jar:?]
    at com.sun.proxy.$Proxy37.completeTask(Unknown Source) ~[?:?]
    at com.netease.arctic.optimizer.common.OptimizerExecutor.lambda$completeTask$2(OptimizerExecutor.java:142) ~[MuiseOptimizer.jar:?]
    at com.netease.arctic.optimizer.common.AbstractOptimizerOperator.callAuthenticatedAms(AbstractOptimizerOperator.java:95) ~[MuiseOptimizer.jar:?]
    at com.netease.arctic.optimizer.common.OptimizerExecutor.completeTask(OptimizerExecutor.java:140) ~[MuiseOptimizer.jar:?]
    at com.netease.arctic.optimizer.common.OptimizerExecutor.start(OptimizerExecutor.java:53) ~[MuiseOptimizer.jar:?]
    at com.netease.arctic.optimizer.flink.FlinkExecutor.lambda$open$0(FlinkExecutor.java:41) ~[MuiseOptimizer.jar:?]
    at java.lang.Thread.run(Thread.java:745) [?:1.8.0_121]

Anything else

No response

Are you willing to submit a PR?

• [X] Yes I am willing to submit a PR!

Code of Conduct

• [X] I agree to follow this project's Code of Conduct