webflux Bad async context state

用JMETER轻压。
源码：
-------------------------------------------------
```
@RestController
@RequestMapping("/ordercenterdatalist/orderlist")
public class OrderListController {

@SentinelResource
@PostMapping(value = "list", produces = "application/json")
public Mono list(@RequestBody OrderListParameter parameter) {
List commands = new ArrayList<>();
{
Command cmd = new Command();
cmd.setName("agent.api.test.getorderlist");
commands.add(cmd);
}
{
Command cmd = new Command();
cmd.setName("agent.api.miandan.getorderlistexception");
commands.add(cmd);
}
return Flux.fromIterable(commands)
.parallel(5)
.runOn(SchedulerFactory.getScheduler("adsf-parallel"))
.flatMap(p -> {
return Mono.just(1)
.publishOn(SchedulerFactory.getScheduler(p.getName()))
.flatMap(pp -> {
if ("agent.api.test.getorderlist".equals(p.getName())) {
return Mono.just(new TestGetOrderListResponse());
}
return Mono.error(new Exception("测试"));
})
.transform(new SentinelReactorTransformer<>(p.getName()))
.onErrorContinue((e, o) -> {

log.error("err:\n{}\no:{}",
ExceptionUtils.getStackTrace(e), o);
});
})
.sequential()
.then(Mono.just(new OrderListResponse()));
}

}
```
-------------------------------------------------
```
public class SchedulerFactory {
private static final ConcurrentHashMap map = new ConcurrentHashMap<>();

public static Scheduler getScheduler(String agentName) {
return map.computeIfAbsent(agentName, k -> Schedulers.newParallel(agentName,
getParallelism(agentName)));
}

private static int getParallelism(String name) {
int parallelism = Runtime.getRuntime().availableProcessors() * 1;
if (StringUtils.isBlank(name)) {
return parallelism;
}

// TODO
return parallelism;
}
}
```
------------------------------------------------------
```
// 加配置：
{
DegradeRule rule = new DegradeRule();
rule.setResource(ruleResource);
rule.setGrade(RuleConstant.DEGRADE_GRADE_EXCEPTION_RATIO); // 异常比例
rule.setTimeWindow(60); // 熔断时长60秒
rule.setStatIntervalMs(1000); // 统计时长1000毫秒
rule.setMinRequestAmount(5); // 最小调用数量
rule.setCount(0.01); // 比例阈值
rules.add(rule);
}
```
------------------------------------------------------
```
// Command.java：
@Data
public class Command {
private String name;
}
```

------------------------------------------------------
错误一：

