GH-2733: Fix Meter Memory Leak

Fixes #2733

`AbstractMessageChannel` exception meters caused a memory leak because they
are stored in a `Set` for destruction purposes but did not implement
`hashCode` and `equals`.

Add `hashCode()` and `equals()` to all meter facades, delegating to the
underlying Micrometer objects.

Author: garyrussell

spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java

@@ -555,8 +555,9 @@ private Message<?> convertPayloadIfNecessary(Message<?> message) {
 	protected abstract boolean doSend(Message<?> message, long timeout);
 
 	@Override
-	public void destroy() throws Exception {
+	public void destroy() throws Exception { // NOSONAR TODO: remove throws in 5.2
 		this.meters.forEach(MeterFacade::remove);
+		this.meters.clear();
 	}
 
 	/**

spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -233,7 +233,7 @@ public boolean hasExecutorInterceptors() {
 	protected abstract Message<?> doReceive(long timeout);
 
 	@Override
-	public void destroy() throws Exception {
+	public void destroy() throws Exception { // NOSONAR TODO: remove throws in 5.2
 		super.destroy();
 		if (this.receiveCounter != null) {
 			this.receiveCounter.remove();

spring-integration-core/src/main/java/org/springframework/integration/support/management/micrometer/MicrometerMetricsCaptor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -195,6 +195,19 @@ public void record(long time, TimeUnit unit) {
 			this.timer.record(time, unit);
 		}
 
+		@Override
+		public int hashCode() {
+			return this.timer.hashCode();
+		}
+
+		@Override
+		public boolean equals(Object obj) {
+			if (!MicroTimer.class.equals(obj.getClass())) {
+				return false;
+			}
+			return this.timer.equals(((MicroTimer) obj).timer);
+		}
+
 	}
 
 	protected static class MicroCounterBuilder implements CounterBuilder {
@@ -246,6 +259,19 @@ public void increment() {
 			this.counter.increment();
 		}
 
+		@Override
+		public int hashCode() {
+			return this.counter.hashCode();
+		}
+
+		@Override
+		public boolean equals(Object obj) {
+			if (!MicroCounter.class.equals(obj.getClass())) {
+				return false;
+			}
+			return this.counter.equals(((MicroCounter) obj).counter);
+		}
+
 	}
 
 	protected static class MicroGaugeBuilder implements GaugeBuilder {
@@ -292,6 +318,19 @@ protected Gauge getMeter() {
 			return this.gauge;
 		}
 
+		@Override
+		public int hashCode() {
+			return this.gauge.hashCode();
+		}
+
+		@Override
+		public boolean equals(Object obj) {
+			if (!MicroGauge.class.equals(obj.getClass())) {
+				return false;
+			}
+			return this.gauge.equals(((MicroGauge) obj).gauge);
+		}
+
 	}
 
 }

spring-integration-core/src/test/java/org/springframework/integration/support/management/micrometer/MicrometerMetricsTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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.
@@ -19,6 +19,8 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.fail;
 
+import java.util.Set;
+
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -40,11 +42,13 @@
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.endpoint.AbstractMessageSource;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit4.SpringRunner;
 
 import io.micrometer.core.instrument.MeterRegistry;
@@ -58,6 +62,7 @@
  *
  */
 @RunWith(SpringRunner.class)
+@DirtiesContext
 public class MicrometerMetricsTests {
 
 	@Autowired
@@ -84,17 +89,26 @@ public class MicrometerMetricsTests {
 	@Autowired
 	private NullChannel nullChannel;
 
+	@SuppressWarnings("unchecked")
 	@Test
 	public void testSend() throws Exception {
 		GenericMessage<String> message = new GenericMessage<>("foo");
 		this.channel.send(message);
 		try {
-			this.channel.send(this.source.receive());
+			this.channel.send(this.source.receive()); // "bar"
+			fail("Expected exception");
+		}
+		catch (MessagingException e) {
+			assertThat(e.getCause().getMessage()).isEqualTo("testErrorCount");
+		}
+		try {
+			this.channel.send(new GenericMessage<>("bar"));
 			fail("Expected exception");
 		}
 		catch (MessagingException e) {
 			assertThat(e.getCause().getMessage()).isEqualTo("testErrorCount");
 		}
+		assertThat(TestUtils.getPropertyValue(this.channel, "meters", Set.class)).hasSize(2);
 		this.channel2.send(message);
 		this.queue.send(message);
 		this.queue.send(message);
@@ -141,12 +155,12 @@ public void testSend() throws Exception {
 		assertThat(registry.get("spring.integration.send")
 				.tag("name", "channel")
 				.tag("result", "failure")
-				.timer().count()).isEqualTo(1);
+				.timer().count()).isEqualTo(2);
 
 		assertThat(registry.get("spring.integration.send")
 				.tag("name", "eipMethod.handler")
 				.tag("result", "failure")
-				.timer().count()).isEqualTo(1);
+				.timer().count()).isEqualTo(2);
 
 		assertThat(registry.get("spring.integration.receive")
 				.tag("name", "queue")
@@ -202,6 +216,9 @@ public void testSend() throws Exception {
 		catch (MeterNotFoundException e) {
 			assertThat(e).hasMessageContaining("No meter with name 'spring.integration.receive' was found");
 		}
+		this.channel.destroy();
+		assertThat(TestUtils.getPropertyValue(this.channel, "meters", Set.class)).hasSize(0);
+
 	}
 
 	@Configuration