make metrics-core as provided dependency and refactor

pom.xml

@@ -100,7 +100,6 @@
         <module>spring-cloud-alicloud-oss</module>
         <module>spring-cloud-alicloud-acm</module>
         <module>spring-cloud-alicloud-ans</module>
-        <module>spring-cloud-starter-bus-rocketmq</module>
     </modules>
 
     <dependencyManagement>

spring-cloud-alibaba-examples/rocketmq-example/pom.xml

@@ -31,6 +31,12 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-actuator</artifactId>
         </dependency>
+
+        <dependency>
+            <groupId>io.dropwizard.metrics</groupId>
+            <artifactId>metrics-core</artifactId>
+        </dependency>
+
     </dependencies>
 
     <build>

spring-cloud-stream-binder-rocketmq/pom.xml

@@ -22,13 +22,16 @@
         </dependency>
 
         <dependency>
-            <groupId>io.dropwizard.metrics</groupId>
+            <groupId>org.apache.rocketmq</groupId>
-            <artifactId>metrics-core</artifactId>
+            <artifactId>rocketmq-client</artifactId>
         </dependency>
 
         <dependency>
-            <groupId>org.apache.rocketmq</groupId>
+            <groupId>io.dropwizard.metrics</groupId>
-            <artifactId>rocketmq-client</artifactId>
+            <artifactId>metrics-core</artifactId>
+            <version>4.0.3</version>
+            <scope>provided</scope>
+            <optional>true</optional>
         </dependency>
 
         <dependency>

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

@@ -34,7 +34,6 @@ public class RocketMQMessageChannelBinder extends
 			.getLogger(RocketMQMessageChannelBinder.class);
 
 	private final RocketMQExtendedBindingProperties extendedBindingProperties;
-	private final RocketMQTopicProvisioner rocketTopicProvisioner;
 	private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
 	private final InstrumentationManager instrumentationManager;
 	private final ConsumersManager consumersManager;
@@ -47,7 +46,6 @@ public class RocketMQMessageChannelBinder extends
 		super(null, provisioningProvider);
 		this.consumersManager = consumersManager;
 		this.extendedBindingProperties = extendedBindingProperties;
-		this.rocketTopicProvisioner = provisioningProvider;
 		this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
 		this.instrumentationManager = instrumentationManager;
 	}
@@ -63,7 +61,7 @@ public class RocketMQMessageChannelBinder extends
 		}
 		else {
 			throw new RuntimeException("Binding for channel " + destination.getName()
-					+ "has been disabled, message can't be delivered");
+					+ " has been disabled, message can't be delivered");
 		}
 	}
 
