@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_OnException() throws Exception {
        final AtomicInteger cc = new AtomicInteger(0);
        final CountDownLatch countDownLatch = new CountDownLatch(1);
        RequestCallback requestCallback = new RequestCallback() {
            @Override
            public void onSuccess(Message message) {

            }

            @Override
            public void onException(Throwable e) {
                cc.incrementAndGet();
                countDownLatch.countDown();
            }
        };
        MessageQueueSelector messageQueueSelector = new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                return null;
            }
        };

        try {
            producer.request(message, requestCallback, 3 * 1000L);
            failBecauseExceptionWasNotThrown(Exception.class);
        } catch (Exception e) {
            ConcurrentHashMap<String, RequestResponseFuture> responseMap = RequestFutureHolder.getInstance().getRequestFutureTable();
            assertThat(responseMap).isNotNull();
            for (Map.Entry<String, RequestResponseFuture> entry : responseMap.entrySet()) {
                RequestResponseFuture future = entry.getValue();
                future.getRequestCallback().onException(e);
            }
        }
        countDownLatch.await(defaultTimeout, TimeUnit.MILLISECONDS);
        assertThat(cc.get()).isEqualTo(1);
    }