Update listener observation operation tag to process

This updates the listener observation operation tag according to latest OpenTelemetry recommendations.

Signed-off-by: Christian Fredriksson <[email protected]>

Author: cfredri4

spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/KafkaListenerObservation.java

@@ -231,7 +231,7 @@ public KeyValues getLowCardinalityKeyValues(KafkaRecordReceiverContext context)
 			KeyValues keyValues = KeyValues.of(
 					ListenerLowCardinalityTags.LISTENER_ID.withValue(context.getListenerId()),
 					ListenerLowCardinalityTags.MESSAGING_SYSTEM.withValue("kafka"),
-					ListenerLowCardinalityTags.MESSAGING_OPERATION.withValue("receive"),
+					ListenerLowCardinalityTags.MESSAGING_OPERATION.withValue("process"),
 					ListenerLowCardinalityTags.MESSAGING_SOURCE_NAME.withValue(context.getSource()),
 					ListenerLowCardinalityTags.MESSAGING_SOURCE_KIND.withValue("topic")
 			);
@@ -269,7 +269,7 @@ public KeyValues getHighCardinalityKeyValues(KafkaRecordReceiverContext context)
 
 		@Override
 		public String getContextualName(KafkaRecordReceiverContext context) {
-			return context.getSource() + " receive";
+			return context.getSource() + " process";
 		}
 
 		private static @Nullable String getConsumerId(@Nullable String groupId, @Nullable String clientId) {

spring-kafka/src/test/java/org/springframework/kafka/support/micrometer/ObservationTests.java

@@ -113,6 +113,7 @@
  * @author Christian Mergenthaler
  * @author Soby Chacko
  * @author Francois Rosiere
+ * @author Christian Fredriksson
  *
  * @since 3.0
  */
@@ -377,14 +378,14 @@ private SimpleSpan assertThatListenerSpanTags(Deque<SimpleSpan> spans, int tagSi
 								Map.entry("messaging.kafka.consumer.group", consumerGroup),
 								Map.entry("messaging.kafka.message.offset", offset),
 								Map.entry("messaging.kafka.source.partition", partition),
-								Map.entry("messaging.operation", "receive"),
+								Map.entry("messaging.operation", "process"),
 								Map.entry("messaging.source.kind", "topic"),
 								Map.entry("messaging.source.name", sourceName),
 								Map.entry("messaging.system", "kafka")));
 		if (keyValues.length > 0) {
 			Arrays.stream(keyValues).forEach(entry -> assertThat(span.getTags()).contains(entry));
 		}
-		assertThat(span.getName()).isEqualTo(sourceName + " receive");
+		assertThat(span.getName()).isEqualTo(sourceName + " process");
 		return span;
 	}
 
@@ -406,7 +407,7 @@ private void assertThatListenerHasTimerWithNameAndTags(MeterRegistryAssert meter
 		meterRegistryAssert.hasTimerWithNameAndTags("spring.kafka.listener",
 				KeyValues.of(
 								"messaging.kafka.consumer.group", consumerGroup,
-								"messaging.operation", "receive",
+								"messaging.operation", "process",
 								"messaging.source.kind", "topic",
 								"messaging.source.name", destName,
 								"messaging.system", "kafka",