@Override
    public Message request(final Message msg, final long timeout) throws RequestTimeoutException, MQClientException,
        RemotingException, MQBrokerException, InterruptedException {
        msg.setTopic(withNamespace(msg.getTopic()));
        return this.defaultMQProducerImpl.request(msg, timeout);
    }
@Test
    public void testAsyncRequest_OnSuccess() throws Exception {
        when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createTopicRoute());
        final CountDownLatch countDownLatch = new CountDownLatch(1);
        RequestCallback requestCallback = new RequestCallback() {
            @Override
            public void onSuccess(Message message) {
                assertThat(message).isExactlyInstanceOf(MessageExt.class);
                assertThat(message.getTopic()).isEqualTo("FooBar");
                assertThat(message.getBody()).isEqualTo(new byte[] {'a'});
                assertThat(message.getFlag()).isEqualTo(1);
                countDownLatch.countDown();
            }

            @Override
            public void onException(Throwable e) {
            }
        };
        producer.request(message, requestCallback, 3 * 1000L);
        ConcurrentHashMap<String, RequestResponseFuture> responseMap = RequestFutureHolder.getInstance().getRequestFutureTable();
        assertThat(responseMap).isNotNull();

        MessageExt responseMsg = new MessageExt();
        responseMsg.setTopic(message.getTopic());
        responseMsg.setBody(message.getBody());
        responseMsg.setFlag(1);
        for (Map.Entry<String, RequestResponseFuture> entry : responseMap.entrySet()) {
            RequestResponseFuture future = entry.getValue();
            future.setSendRequestOk(true);
            future.getRequestCallback().onSuccess(responseMsg);
        }
        countDownLatch.await(defaultTimeout, TimeUnit.MILLISECONDS);
    }