void submitConsumeRequest(ConsumeRequest consumeRequest) {
        try {
            consumeRequestCache.put(consumeRequest);
        } catch (InterruptedException ignore) {
        }
    }
@Test
    public void testSubmitConsumeRequest() throws Exception {
        byte[] body = new byte[] {'1', '2', '3'};
        MessageExt consumedMsg = new MessageExt();
        consumedMsg.setMsgId("NewMsgId");
        consumedMsg.setBody(body);
        consumedMsg.putUserProperty(NonStandardKeys.MESSAGE_DESTINATION, "TOPIC");
        consumedMsg.setTopic("HELLO_QUEUE");

        when(consumeRequest.getMessageExt()).thenReturn(consumedMsg);
        localMessageCache.submitConsumeRequest(consumeRequest);
        assertThat(localMessageCache.poll()).isEqualTo(consumedMsg);
    }