```
java.lang.IllegalStateException: Bad async context state, expected entry: agent.api.miandan.getorderlistexception@953301447, but actual: /demo/orderlist/list@907395445
at com.alibaba.csp.sentinel.AsyncEntry.cleanCurrentEntryInLocal(AsyncEntry.java: 59)
Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:
Assembly trace from producer [com.alibaba.csp.sentinel.adapter.reactor.MonoSentinelOperator
] :
reactor.core.publisher.Mono.transform
com.xx.demo.controller.OrderListController.lambda$list$1(OrderListController.java: 67)
Error has been observed at the following site(s):
|_ Mono.transform ⇢ at com.xx.demo.controller.OrderListController.lambda$list$1(OrderListController.java: 67)
|_ ParallelFlux.flatMap ⇢ at com.xx.demo.controller.OrderListController.list(OrderListController.java: 58)
|_ ParallelFlux.sequential ⇢ at com.xx.demo.controller.OrderListController.list(OrderListController.java: 69)
|_ Flux.then ⇢ at com.xx.demo.controller.OrderListController.list(OrderListController.java: 70)
|_ Mono.from ⇢ at org.springframework.core.ReactiveAdapterRegistry$ReactorAdapter.toPublisher(ReactiveAdapterRegistry.java: 389)
|_ Mono.from ⇢ at org.springframework.http.codec.json.AbstractJackson2Encoder.encode(AbstractJackson2Encoder.java: 145)
|_ Mono.map ⇢ at org.springframework.http.codec.json.AbstractJackson2Encoder.encode(AbstractJackson2Encoder.java: 146)
|_ Mono.flux ⇢ at org.springframework.http.codec.json.AbstractJackson2Encoder.encode(AbstractJackson2Encoder.java: 147)
|_ Flux.singleOrEmpty ⇢ at org.springframework.http.codec.EncoderHttpMessageWriter.write(EncoderHttpMessageWriter.java: 125)
|_ Mono.switchIfEmpty ⇢ at org.springframework.http.codec.EncoderHttpMessageWriter.write(EncoderHttpMessageWriter.java: 126)
|_ Mono.flatMap ⇢ at org.springframework.http.codec.EncoderHttpMessageWriter.write(EncoderHttpMessageWriter.java: 130)
|_ Mono.doOnDiscard ⇢ at org.springframework.http.codec.EncoderHttpMessageWriter.write(EncoderHttpMessageWriter.java: 136)
|_ ⇢ at org.springframework.http.codec.EncoderHttpMessageWriter.write(EncoderHttpMessageWriter.java: 216)
|_ ⇢ at org.springframework.web.reactive.result.method.annotation.AbstractMessageWriterResultHandler.writeBody(AbstractMessageWriterResultHandler.java: 104)
|_ ⇢ at org.springframework.web.reactive.result.method.annotation.ResponseBodyResultHandler.handleResult(ResponseBodyResultHandler.java: 86)
|_ checkpoint ⇢ Handler com.xx.demo.controller.OrderListController#list(OrderListParameter) [DispatcherHandler
]
|_ Mono.flatMap ⇢ at org.springframework.web.reactive.DispatcherHandler.lambda$handleResult$5(DispatcherHandler.java: 172)
|_ Mono.onErrorResume ⇢ at org.springframework.web.reactive.DispatcherHandler.handleResult(DispatcherHandler.java: 171)
|_ ⇢ at org.springframework.web.reactive.DispatcherHandler.lambda$handle$2(DispatcherHandler.java: 147)
|_ Mono.flatMap ⇢ at org.springframework.web.reactive.DispatcherHandler.handle(DispatcherHandler.java: 147)
|_ ⇢ at org.springframework.web.server.handler.DefaultWebFilterChain.lambda$filter$0(DefaultWebFilterChain.java: 120)
|_ Mono.defer ⇢ at org.springframework.web.server.handler.DefaultWebFilterChain.filter(DefaultWebFilterChain.java: 119)
|_ ⇢ at org.springframework.cloud.gateway.filter.WeightCalculatorWebFilter.filter(WeightCalculatorWebFilter.java: 243)
|_ checkpoint ⇢ org.springframework.cloud.gateway.filter.WeightCalculatorWebFilter [DefaultWebFilterChain
]
|_ ⇢ at org.springframework.web.server.handler.DefaultWebFilterChain.lambda$filter$0(DefaultWebFilterChain.java: 120)
|_ Mono.defer ⇢ at org.springframework.web.server.handler.DefaultWebFilterChain.filter(DefaultWebFilterChain.java: 119)
|_ Mono.transform ⇢ at com.alibaba.csp.sentinel.adapter.spring.webflux.SentinelWebFluxFilter.filter(SentinelWebFluxFilter.java: 50)
|_ checkpoint ⇢ com.alibaba.csp.sentinel.adapter.spring.webflux.SentinelWebFluxFilter [DefaultWebFilterChain
]
|_ ⇢ at org.springframework.web.server.handler.DefaultWebFilterChain.lambda$filter$0(DefaultWebFilterChain.java: 120)
|_ Mono.defer ⇢ at org.springframework.web.server.handler.DefaultWebFilterChain.filter(DefaultWebFilterChain.java: 119)
|_ Mono.doOnSuccess ⇢ at org.springframework.boot.actuate.metrics.web.reactive.server.MetricsWebFilter.filter(MetricsWebFilter.java: 78)
|_ Mono.doOnError ⇢ at org.springframework.boot.actuate.metrics.web.reactive.server.MetricsWebFilter.filter(MetricsWebFilter.java: 79)
|_ Mono.transformDeferred ⇢ at org.springframework.boot.actuate.metrics.web.reactive.server.MetricsWebFilter.filter(MetricsWebFilter.java: 73)
|_ checkpoint ⇢ org.springframework.boot.actuate.metrics.web.reactive.server.MetricsWebFilter [DefaultWebFilterChain
]
|_ ⇢ at org.springframework.web.server.handler.DefaultWebFilterChain.lambda$filter$0(DefaultWebFilterChain.java: 120)
|_ Mono.defer ⇢ at org.springframework.web.server.handler.DefaultWebFilterChain.filter(DefaultWebFilterChain.java: 119)
|_ ⇢ at org.springframework.web.server.handler.FilteringWebHandler.handle(FilteringWebHandler.java: 59)
|_ ⇢ at org.springframework.web.server.handler.WebHandlerDecorator.handle(WebHandlerDecorator.java: 56)
|_ Mono.error ⇢ at org.springframework.web.server.handler.ExceptionHandlingWebHandler$CheckpointInsertingHandler.handle(ExceptionHandlingWebHandler.java: 98)
|_ checkpoint ⇢ HTTP POST "/demo/orderlist/list" [ExceptionHandlingWebHandler
]
|_ ⇢ at org.springframework.web.server.handler.ExceptionHandlingWebHandler.lambda$handle$0(ExceptionHandlingWebHandler.java: 77)
|_ Mono.onErrorResume ⇢ at org.springframework.web.server.handler.ExceptionHandlingWebHandler.handle(ExceptionHandlingWebHandler.java: 77)
Stack trace:
at com.alibaba.csp.sentinel.AsyncEntry.cleanCurrentEntryInLocal(AsyncEntry.java: 59)
at com.alibaba.csp.sentinel.CtSph.asyncEntryWithPriorityInternal(CtSph.java: 107)
at com.alibaba.csp.sentinel.CtSph.asyncEntryWithType(CtSph.java: 354)
at com.alibaba.csp.sentinel.SphU.asyncEntry(SphU.java: 366)
at com.alibaba.csp.sentinel.adapter.reactor.SentinelReactorSubscriber.entryWhenSubscribed(SentinelReactorSubscriber.java: 92)
at com.alibaba.csp.sentinel.context.ContextUtil.runOnContext(ContextUtil.java: 276)
at com.alibaba.csp.sentinel.adapter.reactor.SentinelReactorSubscriber.doWithContextOrCurrent(SentinelReactorSubscriber.java: 81)
at com.alibaba.csp.sentinel.adapter.reactor.SentinelReactorSubscriber.hookOnSubscribe(SentinelReactorSubscriber.java: 112)
at com.alibaba.csp.sentinel.adapter.reactor.InheritableBaseSubscriber.onSubscribe(InheritableBaseSubscriber.java: 136)
at com.alibaba.csp.sentinel.adapter.reactor.SentinelReactorSubscriber.onSubscribe(SentinelReactorSubscriber.java: 37)
at reactor.core.publisher.MonoFlatMap.subscribeOrReturn(MonoFlatMap.java: 58)
at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java: 57)
at com.alibaba.csp.sentinel.adapter.reactor.MonoSentinelOperator.subscribe(MonoSentinelOperator.java: 40)
at reactor.core.publisher.Mono.subscribe(Mono.java: 4046)
at reactor.core.publisher.FluxFlatMap$FlatMapMain.onNext(FluxFlatMap.java: 425)
at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.runAsync(FluxPublishOn.java: 440)
at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.run(FluxPublishOn.java: 527)
at reactor.core.scheduler.WorkerTask.call(WorkerTask.java: 84)
at reactor.core.scheduler.WorkerTask.call(WorkerTask.java: 37)
at java.util.concurrent.FutureTask.run$$$capture(FutureTask.java: 266)
at java.util.concurrent.FutureTask.run(FutureTask.java)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java: 180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java: 293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java: 1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java: 617)
at java.lang.Thread.run(Thread.java: 748)
```

