Cooperative thread blocked by scheduling close operation

[martinkrc]

Using BlockHound we observe that cooperative threads can be blocked by scheduling closeProcessor operations. See the stacktrace below:

java.lang.Exception: Blocking call: jdk.internal.misc.Unsafe#park (hz.mmbinst-02.jet.cooperative.thread-0)
	at application.BlockHoundConfigurer.lambda$null$2(BlockHoundConfigurer.java:71)
	at reactor.blockhound.BlockHound$Builder.lambda$install$8(BlockHound.java:427)
	at reactor.blockhound.BlockHoundRuntime.checkBlocking(BlockHoundRuntime.java:89)
	at java.base/jdk.internal.misc.Unsafe.park(Unsafe.java)
	at java.base/java.util.concurrent.locks.LockSupport.park(LockSupport.java:194)
	at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:885)
	at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireQueued(AbstractQueuedSynchronizer.java:917)
	at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:1240)
	at java.base/java.util.concurrent.locks.ReentrantLock.lock(ReentrantLock.java:267)
	at java.base/java.util.concurrent.LinkedBlockingQueue.signalNotEmpty(LinkedBlockingQueue.java:173)
	at java.base/java.util.concurrent.LinkedBlockingQueue.offer(LinkedBlockingQueue.java:421)
	at com.hazelcast.internal.util.executor.CachedExecutorServiceDelegate.execute(CachedExecutorServiceDelegate.java:107)
	at com.hazelcast.internal.util.executor.CachedExecutorServiceDelegate.submit(CachedExecutorServiceDelegate.java:123)
	at com.hazelcast.internal.util.executor.CachedExecutorServiceDelegate.submit(CachedExecutorServiceDelegate.java:129)
	at com.hazelcast.jet.impl.execution.ProcessorTasklet.stateMachineStep(ProcessorTasklet.java:387)
	at com.hazelcast.jet.impl.execution.ProcessorTasklet.stateMachineStep(ProcessorTasklet.java:380)
	at com.hazelcast.jet.impl.execution.ProcessorTasklet.call(ProcessorTasklet.java:247)
	at com.hazelcast.jet.impl.execution.TaskletExecutionService$CooperativeWorker.runTasklet(TaskletExecutionService.java:367)
	at java.base/java.util.concurrent.CopyOnWriteArrayList.forEach(CopyOnWriteArrayList.java:807)
	at com.hazelcast.jet.impl.execution.TaskletExecutionService$CooperativeWorker.run(TaskletExecutionService.java:347)
	at java.base/java.lang.Thread.run(Thread.java:829)

Using Hazelcast Jet 4.5.3.