Change MessageSource.payloadType to Class<?>

- consistency with other endpoints
- avoid the need for SpEL in XML config

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundChannelAdapterSpec.java

@@ -16,8 +16,6 @@
 
 package org.springframework.integration.kafka.dsl;
 
-import java.lang.reflect.Type;
-
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 
 import org.springframework.integration.dsl.MessageSourceSpec;
@@ -70,7 +68,7 @@ public class KafkaInboundChannelAdapterSpec<K, V>
 		return this;
 	}
 
-	public KafkaInboundChannelAdapterSpec<K, V> payloadType(Type type) {
+	public KafkaInboundChannelAdapterSpec<K, V> payloadType(Class<?> type) {
 		this.target.setPayloadType(type);
 		return this;
 	}

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java

@@ -16,7 +16,6 @@
 
 package org.springframework.integration.kafka.inbound;
 
-import java.lang.reflect.Type;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -113,7 +112,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 
 	private RecordMessageConverter messageConverter = new MessagingMessageConverter();
 
-	private Type payloadType;
+	private Class<?> payloadType;
 
 	private ConsumerRebalanceListener rebalanceListener;
 
@@ -202,7 +201,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 		this.messageConverter = messageConverter;
 	}
 
-	protected Type getPayloadType() {
+	protected Class<?> getPayloadType() {
 		return this.payloadType;
 	}
 
@@ -211,7 +210,7 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 	 * Only applies if a type-aware message converter is provided.
 	 * @param payloadType the type to convert to.
 	 */
-	public void setPayloadType(Type payloadType) {
+	public void setPayloadType(Class<?> payloadType) {
 		this.payloadType = payloadType;
 	}
 

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParserTests-context.xml

@@ -16,7 +16,7 @@
 			client-id="client"
 			group-id="group"
 			message-converter="converter"
-			payload-type="#{T(java.lang.String)}"
+			payload-type="java.lang.String"
 			raw-header="true"
 			auto-startup="false"
 			rebalance-listener="rebal">