----------------------------------------------

错误二：

```
2021-09-02 18:01:28.075 ERROR [][][][][][reactor.netty.transport.ServerTransport][error:314]|reactor-http-nio-3|[id: 0x6dcab4cf, L:/127.0.0.1:8080 ! R:/127.0.0.1:51131] onUncaughtException(SimpleConnection{channel=[id: 0x6dcab4cf, L:/127.0.0.1:8080 ! R:/127.0.0.1:51131]})
java.lang.StackOverflowError: null
at com.alibaba.csp.sentinel.CtEntry.exitForContext(CtEntry.java:99)
at com.alibaba.csp.sentinel.AsyncEntry.trueExit(AsyncEntry.java:94)
at com.alibaba.csp.sentinel.CtEntry.exit(CtEntry.java:66)
at com.alibaba.csp.sentinel.AsyncEntry.exit(AsyncEntry.java:30)
at com.alibaba.csp.sentinel.CtEntry.exitForContext(CtEntry.java:99)
at com.alibaba.csp.sentinel.AsyncEntry.trueExit(AsyncEntry.java:94)
```

Start by reproducing the reported WebFlux pattern in OrderListController, especially the parallel execution and SentinelReactorTransformer at lines 58, 67, and 69. Read SentinelReactorSubscriber.java, AsyncEntry.java, and CtSph.java around the reported stack frames to trace async context entry cleanup. Done means the concurrent request no longer raises Bad async context state under the supplied workload.

Written by the indexing model from the issue text.