@Override
	protected MessageProducer createConsumerEndpoint(ConsumerDestination destination,
			String group,
			ExtendedConsumerProperties<RocketMQConsumerProperties> extendedConsumerProperties)
			throws Exception {
		boolean anonymous = !StringUtils.hasLength(group);
		/***
		 * 	When using DLQ, at least the group property must be provided for proper naming of the DLQ destination
		 *  According to https://docs.spring.io/spring-cloud-stream/docs/3.2.1/reference/html/spring-cloud-stream.html#spring-cloud-stream-reference
		 */
		if (anonymous && NamespaceUtil.isDLQTopic(destination.getName())) {
			throw new RuntimeException(
					"group must be configured for DLQ" + destination.getName());
		}
		group = anonymous ? RocketMQUtils.anonymousGroup(destination.getName()) : group;

		RocketMQUtils.mergeRocketMQProperties(binderConfigurationProperties,
				extendedConsumerProperties.getExtension());
		extendedConsumerProperties.getExtension().setGroup(group);

		RocketMQInboundChannelAdapter inboundChannelAdapter = new RocketMQInboundChannelAdapter(
				destination.getName(), extendedConsumerProperties);
		ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(destination,
				group, extendedConsumerProperties);
		if (extendedConsumerProperties.getMaxAttempts() > 1) {
			inboundChannelAdapter
					.setRetryTemplate(buildRetryTemplate(extendedConsumerProperties));
			inboundChannelAdapter.setRecoveryCallback(errorInfrastructure.getRecoverer());
		}
		else {
			inboundChannelAdapter.setErrorChannel(errorInfrastructure.getErrorChannel());
		}
		return inboundChannelAdapter;
	}
A {@link org.springframework.cloud.stream.binder.Binder} that uses RocketMQ as the
underlying middleware.

@author <a href="mailto:fangjian0423@gmail.com">Jim</a>
spring-cloud-alibaba-starters/spring-cloud-starter-stream-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java