@@ -74,7 +72,7 @@ public class RocketMQMessageChannelBinder extends
 			throws Exception {
 		if (group == null || "".equals(group)) {
 			throw new RuntimeException(
-					"'group' must be configured for channel + " + destination.getName());
+					"'group must be configured for channel + " + destination.getName());
 		}
 
 		RocketMQInboundChannelAdapter rocketInboundChannelAdapter = new RocketMQInboundChannelAdapter(

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

@@ -4,12 +4,11 @@ import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderCon
 
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
 
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.actuate.endpoint.annotation.Endpoint;
 import org.springframework.boot.actuate.endpoint.annotation.ReadOperation;
-
+import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
-import com.codahale.metrics.MetricRegistry;
 
 /**
  * @author Timur Valiev
@@ -18,23 +17,22 @@ import com.codahale.metrics.MetricRegistry;
 @Endpoint(id = ENDPOINT_ID)
 public class RocketMQBinderEndpoint {
 
-	private MetricRegistry metricRegistry = new MetricRegistry();
+	@Autowired(required = false)
-	private Map<String, Object> runtime = new ConcurrentHashMap<>();
+	private InstrumentationManager instrumentationManager;
 
 	@ReadOperation
 	public Map<String, Object> invoke() {
 		Map<String, Object> result = new HashMap<>();
-		result.put("metrics", metricRegistry().getMetrics());
+		if (instrumentationManager != null) {
-		result.put("runtime", runtime());
+			result.put("metrics",
+					instrumentationManager.getMetricRegistry().getMetrics());
+			result.put("runtime", instrumentationManager.getRuntime());
+		}
+		else {
+			result.put("warning",
+					"please add metrics-core dependency, we use it for metrics");
+		}
 		return result;
 	}
 
-	public MetricRegistry metricRegistry() {
-		return metricRegistry;
-	}
-
-	public Map<String, Object> runtime() {
-		return runtime;
-	}
-
 }

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

@@ -1,5 +1,6 @@
 package org.springframework.cloud.stream.binder.rocketmq.actuator;
 
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.actuate.health.AbstractHealthIndicator;
 import org.springframework.boot.actuate.health.Health;
 import org.springframework.cloud.stream.binder.rocketmq.metrics.Instrumentation;
@@ -11,28 +12,33 @@ import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationM
  */
 public class RocketMQBinderHealthIndicator extends AbstractHealthIndicator {
 
-	private final InstrumentationManager instrumentationManager;
+	@Autowired(required = false)
-
+	private InstrumentationManager instrumentationManager;
-	public RocketMQBinderHealthIndicator(InstrumentationManager instrumentationManager) {
-		this.instrumentationManager = instrumentationManager;
-	}
 
 	@Override
 	protected void doHealthCheck(Health.Builder builder) throws Exception {
-		if (instrumentationManager.getHealthInstrumentations().stream()
+		if (instrumentationManager != null) {
-				.allMatch(Instrumentation::isUp)) {
+			if (instrumentationManager.getHealthInstrumentations().stream()
-			builder.up();
+					.allMatch(Instrumentation::isUp)) {
-			return;
+				builder.up();
+				return;
+			}
+			if (instrumentationManager.getHealthInstrumentations().stream()
+					.allMatch(Instrumentation::isOutOfService)) {
+				builder.outOfService();
+				return;
+			}
+			builder.down();
+			instrumentationManager.getHealthInstrumentations().stream()
+					.filter(instrumentation -> !instrumentation.isStarted())
+					.forEach(instrumentation1 -> builder
+							.withException(instrumentation1.getStartException()));
 		}
-		if (instrumentationManager.getHealthInstrumentations().stream()
+		else {
-				.allMatch(Instrumentation::isOutOfService)) {
+			builder.down();
-			builder.outOfService();
+			builder.withDetail("warning",
-			return;
+					"please add metrics-core dependency, we use it for metrics");
 		}
-		builder.down();
+
-		instrumentationManager.getHealthInstrumentations().stream()
-				.filter(instrumentation -> !instrumentation.isStarted())
-				.forEach(instrumentation1 -> builder
-						.withException(instrumentation1.getStartException()));
 	}
 }

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

@@ -25,6 +25,9 @@ public class RocketMQBinderAutoConfiguration {
 
 	private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
 
+	@Autowired(required = false)
+	private InstrumentationManager instrumentationManager;
+
 	@Autowired
 	public RocketMQBinderAutoConfiguration(
 			RocketMQExtendedBindingProperties extendedBindingProperties,
@@ -43,7 +46,6 @@ public class RocketMQBinderAutoConfiguration {
 	@Bean
 	public RocketMQMessageChannelBinder rocketMessageChannelBinder(
 			RocketMQTopicProvisioner provisioningProvider,
-			InstrumentationManager instrumentationManager,
 			ConsumersManager consumersManager) {
 		RocketMQMessageChannelBinder binder = new RocketMQMessageChannelBinder(
 				consumersManager, extendedBindingProperties, provisioningProvider,
@@ -52,8 +54,7 @@ public class RocketMQBinderAutoConfiguration {
 	}
 
 	@Bean
-	public ConsumersManager consumersManager(
+	public ConsumersManager consumersManager() {
-			InstrumentationManager instrumentationManager) {
 		return new ConsumersManager(instrumentationManager,
 				rocketBinderConfigurationProperties);
 	}

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

@@ -1,7 +1,9 @@
 package org.springframework.cloud.stream.binder.rocketmq.config;
 
 import org.springframework.boot.actuate.autoconfigure.endpoint.EndpointAutoConfiguration;
+import org.springframework.boot.actuate.endpoint.annotation.Endpoint;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.cloud.stream.binder.rocketmq.actuator.RocketMQBinderEndpoint;
 import org.springframework.cloud.stream.binder.rocketmq.actuator.RocketMQBinderHealthIndicator;
 import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
@@ -13,6 +15,7 @@ import org.springframework.context.annotation.Configuration;
  */
 @Configuration
 @AutoConfigureAfter(EndpointAutoConfiguration.class)
+@ConditionalOnClass(Endpoint.class)
 public class RocketMQBinderEndpointAutoConfiguration {
 
 	@Bean
@@ -21,16 +24,14 @@ public class RocketMQBinderEndpointAutoConfiguration {
 	}
 
 	@Bean
-	public RocketMQBinderHealthIndicator rocketBinderHealthIndicator(
+	public RocketMQBinderHealthIndicator rocketBinderHealthIndicator() {
-			InstrumentationManager instrumentationManager) {
+		return new RocketMQBinderHealthIndicator();
-		return new RocketMQBinderHealthIndicator(instrumentationManager);
 	}
 
 	@Bean
-	public InstrumentationManager instrumentationManager(
+	@ConditionalOnClass(name = "com.codahale.metrics.Counter")
-			RocketMQBinderEndpoint rocketBinderEndpoint) {
+	public InstrumentationManager instrumentationManager() {
-		return new InstrumentationManager(rocketBinderEndpoint.metricRegistry(),
+		return new InstrumentationManager();
-				rocketBinderEndpoint.runtime());
 	}
 
 }

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

@@ -3,6 +3,7 @@ package org.springframework.cloud.stream.binder.rocketmq.consuming;
 import java.util.AbstractMap;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 
 import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
@@ -27,9 +28,10 @@ public class ConsumersManager {
 	private final Map<String, DefaultMQPushConsumer> consumerGroups = new HashMap<>();
 	private final Map<String, Boolean> started = new HashMap<>();
 	private final Map<Map.Entry<String, String>, ExtendedConsumerProperties<RocketMQConsumerProperties>> propertiesMap = new HashMap<>();
-	private final InstrumentationManager instrumentationManager;
 	private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
 
+	private InstrumentationManager instrumentationManager;
+
 	public ConsumersManager(InstrumentationManager instrumentationManager,
 			RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties) {
 		this.instrumentationManager = instrumentationManager;
@@ -41,9 +43,12 @@ public class ConsumersManager {
 			ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties) {
 		propertiesMap.put(new AbstractMap.SimpleEntry<>(group, topic),
 				consumerProperties);
-		ConsumerGroupInstrumentation instrumentation = instrumentationManager
+
-				.getConsumerGroupInstrumentation(group);
+		Optional.ofNullable(instrumentationManager).ifPresent(manager -> {
-		instrumentationManager.addHealthInstrumentation(instrumentation);
+			ConsumerGroupInstrumentation instrumentation = manager
+					.getConsumerGroupInstrumentation(group);
+			instrumentationManager.addHealthInstrumentation(instrumentation);
+		});
 
 		if (consumerGroups.containsKey(group)) {
 			return consumerGroups.get(group);
@@ -87,16 +92,23 @@ public class ConsumersManager {
 		if (started.get(group)) {
 			return;
 		}
-		ConsumerGroupInstrumentation groupInstrumentation = instrumentationManager
+
-				.getConsumerGroupInstrumentation(group);
+		ConsumerGroupInstrumentation groupInstrumentation = null;
-		instrumentationManager.addHealthInstrumentation(groupInstrumentation);
+		if (Optional.ofNullable(instrumentationManager).isPresent()) {
+			groupInstrumentation = instrumentationManager
+					.getConsumerGroupInstrumentation(group);
+			instrumentationManager.addHealthInstrumentation(groupInstrumentation);
+		}
+
 		try {
 			consumerGroups.get(group).start();
 			started.put(group, true);
-			groupInstrumentation.markStartedSuccessfully();
+			Optional.ofNullable(groupInstrumentation)
+					.ifPresent(g -> g.markStartedSuccessfully());
 		}
 		catch (MQClientException e) {
-			groupInstrumentation.markStartFailed(e);
+			Optional.ofNullable(groupInstrumentation)
+					.ifPresent(g -> g.markStartFailed(e));
 			logger.error("RocketMQ Consumer hasn't been started. Caused by "
 					+ e.getErrorMessage(), e);
 			throw e;

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

@@ -4,6 +4,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -48,20 +49,20 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 
 	private ConsumerInstrumentation consumerInstrumentation;
 
+	private InstrumentationManager instrumentationManager;
+
+	private RetryTemplate retryTemplate;
+
+	private RecoveryCallback<? extends Object> recoveryCallback;
+
 	private final ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties;
 
 	private final String destination;
 
 	private final String group;
 
-	private final InstrumentationManager instrumentationManager;
-
 	private final ConsumersManager consumersManager;
 
-	private RetryTemplate retryTemplate;
-
-	private RecoveryCallback<? extends Object> recoveryCallback;
-
 	public RocketMQInboundChannelAdapter(ConsumersManager consumersManager,
 			ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties,
 			String destination, String group,
@@ -75,21 +76,20 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 
 	@Override
 	protected void doStart() {
-		if (!consumerProperties.getExtension().getEnabled()) {
+		if (consumerProperties == null
+				|| !consumerProperties.getExtension().getEnabled()) {
 			return;
 		}
 
-		String tags = consumerProperties == null ? null
+		String tags = consumerProperties.getExtension().getTags();
-				: consumerProperties.getExtension().getTags();
+		Boolean isOrderly = consumerProperties.getExtension().getOrderly();
-		Boolean isOrderly = consumerProperties == null ? false
-				: consumerProperties.getExtension().getOrderly();
 
 		DefaultMQPushConsumer consumer = consumersManager.getOrCreateConsumer(group,
 				destination, consumerProperties);
 
 		final CloudStreamMessageListener listener = isOrderly
-				? new CloudStreamMessageListenerOrderly(instrumentationManager)
+				? new CloudStreamMessageListenerOrderly()
-				: new CloudStreamMessageListenerConcurrently(instrumentationManager);
+				: new CloudStreamMessageListenerConcurrently();
 
 		if (retryTemplate != null) {
 			retryTemplate.registerListener(listener);
@@ -99,9 +99,10 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 				: Arrays.stream(tags.split("\\|\\|")).map(String::trim)
 						.collect(Collectors.toSet());
 
-		consumerInstrumentation = instrumentationManager
+		Optional.ofNullable(instrumentationManager).ifPresent(manager -> {
-				.getConsumerInstrumentation(destination);
+			consumerInstrumentation = manager.getConsumerInstrumentation(destination);
-		instrumentationManager.addHealthInstrumentation(consumerInstrumentation);
+			manager.addHealthInstrumentation(consumerInstrumentation);
+		});
 
 		try {
 			if (!StringUtils.isEmpty(consumerProperties.getExtension().getSql())) {
@@ -111,10 +112,12 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 			else {
 				consumer.subscribe(destination, String.join(" || ", tagsSet));
 			}
-			consumerInstrumentation.markStartedSuccessfully();
+			Optional.ofNullable(consumerInstrumentation)
+					.ifPresent(c -> c.markStartedSuccessfully());
 		}
 		catch (MQClientException e) {
-			consumerInstrumentation.markStartFailed(e);
+			Optional.ofNullable(consumerInstrumentation)
+					.ifPresent(c -> c.markStartFailed(e));
 			logger.error("RocketMQ Consumer hasn't been subscribed. Caused by "
 					+ e.getErrorMessage(), e);
 			throw new RuntimeException("RocketMQ Consumer hasn't been subscribed.", e);
@@ -148,12 +151,6 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 
 	protected class CloudStreamMessageListener implements MessageListener, RetryListener {
 
-		private final InstrumentationManager instrumentationManager;
-
-		CloudStreamMessageListener(InstrumentationManager instrumentationManager) {
-			this.instrumentationManager = instrumentationManager;
-		}
-
 		Acknowledgement consumeMessage(final List<MessageExt> msgs) {
 			boolean enableRetry = RocketMQInboundChannelAdapter.this.retryTemplate != null;
 			try {
@@ -180,10 +177,13 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 				}
 				else {
 					Acknowledgement result = doSendMsgs(msgs, null);
-					instrumentationManager
+					Optional.ofNullable(
-							.getConsumerInstrumentation(
+							RocketMQInboundChannelAdapter.this.instrumentationManager)
-									RocketMQInboundChannelAdapter.this.destination)
+							.ifPresent(manager -> {
-							.markConsumed();
+								manager.getConsumerInstrumentation(
+										RocketMQInboundChannelAdapter.this.destination)
+										.markConsumed();
+							});
 					return result;
 				}
 			}
@@ -191,10 +191,13 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 				logger.error(
 						"RocketMQ Message hasn't been processed successfully. Caused by ",
 						e);
-				instrumentationManager
+				Optional.ofNullable(
-						.getConsumerInstrumentation(
+						RocketMQInboundChannelAdapter.this.instrumentationManager)
-								RocketMQInboundChannelAdapter.this.destination)
+						.ifPresent(manager -> {
-						.markConsumedFailure();
+							manager.getConsumerInstrumentation(
+									RocketMQInboundChannelAdapter.this.destination)
+									.markConsumedFailure();
+						});
 				throw new RuntimeException(
 						"RocketMQ Message hasn't been processed successfully. Caused by ",
 						e);
@@ -232,16 +235,22 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 		public <T, E extends Throwable> void close(RetryContext context,
 				RetryCallback<T, E> callback, Throwable throwable) {
 			if (throwable != null) {
-				instrumentationManager
+				Optional.ofNullable(
-						.getConsumerInstrumentation(
+						RocketMQInboundChannelAdapter.this.instrumentationManager)
-								RocketMQInboundChannelAdapter.this.destination)
+						.ifPresent(manager -> {
-						.markConsumedFailure();
+							manager.getConsumerInstrumentation(
+									RocketMQInboundChannelAdapter.this.destination)
+									.markConsumedFailure();
+						});
 			}
 			else {
-				instrumentationManager
+				Optional.ofNullable(
-						.getConsumerInstrumentation(
+						RocketMQInboundChannelAdapter.this.instrumentationManager)
-								RocketMQInboundChannelAdapter.this.destination)
+						.ifPresent(manager -> {
-						.markConsumed();
+							manager.getConsumerInstrumentation(
+									RocketMQInboundChannelAdapter.this.destination)
+									.markConsumed();
+						});
 			}
 		}
 
@@ -254,11 +263,6 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 	protected class CloudStreamMessageListenerConcurrently
 			extends CloudStreamMessageListener implements MessageListenerConcurrently {
 
-		public CloudStreamMessageListenerConcurrently(
-				InstrumentationManager instrumentationManager) {
-			super(instrumentationManager);
-		}
-
 		@Override
 		public ConsumeConcurrentlyStatus consumeMessage(final List<MessageExt> msgs,
 				ConsumeConcurrentlyContext context) {
@@ -272,11 +276,6 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
 	protected class CloudStreamMessageListenerOrderly extends CloudStreamMessageListener
 			implements MessageListenerOrderly {
 
-		public CloudStreamMessageListenerOrderly(
-				InstrumentationManager instrumentationManager) {
-			super(instrumentationManager);
-		}
-
 		@Override
 		public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
 				ConsumeOrderlyContext context) {

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

@@ -2,6 +2,7 @@ package org.springframework.cloud.stream.binder.rocketmq.integration;
 
 import java.time.Instant;
 import java.util.Map;
+import java.util.Optional;
 
 import org.apache.rocketmq.client.exception.MQBrokerException;
 import org.apache.rocketmq.client.exception.MQClientException;
@@ -30,14 +31,14 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
 
 	private ProducerInstrumentation producerInstrumentation;
 
+	private InstrumentationManager instrumentationManager;
+
 	private final RocketMQProducerProperties producerProperties;
 
 	private final String destination;
 
 	private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
 
-	private final InstrumentationManager instrumentationManager;
-
 	protected volatile boolean running = false;
 
 	public RocketMQMessageHandler(String destination,
@@ -54,9 +55,10 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
 	public void start() {
 		producer = new DefaultMQProducer(destination);
 
-		producerInstrumentation = instrumentationManager
+		Optional.ofNullable(instrumentationManager).ifPresent(manager -> {
-				.getProducerInstrumentation(destination);
+			producerInstrumentation = manager.getProducerInstrumentation(destination);
-		instrumentationManager.addHealthInstrumentation(producerInstrumentation);
+			manager.addHealthInstrumentation(producerInstrumentation);
+		});
 
 		producer.setNamesrvAddr(rocketBinderConfigurationProperties.getNamesrvAddr());
 
@@ -66,10 +68,12 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
 
 		try {
 			producer.start();
-			producerInstrumentation.markStartedSuccessfully();
+			Optional.ofNullable(producerInstrumentation)
+					.ifPresent(p -> p.markStartedSuccessfully());
 		}
 		catch (MQClientException e) {
-			producerInstrumentation.markStartFailed(e);
+			Optional.ofNullable(producerInstrumentation)
+					.ifPresent(p -> p.markStartFailed(e));
 			logger.error(
 					"RocketMQ Message hasn't been sent. Caused by " + e.getMessage());
 			throw new MessagingException(e.getMessage(), e);
@@ -127,14 +131,16 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
 				RocketMQMessageHeaderAccessor.putSendResult((MutableMessage) message,
 						sendRes);
 			}
-			instrumentationManager.getRuntime().put(
+			Optional.ofNullable(instrumentationManager).ifPresent(manager -> {
-					RocketMQBinderConstants.LASTSEND_TIMESTAMP,
+				manager.getRuntime().put(RocketMQBinderConstants.LASTSEND_TIMESTAMP,
-					Instant.now().toEpochMilli());
+						Instant.now().toEpochMilli());
-			producerInstrumentation.markSent();
+			});
+			Optional.ofNullable(producerInstrumentation).ifPresent(p -> p.markSent());
 		}
 		catch (MQClientException | RemotingException | MQBrokerException
 				| InterruptedException | UnsupportedOperationException e) {
-			producerInstrumentation.markSentFailure();
+			Optional.ofNullable(producerInstrumentation)
+					.ifPresent(p -> p.markSentFailure());
 			logger.error(
 					"RocketMQ Message hasn't been sent. Caused by " + e.getMessage());
 			throw new MessagingException(e.getMessage(), e);

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

@@ -3,6 +3,7 @@ package org.springframework.cloud.stream.binder.rocketmq.metrics;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 
 import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.Metrics.Consumer;
@@ -15,22 +16,17 @@ import com.codahale.metrics.MetricRegistry;
  * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
  */
 public class InstrumentationManager {
-	private final MetricRegistry metricRegistry;
+
-	private final Map<String, Object> runtime;
+	private final MetricRegistry metricRegistry = new MetricRegistry();
+	private final Map<String, Object> runtime = new ConcurrentHashMap<>();
+
 	private final Map<String, ProducerInstrumentation> producerInstrumentations = new HashMap<>();
 	private final Map<String, ConsumerInstrumentation> consumeInstrumentations = new HashMap<>();
 	private final Map<String, ConsumerGroupInstrumentation> consumerGroupsInstrumentations = new HashMap<>();
 
 	private final Map<String, Instrumentation> healthInstrumentations = new HashMap<>();
 
-	public InstrumentationManager(MetricRegistry metricRegistry,
-			Map<String, Object> runtime) {
-		this.metricRegistry = metricRegistry;
-		this.runtime = runtime;
-	}
-
 	public ProducerInstrumentation getProducerInstrumentation(String destination) {
-
 		String key = Producer.PREFIX + destination;
 		producerInstrumentations.putIfAbsent(key,
 				new ProducerInstrumentation(metricRegistry, key));
@@ -63,4 +59,8 @@ public class InstrumentationManager {
 	public Map<String, Object> getRuntime() {
 		return runtime;
 	}
+
+	public MetricRegistry getMetricRegistry() {
+		return metricRegistry;
+	}
 }