Polish #145

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQBinderUtils.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright (C) 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.rocketmq;
+
+import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
+import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
+import org.springframework.util.StringUtils;
+
+/**
+ * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ */
+public class RocketMQBinderUtils {
+
+	public static RocketMQBinderConfigurationProperties mergeProducerProperties(
+			RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
+			RocketMQProperties rocketMQProperties) {
+		RocketMQBinderConfigurationProperties result = new RocketMQBinderConfigurationProperties();
+		if (StringUtils.isEmpty(rocketMQProperties.getNameServer())) {
+			result.setNameServer(rocketBinderConfigurationProperties.getNameServer());
+		}
+		else {
+			result.setNameServer(rocketMQProperties.getNameServer());
+		}
+		if (rocketMQProperties.getProducer() == null
+				|| StringUtils.isEmpty(rocketMQProperties.getProducer().getAccessKey())) {
+			result.setAccessKey(rocketBinderConfigurationProperties.getAccessKey());
+		}
+		else {
+			result.setAccessKey(rocketMQProperties.getProducer().getAccessKey());
+		}
+		if (rocketMQProperties.getProducer() == null
+				|| StringUtils.isEmpty(rocketMQProperties.getProducer().getSecretKey())) {
+			result.setSecretKey(rocketBinderConfigurationProperties.getSecretKey());
+		}
+		else {
+			result.setSecretKey(rocketMQProperties.getProducer().getSecretKey());
+		}
+		if (rocketMQProperties.getProducer() == null || StringUtils
+				.isEmpty(rocketMQProperties.getProducer().getCustomizedTraceTopic())) {
+			result.setCustomizedTraceTopic(
+					rocketBinderConfigurationProperties.getCustomizedTraceTopic());
+		}
+		else {
+			result.setCustomizedTraceTopic(
+					rocketMQProperties.getProducer().getCustomizedTraceTopic());
+		}
+		if (rocketMQProperties.getProducer() != null
+				&& rocketMQProperties.getProducer().isEnableMsgTrace()) {
+			result.setEnableMsgTrace(Boolean.TRUE);
+		}
+		else {
+			result.setEnableMsgTrace(
+					rocketBinderConfigurationProperties.isEnableMsgTrace());
+		}
+		return result;
+	}
+
+}

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java

@@ -21,10 +21,13 @@ import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
 
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.SessionCredentials;
 import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
 import org.apache.rocketmq.spring.core.RocketMQTemplate;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.rocketmq.spring.support.RocketMQUtil;
 import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
 import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
 import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
