GH-2914: Make delivery_tag as a high cardinality observation tag

Fixes: #2914

The `delivery_tag` on consumer side is really per message,
so that makes too many metric timers when this property is exposed as a low cardinality tag.

* Fix `RabbitListenerObservation` moving the `ListenerLowCardinalityTags.DELIVERY_TAG` into the `ListenerHighCardinalityTags.DELIVERY_TAG`
* Leave `ListenerLowCardinalityTags.DELIVERY_TAG` as deprecated to avoid compatibility issues,
but exclude it from the `getLowCardinalityKeyNames()` implementation

Author: artembilan

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/micrometer/RabbitListenerObservation.java

@@ -16,6 +16,8 @@
 
 package org.springframework.amqp.rabbit.support.micrometer;
 
+import java.util.Arrays;
+
 import io.micrometer.common.KeyValues;
 import io.micrometer.common.docs.KeyName;
 import io.micrometer.observation.Observation.Context;
@@ -45,7 +47,14 @@ public Class<? extends ObservationConvention<? extends Context>> getDefaultConve
 
 		@Override
 		public KeyName[] getLowCardinalityKeyNames() {
-			return ListenerLowCardinalityTags.values();
+			return Arrays.stream(ListenerLowCardinalityTags.values())
+					.filter((key) -> !ListenerLowCardinalityTags.DELIVERY_TAG.equals(key))
+					.toArray(KeyName[]::new);
+		}
+
+		@Override
+		public KeyName[] getHighCardinalityKeyNames() {
+			return ListenerHighCardinalityTags.values();
 		}
 
 	};
@@ -83,8 +92,33 @@ public String asString() {
 
 		/**
 		 * The delivery tag.
+		 * After deprecation this key is not exposed as a low cardinality tag.
 		 *
 		 * @since 3.2
+		 *
+		 * @deprecated in favor of {@link ListenerHighCardinalityTags#DELIVERY_TAG}
+		 */
+		@Deprecated(since = "3.2.1", forRemoval = true)
+		DELIVERY_TAG {
+
+			@Override
+			public String asString() {
+				return "messaging.rabbitmq.message.delivery_tag";
+			}
+
+		}
+
+	}
+
+	/**
+	 * High cardinality tags.
+	 *
+	 * @since 3.2.1
+	 */
+	public enum ListenerHighCardinalityTags implements KeyName {
+
+		/**
+		 * The delivery tag.
 		 */
 		DELIVERY_TAG {
 
@@ -97,6 +131,7 @@ public String asString() {
 
 	}
 
+
 	/**
 	 * Default {@link RabbitListenerObservationConvention} for Rabbit listener key values.
 	 */
@@ -112,12 +147,16 @@ public static class DefaultRabbitListenerObservationConvention implements Rabbit
 		public KeyValues getLowCardinalityKeyValues(RabbitMessageReceiverContext context) {
 			final var messageProperties = context.getCarrier().getMessageProperties();
 			return KeyValues.of(
-					RabbitListenerObservation.ListenerLowCardinalityTags.LISTENER_ID.asString(), context.getListenerId(),
+					RabbitListenerObservation.ListenerLowCardinalityTags.LISTENER_ID.asString(),
+					context.getListenerId(),
 					RabbitListenerObservation.ListenerLowCardinalityTags.DESTINATION_NAME.asString(),
-					messageProperties.getConsumerQueue(),
-					RabbitListenerObservation.ListenerLowCardinalityTags.DELIVERY_TAG.asString(),
-					String.valueOf(messageProperties.getDeliveryTag())
-			);
+					messageProperties.getConsumerQueue());
+		}
+
+		@Override
+		public KeyValues getHighCardinalityKeyValues(RabbitMessageReceiverContext context) {
+			return KeyValues.of(RabbitListenerObservation.ListenerHighCardinalityTags.DELIVERY_TAG.asString(),
+					String.valueOf(context.getCarrier().getMessageProperties().getDeliveryTag()));
 		}
 
 		@Override

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/support/micrometer/ObservationIntegrationTests.java

@@ -47,6 +47,7 @@
 /**
  * @author Artem Bilan
  * @author Gary Russell
+ *
  * @since 3.0
  */
 @RabbitAvailable(queues = { "int.observation.testQ1", "int.observation.testQ2" })
@@ -120,15 +121,13 @@ public SampleTestRunnerConsumer yourCode() {
 					.hasTimerWithNameAndTags("spring.rabbit.listener",
 							KeyValues.of(
 									KeyValue.of("spring.rabbit.listener.id", "obs1"),
-									KeyValue.of("messaging.destination.name", "int.observation.testQ1"),
-									KeyValue.of("messaging.rabbitmq.message.delivery_tag", "1")
+									KeyValue.of("messaging.destination.name", "int.observation.testQ1")
 							)
 					)
 					.hasTimerWithNameAndTags("spring.rabbit.listener",
 							KeyValues.of(
 									KeyValue.of("spring.rabbit.listener.id", "obs2"),
-									KeyValue.of("messaging.destination.name", "int.observation.testQ2"),
-									KeyValue.of("messaging.rabbitmq.message.delivery_tag", "1")
+									KeyValue.of("messaging.destination.name", "int.observation.testQ2")
 							)
 					);
 		};