Fix ProducerInterceptor usage in KafkaTempalte

The result of `ProducerInterceptor.onSend()` is out of use

**Cherry-pick to `3.0.x`**

Author: Wzy19930507

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

@@ -764,6 +764,7 @@ private CompletableFuture<SendResult<K, V>> observeSend(final ProducerRecord<K,
 			throw ex;
 		}
 	}
+
 	/**
 	 * Send the producer record.
 	 * @param producerRecord the producer record.
@@ -781,11 +782,9 @@ protected CompletableFuture<SendResult<K, V>> doSend(final ProducerRecord<K, V>
 		if (this.micrometerHolder != null) {
 			sample = this.micrometerHolder.start();
 		}
-		if (this.producerInterceptor != null) {
-			this.producerInterceptor.onSend(producerRecord);
-		}
+		ProducerRecord<K, V> interceptedRecord = interceptorProducerRecord(producerRecord);
 		Future<RecordMetadata> sendFuture =
-				producer.send(producerRecord, buildCallback(producerRecord, producer, future, sample, observation));
+				producer.send(interceptedRecord, buildCallback(interceptedRecord, producer, future, sample, observation));
 		// Maybe an immediate failure
 		if (sendFuture.isDone()) {
 			try {
@@ -802,10 +801,17 @@ protected CompletableFuture<SendResult<K, V>> doSend(final ProducerRecord<K, V>
 		if (this.autoFlush) {
 			flush();
 		}
-		this.logger.trace(() -> "Sent: " + KafkaUtils.format(producerRecord));
+		this.logger.trace(() -> "Sent: " + KafkaUtils.format(interceptedRecord));
 		return future;
 	}
 
+	private ProducerRecord<K, V> interceptorProducerRecord(ProducerRecord<K, V> producerRecord) {
+		if (this.producerInterceptor != null) {
+			return this.producerInterceptor.onSend(producerRecord);
+		}
+		return producerRecord;
+	}
+
 	private Callback buildCallback(final ProducerRecord<K, V> producerRecord, final Producer<K, V> producer,
 			final CompletableFuture<SendResult<K, V>> future, @Nullable Object sample, Observation observation) {
 

spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2022 the original author or authors.
+ * Copyright 2016-2023 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.
@@ -559,19 +559,21 @@ void testFutureFailureOnSend() {
 
 	@SuppressWarnings("unchecked")
 	@Test
-	void testProducerInterceptorManagedOnKafkaTemplate() {
+	void testProducerInterceptorManagedOnKafkaTemplate() throws Exception {
 
 		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
 		DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
 		ProducerInterceptor<Integer, String> producerInterceptor = Mockito.mock(ProducerInterceptor.class);
+		willAnswer(inv -> new ProducerRecord<>("prod-interceptor-test-1", "bar")).given(producerInterceptor).onSend(any());
 		template.setProducerInterceptor(producerInterceptor);
 
 		template.setDefaultTopic("prod-interceptor-test-1");
-		template.sendDefault("foo");
+		CompletableFuture<SendResult<Integer, String>> resultCompletableFuture = template.sendDefault("foo");
 
 		verify(producerInterceptor, times(1)).onSend(any(ProducerRecord.class));
 		verify(producerInterceptor, times(1)).onAcknowledgement(any(RecordMetadata.class), Mockito.isNull());
+		assertThat(resultCompletableFuture.get(10, TimeUnit.SECONDS).getProducerRecord()).isEqualTo(new ProducerRecord<>("prod-interceptor-test-1", "bar"));
 	}
 
 	@SuppressWarnings("unchecked")
@@ -591,13 +593,15 @@ void testProducerInterceptorNotSetOnKafkaTemplateNotInvoked() {
 
 	@SuppressWarnings("unchecked")
 	@Test
-	void testCompositeProducerInterceptor() {
+	void testCompositeProducerInterceptor() throws Exception {
 
 		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
 		DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
 		ProducerInterceptor<Integer, String> producerInterceptor1 = Mockito.mock(ProducerInterceptor.class);
 		ProducerInterceptor<Integer, String> producerInterceptor2 = Mockito.mock(ProducerInterceptor.class);
+		willAnswer(inv -> new ProducerRecord<>("prod-interceptor-test-3", "bar")).given(producerInterceptor1).onSend(any());
+		willAnswer(inv -> new ProducerRecord<>("prod-interceptor-test-3", "baz")).given(producerInterceptor2).onSend(any());
 		CompositeProducerInterceptor<Integer, String> compositeProducerInterceptor =
 				new CompositeProducerInterceptor<>(producerInterceptor1, producerInterceptor2);
 		template.setProducerInterceptor(compositeProducerInterceptor);
@@ -606,14 +610,15 @@ void testCompositeProducerInterceptor() {
 		doReturn(mockProducerRecord).when(producerInterceptor1).onSend(any(ProducerRecord.class));
 
 		template.setDefaultTopic("prod-interceptor-test-3");
-		template.sendDefault("foo");
+		CompletableFuture<SendResult<Integer, String>> result = template.sendDefault("foo");
 
 		InOrder inOrder = inOrder(producerInterceptor1, producerInterceptor2);
 
 		inOrder.verify(producerInterceptor1).onSend(any(ProducerRecord.class));
 		inOrder.verify(producerInterceptor2).onSend(any(ProducerRecord.class));
 		inOrder.verify(producerInterceptor1).onAcknowledgement(any(RecordMetadata.class), Mockito.isNull());
 		inOrder.verify(producerInterceptor2).onAcknowledgement(any(RecordMetadata.class), Mockito.isNull());
+		assertThat(result.get(10, TimeUnit.SECONDS).getProducerRecord()).isEqualTo(new ProducerRecord<>("prod-interceptor-test-3", "baz"));
 	}
 
 	@ParameterizedTest(name = "{0} is invalid")