@@ -44,6 +47,7 @@ import org.springframework.cloud.stream.provisioning.ProducerDestination;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.util.StringUtils;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 
@@ -55,12 +59,10 @@ public class RocketMQMessageChannelBinder extends
 		implements
 		ExtendedPropertiesBinder<MessageChannel, RocketMQConsumerProperties, RocketMQProducerProperties> {
 
-	private static final Logger logger = LoggerFactory
-			.getLogger(RocketMQMessageChannelBinder.class);
-
 	private RocketMQExtendedBindingProperties extendedBindingProperties = new RocketMQExtendedBindingProperties();
 
 	private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
+	private final RocketMQProperties rocketMQProperties;
 	private final InstrumentationManager instrumentationManager;
 
 	private Set<String> clientConfigId = new HashSet<>();
@@ -69,10 +71,12 @@ public class RocketMQMessageChannelBinder extends
 	public RocketMQMessageChannelBinder(RocketMQTopicProvisioner provisioningProvider,
 			RocketMQExtendedBindingProperties extendedBindingProperties,
 			RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
+			RocketMQProperties rocketMQProperties,
 			InstrumentationManager instrumentationManager) {
 		super(null, provisioningProvider);
 		this.extendedBindingProperties = extendedBindingProperties;
 		this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
+		this.rocketMQProperties = rocketMQProperties;
 		this.instrumentationManager = instrumentationManager;
 	}
 
@@ -82,6 +86,10 @@ public class RocketMQMessageChannelBinder extends
 			MessageChannel errorChannel) throws Exception {
 		if (producerProperties.getExtension().getEnabled()) {
 
+			RocketMQBinderConfigurationProperties mergedProperties = RocketMQBinderUtils
+					.mergeProducerProperties(rocketBinderConfigurationProperties,
+							rocketMQProperties);
+
 			RocketMQTemplate rocketMQTemplate;
 			if (producerProperties.getExtension().getTransactional()) {
 				Map<String, RocketMQTemplate> rocketMQTemplates = getBeanFactory()
@@ -101,9 +109,27 @@ public class RocketMQMessageChannelBinder extends
 				rocketMQTemplate = new RocketMQTemplate();
 				rocketMQTemplate.setObjectMapper(this.getApplicationContext()
 						.getBeansOfType(ObjectMapper.class).values().iterator().next());
-				DefaultMQProducer producer = new DefaultMQProducer(destination.getName());
-				producer.setNamesrvAddr(
-						rocketBinderConfigurationProperties.getNamesrvAddr());
+				DefaultMQProducer producer;
+				String ak = mergedProperties.getAccessKey();
+				String sk = mergedProperties.getSecretKey();
+				if (!StringUtils.isEmpty(ak) && !StringUtils.isEmpty(sk)) {
+					RPCHook rpcHook = new AclClientRPCHook(
+							new SessionCredentials(ak, sk));
+					producer = new DefaultMQProducer(
+							producerProperties.getExtension().getGroup(), rpcHook,
+							mergedProperties.isEnableMsgTrace(),
+							mergedProperties.getCustomizedTraceTopic());
+					producer.setVipChannelEnabled(false);
+					producer.setInstanceName(
+							RocketMQUtil.getInstanceName(rpcHook, destination.getName()));
+				}
+				else {
+					producer = new DefaultMQProducer(
+							producerProperties.getExtension().getGroup());
+					producer.setVipChannelEnabled(
+							producerProperties.getExtension().getVipChannelEnabled());
+				}
+				producer.setNamesrvAddr(mergedProperties.getNameServer());
 				producer.setSendMsgTimeout(
 						producerProperties.getExtension().getSendMessageTimeout());
 				producer.setRetryTimesWhenSendFailed(
@@ -116,14 +142,13 @@ public class RocketMQMessageChannelBinder extends
 						producerProperties.getExtension().isRetryNextServer());
 				producer.setMaxMessageSize(
 						producerProperties.getExtension().getMaxMessageSize());
-				producer.setVipChannelEnabled(
-						producerProperties.getExtension().getVipChannelEnabled());
 				rocketMQTemplate.setProducer(producer);
 				clientConfigId.add(producer.buildMQClientId());
 			}
 
 			RocketMQMessageHandler messageHandler = new RocketMQMessageHandler(
 					rocketMQTemplate, destination.getName(),
+					producerProperties.getExtension().getGroup(),
 					producerProperties.getExtension().getTransactional(),
 					instrumentationManager);
 			messageHandler.setBeanFactory(this.getApplicationContext().getBeanFactory());
@@ -160,7 +185,7 @@ public class RocketMQMessageChannelBinder extends
 		listenerContainer.setDelayLevelWhenNextConsume(
 				consumerProperties.getExtension().getDelayLevelWhenNextConsume());
 		listenerContainer
-				.setNameServer(rocketBinderConfigurationProperties.getNamesrvAddr());
+				.setNameServer(rocketBinderConfigurationProperties.getNameServer());
 
 		RocketMQInboundChannelAdapter rocketInboundChannelAdapter = new RocketMQInboundChannelAdapter(
 				listenerContainer, consumerProperties, instrumentationManager);

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderAutoConfiguration.java

@@ -16,6 +16,8 @@
 
 package org.springframework.cloud.stream.binder.rocketmq.config;
 
+import org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration;
+import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
@@ -39,7 +41,8 @@ import io.micrometer.core.instrument.binder.MeterBinder;
  * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
  */
 @Configuration
-@Import(RocketMQBinderHealthIndicatorAutoConfiguration.class)
+@Import({ RocketMQAutoConfiguration.class,
+		RocketMQBinderHealthIndicatorAutoConfiguration.class })
 @EnableConfigurationProperties({ RocketMQBinderConfigurationProperties.class,
 		RocketMQExtendedBindingProperties.class })
 public class RocketMQBinderAutoConfiguration {
@@ -48,6 +51,9 @@ public class RocketMQBinderAutoConfiguration {
 
 	private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
 
+	@Autowired(required = false)
+	private RocketMQProperties rocketMQProperties = new RocketMQProperties();
+
 	@Autowired
 	public RocketMQBinderAutoConfiguration(
 			RocketMQExtendedBindingProperties extendedBindingProperties,
@@ -66,8 +72,9 @@ public class RocketMQBinderAutoConfiguration {
 			RocketMQTopicProvisioner provisioningProvider,
 			InstrumentationManager instrumentationManager) {
 		RocketMQMessageChannelBinder binder = new RocketMQMessageChannelBinder(
-                provisioningProvider, extendedBindingProperties,
-				rocketBinderConfigurationProperties, instrumentationManager);
+				provisioningProvider, extendedBindingProperties,
+				rocketBinderConfigurationProperties, rocketMQProperties,
+				instrumentationManager);
 		binder.setExtendedBindingProperties(extendedBindingProperties);
 		return binder;
 	}

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQComponent4BinderAutoConfiguration.java

@@ -16,11 +16,10 @@
 
 package org.springframework.cloud.stream.binder.rocketmq.config;
 
-import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.SessionCredentials;
 import org.apache.rocketmq.client.producer.DefaultMQProducer;
 import org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration;
-import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
-import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties.Producer;
 import org.apache.rocketmq.spring.config.RocketMQConfigUtils;
 import org.apache.rocketmq.spring.config.RocketMQTransactionAnnotationProcessor;
 import org.apache.rocketmq.spring.config.TransactionHandlerRegistry;
@@ -32,7 +31,7 @@ import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.env.Environment;
-import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 
@@ -51,48 +50,34 @@ public class RocketMQComponent4BinderAutoConfiguration {
 	}
 
 	@Bean
+	@ConditionalOnMissingBean(DefaultMQProducer.class)
 	public DefaultMQProducer defaultMQProducer() {
-		RocketMQProperties rocketMQProperties = new RocketMQProperties();
-		String configNameServer = environment
-				.getProperty("spring.cloud.stream.rocketmq.binder.namesrv-addr");
-		if (StringUtils.isEmpty(configNameServer)) {
-			rocketMQProperties.setNameServer(RocketMQBinderConstants.DEFAULT_NAME_SERVER);
+		DefaultMQProducer producer;
+		String configNameServer = environment.resolveRequiredPlaceholders(
+				"${spring.cloud.stream.rocketmq.binder.name-server:${rocketmq.producer.name-server:}}");
+		String ak = environment.resolveRequiredPlaceholders(
+				"${spring.cloud.stream.rocketmq.binder.access-key:${rocketmq.producer.access-key:}}");
+		String sk = environment.resolveRequiredPlaceholders(
+				"${spring.cloud.stream.rocketmq.binder.secret-key:${rocketmq.producer.secret-key:}}");
+		if (!StringUtils.isEmpty(ak) && !StringUtils.isEmpty(sk)) {
+			producer = new DefaultMQProducer(RocketMQBinderConstants.DEFAULT_GROUP,
+					new AclClientRPCHook(new SessionCredentials(ak, sk)));
+			producer.setVipChannelEnabled(false);
 		}
 		else {
-			rocketMQProperties.setNameServer(configNameServer);
+			producer = new DefaultMQProducer(RocketMQBinderConstants.DEFAULT_GROUP);
 		}
-		RocketMQProperties.Producer producerConfig = new Producer();
-		rocketMQProperties.setProducer(producerConfig);
-		producerConfig.setGroup(RocketMQBinderConstants.DEFAULT_GROUP);
-
-		String nameServer = rocketMQProperties.getNameServer();
-		String groupName = producerConfig.getGroup();
-		Assert.hasText(nameServer, "[rocketmq.name-server] must not be null");
-		Assert.hasText(groupName, "[rocketmq.producer.group] must not be null");
-
-		DefaultMQProducer producer = new DefaultMQProducer(groupName);
-		producer.setNamesrvAddr(nameServer);
-		producer.setSendMsgTimeout(producerConfig.getSendMessageTimeout());
-		producer.setRetryTimesWhenSendFailed(
-				producerConfig.getRetryTimesWhenSendFailed());
-		producer.setRetryTimesWhenSendAsyncFailed(
-				producerConfig.getRetryTimesWhenSendAsyncFailed());
-		producer.setMaxMessageSize(producerConfig.getMaxMessageSize());
-		producer.setCompressMsgBodyOverHowmuch(
-				producerConfig.getCompressMessageBodyThreshold());
-		producer.setRetryAnotherBrokerWhenNotStoreOK(producerConfig.isRetryNextServer());
-
+		producer.setNamesrvAddr(configNameServer);
 		return producer;
 	}
 
 	@Bean(destroyMethod = "destroy")
-	@ConditionalOnBean(DefaultMQProducer.class)
-	@ConditionalOnMissingBean(RocketMQTemplate.class)
+	@ConditionalOnMissingBean
 	public RocketMQTemplate rocketMQTemplate(DefaultMQProducer mqProducer,
-			ObjectMapper rocketMQMessageObjectMapper) {
+			ObjectMapper objectMapper) {
 		RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
 		rocketMQTemplate.setProducer(mqProducer);
-		rocketMQTemplate.setObjectMapper(rocketMQMessageObjectMapper);
+		rocketMQTemplate.setObjectMapper(objectMapper);
 		return rocketMQTemplate;
 	}
 

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/consuming/RocketMQListenerBindingContainer.java

@@ -19,6 +19,8 @@ package org.springframework.cloud.stream.binder.rocketmq.consuming;
 import java.util.List;
 import java.util.Objects;
 
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.SessionCredentials;
 import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
 import org.apache.rocketmq.client.consumer.MessageSelector;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
@@ -27,8 +29,10 @@ import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
 import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
 import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
 import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
+import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.remoting.RPCHook;
 import org.apache.rocketmq.spring.annotation.ConsumeMode;
 import org.apache.rocketmq.spring.annotation.MessageModel;
 import org.apache.rocketmq.spring.annotation.SelectorType;
@@ -41,6 +45,7 @@ import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
 import org.springframework.cloud.stream.binder.rocketmq.RocketMQMessageChannelBinder;
+import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
 import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.util.Assert;
@@ -84,6 +89,7 @@ public class RocketMQListenerBindingContainer
 	private final ExtendedConsumerProperties<RocketMQConsumerProperties> rocketMQConsumerProperties;
 
 	private final RocketMQMessageChannelBinder rocketMQMessageChannelBinder;
+	private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
 
 	// The following properties came from RocketMQConsumerProperties.
 	private ConsumeMode consumeMode;
@@ -93,8 +99,10 @@ public class RocketMQListenerBindingContainer
 
 	public RocketMQListenerBindingContainer(
 			ExtendedConsumerProperties<RocketMQConsumerProperties> rocketMQConsumerProperties,
+			RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
 			RocketMQMessageChannelBinder rocketMQMessageChannelBinder) {
 		this.rocketMQConsumerProperties = rocketMQConsumerProperties;
+		this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
 		this.rocketMQMessageChannelBinder = rocketMQMessageChannelBinder;
 		this.consumeMode = rocketMQConsumerProperties.getExtension().getOrderly()
 				? ConsumeMode.ORDERLY
@@ -190,7 +198,23 @@ public class RocketMQListenerBindingContainer
 		Assert.notNull(nameServer, "Property 'nameServer' is required");
 		Assert.notNull(topic, "Property 'topic' is required");
 
-		consumer = new DefaultMQPushConsumer(consumerGroup);
+		String ak = rocketBinderConfigurationProperties.getAccessKey();
+		String sk = rocketBinderConfigurationProperties.getSecretKey();
+		if (!StringUtils.isEmpty(ak) && !StringUtils.isEmpty(sk)) {
+			RPCHook rpcHook = new AclClientRPCHook(new SessionCredentials(ak, sk));
+			consumer = new DefaultMQPushConsumer(consumerGroup, rpcHook,
+					new AllocateMessageQueueAveragely(),
+					rocketBinderConfigurationProperties.isEnableMsgTrace(),
+					rocketBinderConfigurationProperties.getCustomizedTraceTopic());
+			consumer.setInstanceName(RocketMQUtil.getInstanceName(rpcHook, topic));
+			consumer.setVipChannelEnabled(false);
+		}
+		else {
+			consumer = new DefaultMQPushConsumer(consumerGroup,
+					rocketBinderConfigurationProperties.isEnableMsgTrace(),
+					rocketBinderConfigurationProperties.getCustomizedTraceTopic());
+		}
+
 		consumer.setNamesrvAddr(nameServer);
 		consumer.setConsumeThreadMax(rocketMQConsumerProperties.getConcurrency());
 		consumer.setConsumeThreadMin(rocketMQConsumerProperties.getConcurrency());
@@ -236,7 +260,15 @@ public class RocketMQListenerBindingContainer
 		}
 
 		rocketMQMessageChannelBinder.getClientConfigId().add(consumer.buildMQClientId());
+	}
 
+	@Override
+	public String toString() {
+		return "RocketMQListenerBindingContainer{" + "consumerGroup='" + consumerGroup
+				+ '\'' + ", nameServer='" + nameServer + '\'' + ", topic='" + topic + '\''
+				+ ", consumeMode=" + consumeMode + ", selectorType=" + selectorType
+				+ ", selectorExpression='" + selectorExpression + '\'' + ", messageModel="
+				+ messageModel + '}';
 	}
 
 	public long getSuspendCurrentQueueTimeMillis() {

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQInboundChannelAdapter.java

@@ -40,7 +40,7 @@ import org.springframework.util.Assert;
  */
 public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 
-	private static final Logger logger = LoggerFactory
+	private static final Logger log = LoggerFactory
 			.getLogger(RocketMQInboundChannelAdapter.class);
 
 	private RetryTemplate retryTemplate;
@@ -88,7 +88,7 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 
 		}
 		catch (Exception e) {
-			logger.error("rocketMQListenerContainer init error: " + e.getMessage(), e);
+			log.error("rocketMQListenerContainer init error: " + e.getMessage(), e);
 			throw new IllegalArgumentException(
 					"rocketMQListenerContainer init error: " + e.getMessage(), e);
 		}
@@ -116,7 +116,7 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 					.getHealthInstrumentation(rocketMQListenerContainer.getTopic()
 							+ rocketMQListenerContainer.getConsumerGroup())
 					.markStartFailed(e);
-			logger.error("RocketMQTemplate startup failed, Caused by " + e.getMessage());
+			log.error("RocketMQTemplate startup failed, Caused by " + e.getMessage());
 			throw new MessagingException(MessageBuilder.withPayload(
 					"RocketMQTemplate startup failed, Caused by " + e.getMessage())
 					.build(), e);

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQMessageHandler.java

@@ -25,6 +25,8 @@ import org.apache.rocketmq.client.producer.SendStatus;
 import org.apache.rocketmq.common.message.MessageConst;
 import org.apache.rocketmq.spring.core.RocketMQTemplate;
 import org.apache.rocketmq.spring.support.RocketMQHeaders;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants;
 import org.springframework.cloud.stream.binder.rocketmq.metrics.Instrumentation;
 import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
@@ -44,6 +46,9 @@ import org.springframework.util.StringUtils;
  */
 public class RocketMQMessageHandler extends AbstractMessageHandler implements Lifecycle {
 
+	private final static Logger log = LoggerFactory
+			.getLogger(RocketMQMessageHandler.class);
+
 	private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy();
 
 	private MessageChannel sendFailureChannel;
@@ -54,6 +59,8 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
 
 	private final String destination;
 
+	private final String groupName;
+
 	private final InstrumentationManager instrumentationManager;
 
 	private boolean sync = false;
@@ -61,9 +68,11 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
 	private volatile boolean running = false;
 
 	public RocketMQMessageHandler(RocketMQTemplate rocketMQTemplate, String destination,
-			Boolean transactional, InstrumentationManager instrumentationManager) {
+			String groupName, Boolean transactional,
+			InstrumentationManager instrumentationManager) {
 		this.rocketMQTemplate = rocketMQTemplate;
 		this.destination = destination;
+		this.groupName = groupName;
 		this.transactional = transactional;
 		this.instrumentationManager = instrumentationManager;
 	}
@@ -81,8 +90,7 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
 			catch (Exception e) {
 				instrumentationManager.getHealthInstrumentation(destination)
 						.markStartFailed(e);
-				logger.error(
-						"RocketMQTemplate startup failed, Caused by " + e.getMessage());
+				log.error("RocketMQTemplate startup failed, Caused by " + e.getMessage());
 				throw new MessagingException(MessageBuilder.withPayload(
 						"RocketMQTemplate startup failed, Caused by " + e.getMessage())
 						.build(), e);
@@ -108,18 +116,19 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
 	protected void handleMessageInternal(org.springframework.messaging.Message<?> message)
 			throws Exception {
 		try {
-			StringBuilder topicWithTags = new StringBuilder(destination);
+			final StringBuilder topicWithTags = new StringBuilder(destination);
 			String tags = Optional
 					.ofNullable(message.getHeaders().get(RocketMQHeaders.TAGS)).orElse("")
 					.toString();
 			if (!StringUtils.isEmpty(tags)) {
-				topicWithTags = topicWithTags.append(":").append(tags);
+				topicWithTags.append(":").append(tags);
 			}
 			SendResult sendRes = null;
 			if (transactional) {
-				sendRes = rocketMQTemplate.sendMessageInTransaction(destination,
+				sendRes = rocketMQTemplate.sendMessageInTransaction(groupName,
 						topicWithTags.toString(), message, message.getHeaders()
 								.get(RocketMQBinderConstants.ROCKET_TRANSACTIONAL_ARG));
+				log.debug("transactional send to topic " + topicWithTags + " " + sendRes);
 			}
 			else {
 				int delayLevel = 0;
@@ -140,17 +149,22 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
 					sendRes = rocketMQTemplate.syncSend(topicWithTags.toString(), message,
 							rocketMQTemplate.getProducer().getSendMsgTimeout(),
 							delayLevel);
+					log.debug("sync send to topic " + topicWithTags + " " + sendRes);
 				}
 				else {
 					rocketMQTemplate.asyncSend(topicWithTags.toString(), message,
 							new SendCallback() {
 								@Override
 								public void onSuccess(SendResult sendResult) {
-
+									log.debug("async send to topic " + topicWithTags + " "
+											+ sendResult);
 								}
 
 								@Override
 								public void onException(Throwable e) {
+									log.error(
+											"RocketMQ Message hasn't been sent. Caused by "
+													+ e.getMessage());
 									if (getSendFailureChannel() != null) {
 										getSendFailureChannel().send(
 												RocketMQMessageHandler.this.errorMessageStrategy
@@ -174,8 +188,7 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
 			}
 		}
 		catch (Exception e) {
-			logger.error(
-					"RocketMQ Message hasn't been sent. Caused by " + e.getMessage());
+			log.error("RocketMQ Message hasn't been sent. Caused by " + e.getMessage());
 			if (getSendFailureChannel() != null) {
 				getSendFailureChannel().send(this.errorMessageStrategy
 						.buildErrorMessage(new MessagingException(message, e), null));

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/RocketMQBinderMetrics.java

@@ -59,9 +59,9 @@ public class RocketMQBinderMetrics
 	public void bindTo(@NonNull MeterRegistry registry) {
 		DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer();
 		pushConsumer
-				.setNamesrvAddr(rocketMQBinderConfigurationProperties.getNamesrvAddr());
+				.setNamesrvAddr(rocketMQBinderConfigurationProperties.getNameServer());
 		DefaultMQProducer producer = new DefaultMQProducer();
-		producer.setNamesrvAddr(rocketMQBinderConfigurationProperties.getNamesrvAddr());
+		producer.setNamesrvAddr(rocketMQBinderConfigurationProperties.getNameServer());
 
 		rocketMQMessageChannelBinder.getTopicInUse().forEach((topic, group) -> {
 			Gauge.builder(METRIC_NAME, this, o -> calculateMsgQueueOffset(topic, group))

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQBinderConfigurationProperties.java

@@ -16,6 +16,7 @@
 
 package org.springframework.cloud.stream.binder.rocketmq.properties;
 
+import org.apache.rocketmq.common.MixAll;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants;
 
@@ -26,14 +27,69 @@ import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants;
 @ConfigurationProperties(prefix = "spring.cloud.stream.rocketmq.binder")
 public class RocketMQBinderConfigurationProperties {
 
-	private String namesrvAddr = RocketMQBinderConstants.DEFAULT_NAME_SERVER;
+	/**
+	 * The name server for rocketMQ, formats: `host:port;host:port`.
+	 */
+	private String nameServer = RocketMQBinderConstants.DEFAULT_NAME_SERVER;
 
-	public String getNamesrvAddr() {
-		return namesrvAddr;
+	/**
+	 * The property of "access-key".
+	 */
+	private String accessKey;
+
+	/**
+	 * The property of "secret-key".
+	 */
+	private String secretKey;
+
+	/**
+	 * Switch flag instance for message trace.
+	 */
+	private boolean enableMsgTrace = true;
+
+	/**
+	 * The name value of message trace topic.If you don't config,you can use the default
+	 * trace topic name.
+	 */
+	private String customizedTraceTopic = MixAll.RMQ_SYS_TRACE_TOPIC;
+
+	public String getNameServer() {
+		return nameServer;
 	}
 
-	public void setNamesrvAddr(String namesrvAddr) {
-		this.namesrvAddr = namesrvAddr;
+	public void setNameServer(String nameServer) {
+		this.nameServer = nameServer;
 	}
 
+	public String getAccessKey() {
+		return accessKey;
+	}
+
+	public void setAccessKey(String accessKey) {
+		this.accessKey = accessKey;
+	}
+
+	public String getSecretKey() {
+		return secretKey;
+	}
+
+	public void setSecretKey(String secretKey) {
+		this.secretKey = secretKey;
+	}
+
+	public boolean isEnableMsgTrace() {
+		return enableMsgTrace;
+	}
+
+	public void setEnableMsgTrace(boolean enableMsgTrace) {
+		this.enableMsgTrace = enableMsgTrace;
+	}
+
+	public String getCustomizedTraceTopic() {
+		return customizedTraceTopic;
+	}
+
+	public void setCustomizedTraceTopic(String customizedTraceTopic) {
+		this.customizedTraceTopic = customizedTraceTopic;
+	}
 }

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQProducerProperties.java

@@ -27,7 +27,12 @@ public class RocketMQProducerProperties {
 	private Boolean enabled = true;
 
 	/**
-	 * Maximum allowed message size in bytes {@link DefaultMQProducer#maxMessageSize}
+	 * Name of producer.
+	 */
+	private String group;
+
+	/**
+	 * Maximum allowed message size in bytes {@link DefaultMQProducer#maxMessageSize}.
 	 */
 	private Integer maxMessageSize = 1024 * 1024 * 4;
 
@@ -70,6 +75,14 @@ public class RocketMQProducerProperties {
 	 */
 	private boolean retryNextServer = false;
 
+	public String getGroup() {
+		return group;
+	}
+
+	public void setGroup(String group) {
+		this.group = group;
+	}
+
 	public Boolean getEnabled() {
 		return enabled;
 	}

spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/provisioning/RocketMQTopicProvisioner.java

@@ -18,8 +18,6 @@ package org.springframework.cloud.stream.binder.rocketmq.provisioning;
 
 import org.apache.rocketmq.client.Validators;
 import org.apache.rocketmq.client.exception.MQClientException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
 import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
 import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
@@ -36,9 +34,6 @@ import org.springframework.cloud.stream.provisioning.ProvisioningProvider;
 public class RocketMQTopicProvisioner implements
 		ProvisioningProvider<ExtendedConsumerProperties<RocketMQConsumerProperties>, ExtendedProducerProperties<RocketMQProducerProperties>> {
 
-	private static final Logger logger = LoggerFactory
-			.getLogger(RocketMQTopicProvisioner.class);
-
 	@Override
 	public ProducerDestination provisionProducerDestination(String name,
 			ExtendedProducerProperties<RocketMQProducerProperties> properties)

spring-cloud-stream-binder-rocketmq/src/test/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQAutoConfigurationTests.java

@@ -53,7 +53,7 @@ public class RocketMQAutoConfigurationTests {
 		this.contextRunner.run(context -> {
 			RocketMQBinderConfigurationProperties binderConfigurationProperties = context
 					.getBean(RocketMQBinderConfigurationProperties.class);
-			assertThat(binderConfigurationProperties.getNamesrvAddr())
+			assertThat(binderConfigurationProperties.getNameServer())
 					.isEqualTo("127.0.0.1:9876");
 			RocketMQExtendedBindingProperties bindingProperties = context
 					.getBean(RocketMQExtendedBindingProperties.class);