Rework Kafka Publisher to use DSL Spec for bean

It is not recommended to use `IntegrationComponentSpec.getObject()` manually.

* Change `KafkaPublisherConfiguration` to use `KafkaProducerMessageHandlerSpec` for bean definition
* Fix `KafkaPublisherConfigurationTests` to add an `IntegrationAutoConfiguration` into the picture as well

consumer/kafka-publisher/src/main/java/org/springframework/cloud/fn/consumer/kafka/KafkaPublisherConfiguration.java

@@ -59,7 +59,7 @@ public Consumer<Message<?>> kafkaPublisher(KafkaProducerMessageHandler<?, ?> kaf
 	}
 
 	@Bean
-	public KafkaProducerMessageHandler<?, ?> kafkaProducerMessageHandlerSpec(KafkaTemplate<?, ?> kafkaTemplate,
+	public KafkaProducerMessageHandlerSpec<?, ?, ?> kafkaProducerMessageHandlerSpec(KafkaTemplate<?, ?> kafkaTemplate,
 			KafkaPublisherProperties kafkaPublisherProperties,
 			PublishSubscribeChannel kafkaPublisherSuccessChannel,
 			PublishSubscribeChannel kafkaPublisherFailureChannel,
@@ -91,7 +91,7 @@ public Consumer<Message<?>> kafkaPublisher(KafkaProducerMessageHandler<?, ?> kaf
 			kafkaProducerSpecComponentCustomizer.customize(kafkaProducerMessageHandlerSpec);
 		}
 
-		return kafkaProducerMessageHandlerSpec.getObject();
+		return kafkaProducerMessageHandlerSpec;
 	}
 
 	/**

consumer/kafka-publisher/src/test/java/org/springframework/cloud/fn/consumer/kafka/KafkaPublisherConfigurationTests.java

@@ -30,6 +30,7 @@
 import reactor.core.publisher.Sinks;
 
 import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.autoconfigure.integration.IntegrationAutoConfiguration;
 import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
 import org.springframework.boot.test.context.runner.ApplicationContextRunner;
 import org.springframework.cloud.fn.common.config.SpelExpressionConverterConfiguration;
@@ -59,6 +60,7 @@ public class KafkaPublisherConfigurationTests {
 
 	final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
 			.withConfiguration(AutoConfigurations.of(
+					IntegrationAutoConfiguration.class,
 					KafkaAutoConfiguration.class,
 					KafkaPublisherConfiguration.class,
 					SpelExpressionConverterConfiguration.class));