@SuppressWarnings("unchecked")
    private Object doConvertMessage(MessageExt messageExt) {
        if (Objects.equals(messageType, MessageExt.class) || Objects.equals(messageType, org.apache.rocketmq.common.message.Message.class)) {
            return messageExt;
        } else {
            String str = new String(messageExt.getBody(), Charset.forName(charset));
            if (Objects.equals(messageType, String.class)) {
                return str;
            } else {
                // If msgType not string, use objectMapper change it.
                try {
                    if (messageType instanceof Class) {
                        //if the messageType has not Generic Parameter
                        return this.getMessageConverter().fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) messageType);
                    } else {
                        //if the messageType has Generic Parameter, then use SmartMessageConverter#fromMessage with third parameter "conversionHint".
                        //we have validate the MessageConverter is SmartMessageConverter in this#getMethodParameter.
                        return ((SmartMessageConverter) this.getMessageConverter()).fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) ((ParameterizedType) messageType).getRawType(), methodParameter);
                    }
                } catch (Exception e) {
                    log.info("convert failed. str:{}, msgType:{}", str, messageType);
                    throw new RuntimeException("cannot convert message to " + messageType, e);
                }
            }
        }
    }
@Test
    public void testDoConvertMessage() throws Exception {
        DefaultRocketMQListenerContainer listenerContainer = new DefaultRocketMQListenerContainer();
        Method doConvertMessage = DefaultRocketMQListenerContainer.class.getDeclaredMethod("doConvertMessage", MessageExt.class);
        doConvertMessage.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);
        messageExt.setBody("hello".getBytes());
        String result = (String) doConvertMessage.invoke(listenerContainer, messageExt);
        assertThat(result).isEqualTo("hello");

        listenerContainer.setRocketMQListener(new RocketMQListener<MessageExt>() {
            @Override
            public void onMessage(MessageExt message) {
            }
        });
        Field messageType2 = DefaultRocketMQListenerContainer.class.getDeclaredField("messageType");
        messageType2.setAccessible(true);
        messageType2.set(listenerContainer, MessageExt.class);
        messageExt = new MessageExt(0, System.currentTimeMillis(), null, System.currentTimeMillis(), null, null);
        messageExt.setBody("hello".getBytes());
        MessageExt result2 = (MessageExt) doConvertMessage.invoke(listenerContainer, messageExt);
        assertThat(result2).isEqualTo(messageExt);

        //support message
        listenerContainer.setRocketMQListener(new RocketMQListener<Message>() {
            @Override
            public void onMessage(Message message) {
            }
        });
        Field messageType3 = DefaultRocketMQListenerContainer.class.getDeclaredField("messageType");
        messageType3.setAccessible(true);
        messageType3.set(listenerContainer, Message.class);
        Message message = new MessageExt(0, System.currentTimeMillis(), null, System.currentTimeMillis(), null, null);
        message.setBody("hello".getBytes());
        Message result3 = (Message) doConvertMessage.invoke(listenerContainer, message);
        assertThat(result3).isEqualTo(message);

        listenerContainer.setRocketMQListener(new RocketMQListener<User>() {
            @Override
            public void onMessage(User message) {
            }
        });

        listenerContainer.setRocketMQListener(new RocketMQListener<User>() {
            @Override
            public void onMessage(User message) {
            }
        });
    }