Move getContextualContext to conventions.

Author: garyrussell

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

@@ -44,4 +44,9 @@ public KeyValues getHighCardinalityKeyValues(KafkaRecordReceiverContext context)
 		return KeyValues.empty();
 	}
 
+	@Override
+	public String getContextualName(KafkaRecordReceiverContext context) {
+		return context.getSource() + " receive";
+	}
+
 }

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

@@ -44,4 +44,9 @@ public KeyValues getHighCardinalityKeyValues(KafkaRecordSenderContext context) {
 		return KeyValues.empty();
 	}
 
+	@Override
+	public String getContextualName(KafkaRecordSenderContext context) {
+		return context.getDestination() + " send";
+	}
+
 }

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

@@ -58,9 +58,12 @@ public String getName() {
 		return "spring.kafka.listener";
 	}
 
-	@Override
-	public String getContextualName() {
-		return this.record.topic() + " receive";
+	/**
+	 * Return the source topic.
+	 * @return the source.
+	 */
+	public String getSource() {
+		return this.record.topic();
 	}
 
 }

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

@@ -51,9 +51,12 @@ public String getName() {
 		return "spring.kafka.template";
 	}
 
-	@Override
-	public String getContextualName() {
-		return this.destination + " send";
+	/**
+	 * Return the destination topic.
+	 * @return the topic.
+	 */
+	public String getDestination() {
+		return this.destination;
 	}
 
 }