void cleanExpiredRequestInQueue(final BlockingQueue<Runnable> blockingQueue, final long maxWaitTimeMillsInQueue) {
        while (true) {
            try {
                if (!blockingQueue.isEmpty()) {
                    final Runnable runnable = blockingQueue.peek();
                    if (null == runnable) {
                        break;
                    }
                    final RequestTask rt = castRunnable(runnable);
                    if (rt == null || rt.isStopRun()) {
                        break;
                    }

                    final long behind = System.currentTimeMillis() - rt.getCreateTimestamp();
                    if (behind >= maxWaitTimeMillsInQueue) {
                        if (blockingQueue.remove(runnable)) {
                            rt.setStopRun(true);
                            rt.returnResponse(RemotingSysResponseCode.SYSTEM_BUSY, String.format("[TIMEOUT_CLEAN_QUEUE]broker busy, start flow control for a while, period in queue: %sms, size of queue: %d", behind, blockingQueue.size()));
                            if (System.currentTimeMillis() - jstackTime > 15000) {
                                jstackTime = System.currentTimeMillis();
                                LOGGER.warn("broker jstack \n " + UtilAll.jstack());
                            }
                        }
                    } else {
                        break;
                    }
                } else {
                    break;
                }
            } catch (Throwable ignored) {
            }
        }
    }
@Test
    public void testCleanExpiredRequestInQueue() throws Exception {
        BrokerFastFailure brokerFastFailure = new BrokerFastFailure(brokerController);

        BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
        brokerFastFailure.cleanExpiredRequestInQueue(queue, 1);
        assertThat(queue.size()).isZero();

        //Normal Runnable
        Runnable runnable = new Runnable() {
            @Override
            public void run() {

            }
        };
        queue.add(runnable);

        assertThat(queue.size()).isEqualTo(1);
        brokerFastFailure.cleanExpiredRequestInQueue(queue, 1);
        assertThat(queue.size()).isEqualTo(1);

        queue.clear();

        //With expired request
        RequestTask expiredRequest = new RequestTask(runnable, null, null);
        queue.add(new FutureTaskExt<>(expiredRequest, null));
        TimeUnit.MILLISECONDS.sleep(100);

        RequestTask requestTask = new RequestTask(runnable, null, null);
        queue.add(new FutureTaskExt<>(requestTask, null));

        assertThat(queue.size()).isEqualTo(2);
        brokerFastFailure.cleanExpiredRequestInQueue(queue, 100);
        assertThat(queue.size()).isEqualTo(1);
        assertThat(((FutureTaskExt) queue.peek()).getRunnable()).isEqualTo(requestTask);
    }