public void handleMessage(
        MessageExt messageExt) throws MQClientException, RemotingException, InterruptedException {
        if (rocketMQListener != null) {
            rocketMQListener.onMessage(doConvertMessage(messageExt));
        } else if (rocketMQReplyListener != null) {
            Object replyContent = rocketMQReplyListener.onMessage(doConvertMessage(messageExt));
            Message<?> message = MessageBuilder.withPayload(replyContent).build();

            org.apache.rocketmq.common.message.Message replyMessage = MessageUtil.createReplyMessage(messageExt, convertToBytes(message));
            DefaultMQProducer producer = consumer.getDefaultMQPushConsumerImpl().getmQClientFactory().getDefaultMQProducer();
            producer.setSendMsgTimeout(replyTimeout);
            producer.send(replyMessage, new SendCallback() {
                @Override public void onSuccess(SendResult sendResult) {
                    if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
                        log.error("Consumer replies message failed. SendStatus: {}", sendResult.getSendStatus());
                    } else {
                        log.debug("Consumer replies message success.");
                    }
                }

                @Override public void onException(Throwable e) {
                    log.error("Consumer replies message failed. error: {}", e.getLocalizedMessage());
                }
            });
        }
    }
@Test
    public void testHandleMessage() throws Exception {
        DefaultRocketMQListenerContainer listenerContainer = new DefaultRocketMQListenerContainer();
        Method handleMessage = DefaultRocketMQListenerContainer.class.getDeclaredMethod("handleMessage", MessageExt.class);
        handleMessage.setAccessible(true);
        listenerContainer.setRocketMQListener(new RocketMQListener<String>() {
            @Override
            public void onMessage(String message) {
            }
        });
        Field messageType = DefaultRocketMQListenerContainer.class.getDeclaredField("messageType");
        messageType.setAccessible(true);
        messageType.set(listenerContainer, String.class);
        MessageExt messageExt = new MessageExt(0, System.currentTimeMillis(), null, System.currentTimeMillis(), null, null);
        MessageAccessor.putProperty(messageExt, MessageConst.PROPERTY_CLUSTER, "defaultCluster");
        messageExt.setBody("hello".getBytes());
        handleMessage.invoke(listenerContainer, messageExt);

        // reply message
        listenerContainer.setRocketMQListener(null);
        DefaultMQPushConsumer consumer = mock(DefaultMQPushConsumer.class);
        DefaultMQPushConsumerImpl pushConsumer = mock(DefaultMQPushConsumerImpl.class);
        MQClientInstance mqClientInstance = mock(MQClientInstance.class);
        DefaultMQProducer producer = mock(DefaultMQProducer.class);
        when(consumer.getDefaultMQPushConsumerImpl()).thenReturn(pushConsumer);
        when(pushConsumer.getmQClientFactory()).thenReturn(mqClientInstance);
        when(mqClientInstance.getDefaultMQProducer()).thenReturn(producer);
        listenerContainer.setConsumer(consumer);
        listenerContainer.setMessageConverter(new CompositeMessageConverter(Arrays.asList(new StringMessageConverter(), new MappingJackson2MessageConverter())));
        doNothing().when(producer).send(any(MessageExt.class), any(SendCallback.class));
        listenerContainer.setRocketMQReplyListener(new RocketMQReplyListener<String, String>() {
            @Override
            public String onMessage(String message) {
                return "test";
            }
        });
        handleMessage.invoke(listenerContainer, messageExt);
    }