Add schema resolution to binder

2023-01-21 22:52:55 -06:00
parent 34a9101076
commit bd75eefdce
7 changed files with 534 additions and 67 deletions
--- a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java
+++ b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java
@@ -20,6 +20,7 @@ import java.util.Objects;

 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.common.schema.KeyValue;
 import org.apache.pulsar.common.schema.SchemaType;

 import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
@@ -30,8 +31,10 @@ import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
 import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
 import org.springframework.cloud.stream.provisioning.ConsumerDestination;
 import org.springframework.cloud.stream.provisioning.ProducerDestination;
+import org.springframework.core.ResolvableType;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
@@ -77,10 +80,14 @@ public class PulsarMessageChannelBinder extends
 	@Override
 	protected MessageHandler createProducerMessageHandler(ProducerDestination destination,
 			ExtendedProducerProperties<PulsarProducerProperties> producerProperties, MessageChannel errorChannel) {
-		SchemaType schemaType = producerProperties.getExtension().getSchemaType();
-		if (producerProperties.isUseNativeEncoding() && schemaType != null) {
-			Schema<Object> schema = Objects.requireNonNull(this.schemaResolver.getSchema(schemaType, null));
-			this.pulsarTemplate.setSchema(schema);
+
+		if (producerProperties.isUseNativeEncoding()) {
+			Schema<Object> schema = resolveSchema(producerProperties.getExtension().getSchemaType(),
+					producerProperties.getExtension().getMessageType(),
+					producerProperties.getExtension().getMessageKeyType(),
+					producerProperties.getExtension().getMessageValueType());
+			this.pulsarTemplate.setSchema(
+					Objects.requireNonNull(schema, "Could not determine producer schema for " + destination.getName()));
 		}
 		return message -> {
 			try {
@@ -102,14 +109,18 @@ public class PulsarMessageChannelBinder extends
 			Message<Object> message = MessageBuilder.withPayload(msg.getValue()).build();
 			pulsarMessageDrivenChannelAdapter.send(message);
 		});
-		SchemaType schemaType = properties.getExtension().getSchemaType();
-		if (properties.isUseNativeDecoding() && schemaType != null) {
-			pulsarContainerProperties
-					.setSchema(Objects.requireNonNull(this.schemaResolver.getSchema(schemaType, null)));
+
+		if (properties.isUseNativeDecoding()) {
+			Schema<Object> schema = resolveSchema(properties.getExtension().getSchemaType(),
+					properties.getExtension().getMessageType(), properties.getExtension().getMessageKeyType(),
+					properties.getExtension().getMessageValueType());
+			pulsarContainerProperties.setSchema(
+					Objects.requireNonNull(schema, "Could not determine consumer schema for " + destination.getName()));
 		}
 		else {
 			pulsarContainerProperties.setSchema(Schema.BYTES);
 		}
+
 		String subscriptionName = PulsarBinderUtils.subscriptionName(properties.getExtension(), destination);
 		pulsarContainerProperties.setSubscriptionName(subscriptionName);
 		DefaultPulsarMessageListenerContainer<?> container = new DefaultPulsarMessageListenerContainer<>(
@@ -118,6 +129,40 @@ public class PulsarMessageChannelBinder extends
 		return pulsarMessageDrivenChannelAdapter;
 	}

+	// VisibleForTesting
+	@Nullable
+	Schema<Object> resolveSchema(@Nullable SchemaType schemaType, @Nullable Class<?> messageType,
+			@Nullable Class<?> messageKeyType, @Nullable Class<?> messageValueType) {
+		if (schemaType == null) {
+			schemaType = SchemaType.NONE;
+		}
+		ResolvableType resolvableType = null;
+		if (schemaType.isStruct()) {
+			resolvableType = ResolvableType.forClass(Objects.requireNonNull(messageType,
+					"'message-type' required for 'schema-type' " + schemaType.name()));
+		}
+		else if (schemaType == SchemaType.KEY_VALUE) {
+			resolvableType = ResolvableType.forClassWithGenerics(KeyValue.class,
+					Objects.requireNonNull(messageKeyType, "'message-key-type' required for 'schema-type' KEY_VALUE"),
+					Objects.requireNonNull(messageValueType,
+							"'message-value-type' required for 'schema-type' KEY_VALUE"));
+		}
+		else if (schemaType == SchemaType.NONE) {
+			if (messageType != null) {
+				resolvableType = ResolvableType.forClass(messageType);
+			}
+			else if (messageKeyType != null && messageValueType != null) {
+				resolvableType = ResolvableType.forClassWithGenerics(KeyValue.class, messageKeyType, messageValueType);
+			}
+			if (resolvableType == null) {
+				throw new IllegalArgumentException(
+						"'message-type' OR ('message-key-type' AND 'message-value-type') required for 'schema-type' NONE");
+			}
+		}
+		// TODO if schema == null then default lookup bean Schema<?> w/ name == binding
+		return this.schemaResolver.getSchema(schemaType, resolvableType);
+	}
+
 	@Override
 	public PulsarConsumerProperties getExtendedConsumerProperties(String channelName) {
 		return this.extendedBindingProperties.getExtendedConsumerProperties(channelName);
--- a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/properties/PulsarConsumerProperties.java
+++ b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/properties/PulsarConsumerProperties.java
@@ -25,17 +25,27 @@ import org.springframework.lang.Nullable;
 * Pulsar consumer properties used by the binder.
 *
 * @author Soby Chacko
+ * @author Chris Bono
 */
 public class PulsarConsumerProperties {

 	@Nullable
 	private String subscriptionName;

+	@Nullable
+	private SubscriptionType subscriptionType;
+
 	@Nullable
 	private SchemaType schemaType;

 	@Nullable
-	private SubscriptionType subscriptionType;
+	private Class<?> messageType;
+
+	@Nullable
+	private Class<?> messageKeyType;
+
+	@Nullable
+	private Class<?> messageValueType;

 	@Nullable
 	public String getSubscriptionName() {
@@ -46,15 +56,6 @@ public class PulsarConsumerProperties {
 		this.subscriptionName = subscriptionName;
 	}

-	@Nullable
-	public SchemaType getSchemaType() {
-		return this.schemaType;
-	}
-
-	public void setSchemaType(SchemaType schemaType) {
-		this.schemaType = schemaType;
-	}
-
 	@Nullable
 	public SubscriptionType getSubscriptionType() {
 		return this.subscriptionType;
@@ -64,4 +65,40 @@ public class PulsarConsumerProperties {
 		this.subscriptionType = subscriptionType;
 	}

+	@Nullable
+	public SchemaType getSchemaType() {
+		return this.schemaType;
+	}
+
+	public void setSchemaType(@Nullable SchemaType schemaType) {
+		this.schemaType = schemaType;
+	}
+
+	@Nullable
+	public Class<?> getMessageType() {
+		return this.messageType;
+	}
+
+	public void setMessageType(@Nullable Class<?> messageType) {
+		this.messageType = messageType;
+	}
+
+	@Nullable
+	public Class<?> getMessageKeyType() {
+		return this.messageKeyType;
+	}
+
+	public void setMessageKeyType(@Nullable Class<?> messageKeyType) {
+		this.messageKeyType = messageKeyType;
+	}
+
+	@Nullable
+	public Class<?> getMessageValueType() {
+		return this.messageValueType;
+	}
+
+	public void setMessageValueType(@Nullable Class<?> messageValueType) {
+		this.messageValueType = messageValueType;
+	}
+
 }
--- a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/properties/PulsarProducerProperties.java
+++ b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/properties/PulsarProducerProperties.java
@@ -24,19 +24,56 @@ import org.springframework.lang.Nullable;
 * Pulsar producer properties used by the binder.
 *
 * @author Soby Chacko
+ * @author Chris Bono
 */
 public class PulsarProducerProperties {

 	@Nullable
 	private SchemaType schemaType;

+	@Nullable
+	private Class<?> messageType;
+
+	@Nullable
+	private Class<?> messageKeyType;
+
+	@Nullable
+	private Class<?> messageValueType;
+
 	@Nullable
 	public SchemaType getSchemaType() {
 		return this.schemaType;
 	}

-	public void setSchemaType(SchemaType schemaType) {
+	public void setSchemaType(@Nullable SchemaType schemaType) {
 		this.schemaType = schemaType;
 	}

+	@Nullable
+	public Class<?> getMessageType() {
+		return this.messageType;
+	}
+
+	public void setMessageType(@Nullable Class<?> messageType) {
+		this.messageType = messageType;
+	}
+
+	@Nullable
+	public Class<?> getMessageKeyType() {
+		return this.messageKeyType;
+	}
+
+	public void setMessageKeyType(@Nullable Class<?> messageKeyType) {
+		this.messageKeyType = messageKeyType;
+	}
+
+	@Nullable
+	public Class<?> getMessageValueType() {
+		return this.messageValueType;
+	}
+
+	public void setMessageValueType(@Nullable Class<?> messageValueType) {
+		this.messageValueType = messageValueType;
+	}
+
 }
--- a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java
+++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java
@@ -20,7 +20,10 @@ import java.time.Duration;
 import java.util.function.Consumer;
 import java.util.function.Supplier;

+import org.apache.pulsar.client.impl.schema.JSONSchema;
+import org.apache.pulsar.common.schema.KeyValue;
 import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.slf4j.Logger;
@@ -34,66 +37,187 @@ import org.springframework.boot.test.system.CapturedOutput;
 import org.springframework.boot.test.system.OutputCaptureExtension;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
+import org.springframework.pulsar.core.DefaultSchemaResolver;
+import org.springframework.pulsar.core.SchemaResolver.SchemaResolverCustomizer;
 import org.springframework.pulsar.test.support.PulsarTestContainerSupport;

 /**
+ * Integration tests for {@link PulsarBinderIntegrationTests}.
+ *
 * @author Soby Chacko
+ * @author Chris Bono
 */
@ExtendWith(OutputCaptureExtension.class)
 class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {

-	@Test
-	void basicProducerConsumerBindingEndToEnd(CapturedOutput output) {
-		SpringApplication app = new SpringApplication(BasicScenarioConfig.class);
-		app.setWebApplicationType(WebApplicationType.NONE);
-		try (ConfigurableApplicationContext ignored = app.run(
-				"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
-				"--spring.cloud.function.definition=textSupplier;textLogger",
-				"--spring.cloud.stream.bindings.textLogger-in-0.destination=textSupplier-out-0",
-				"--spring.cloud.stream.pulsar.bindings.textLogger-in-0.consumer.subscription-name=basic-scenario-sub-1")) {
-			Awaitility.await().atMost(Duration.ofSeconds(10))
-					.until(() -> output.toString().contains("Hello binder: test-basic-scenario"));
+	@Nested
+	class DefaultEncoding {
+
+		@Test
+		void primitiveTypeString(CapturedOutput output) {
+			SpringApplication app = new SpringApplication(PrimitiveTextConfig.class);
+			app.setWebApplicationType(WebApplicationType.NONE);
+			try (ConfigurableApplicationContext ignored = app.run(
+					"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+					"--spring.cloud.function.definition=textSupplier;textLogger",
+					"--spring.cloud.stream.bindings.textLogger-in-0.destination=textSupplier-out-0",
+					"--spring.cloud.stream.pulsar.bindings.textLogger-in-0.consumer.subscription-name=pbit-text-sub1")) {
+				Awaitility.await().atMost(Duration.ofSeconds(10))
+						.until(() -> output.toString().contains("Hello binder: test-basic-scenario"));
+			}
 		}
+
+		@Test
+		void primitiveTypeFloat(CapturedOutput output) {
+			SpringApplication app = new SpringApplication(PrimitiveFloatConfig.class);
+			app.setWebApplicationType(WebApplicationType.NONE);
+			try (ConfigurableApplicationContext ignored = app.run(
+					"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+					"--spring.cloud.function.definition=piSupplier;piLogger",
+					"--spring.cloud.stream.bindings.piSupplier-out-0.destination=pi-stream",
+					"--spring.cloud.stream.bindings.piLogger-in-0.destination=pi-stream",
+					"--spring.cloud.stream.pulsar.bindings.piLogger-in-0.consumer.subscription-name=pbit-float-sub1")) {
+				Awaitility.await().atMost(Duration.ofSeconds(10))
+						.until(() -> output.toString().contains("Hello binder: 3.14"));
+			}
+		}
+
 	}

-	@Test
-	void useNativeEncodingDecodingWorkAsExpected(CapturedOutput output) {
-		SpringApplication app = new SpringApplication(PiStreamConfig.class);
-		app.setWebApplicationType(WebApplicationType.NONE);
-		try (ConfigurableApplicationContext ignored = app.run(
-				"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
-				"--spring.cloud.function.definition=piSupplier;piLogger",
-				"--spring.cloud.stream.bindings.piLogger-in-0.destination=piSupplier-out-0",
-				"--spring.cloud.stream.bindings.piSupplier-out-0.producer.use-native-encoding=true",
-				"--spring.cloud.stream.bindings.piLogger-in-0.consumer.use-native-decoding=true",
-				"--spring.cloud.stream.pulsar.bindings.piLogger-in-0.consumer.schema-type=FLOAT",
-				"--spring.cloud.stream.pulsar.bindings.piSupplier-out-0.producer.schema-type=FLOAT",
-				"--spring.cloud.stream.pulsar.bindings.piLogger-in-0.consumer.subscription-name=native-encoding-decoding-sub-1")) {
-			Awaitility.await().atMost(Duration.ofSeconds(10))
-					.until(() -> output.toString().contains("Hello binder: 3.14"));
-		}
-	}
+	@Nested
+	class NativeEncoding {

-	@Test
-	void basicProducerConsumerBindingEndToEndWithNonTextPayloadType(CapturedOutput output) {
-		SpringApplication app = new SpringApplication(PiStreamConfig.class);
-		app.setWebApplicationType(WebApplicationType.NONE);
-		try (ConfigurableApplicationContext ignored = app.run(
-				"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
-				"--spring.cloud.function.definition=piSupplier;piLogger",
-				"--spring.cloud.stream.bindings.piSupplier-out-0.destination=pi-stream",
-				"--spring.cloud.stream.bindings.piLogger-in-0.destination=pi-stream",
-				"--spring.cloud.stream.pulsar.bindings.piLogger-in-0.consumer.subscription-name=native-encoding-decoding-sub-2")) {
-			Awaitility.await().atMost(Duration.ofSeconds(10))
-					.until(() -> output.toString().contains("Hello binder: 3.14"));
+		@Test
+		void primitiveTypeFloat(CapturedOutput output) {
+			SpringApplication app = new SpringApplication(PrimitiveFloatConfig.class);
+			app.setWebApplicationType(WebApplicationType.NONE);
+			try (ConfigurableApplicationContext ignored = app.run(
+					"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+					"--spring.cloud.function.definition=piSupplier;piLogger",
+					"--spring.cloud.stream.bindings.piLogger-in-0.destination=piSupplier-out-0",
+					"--spring.cloud.stream.bindings.piSupplier-out-0.producer.use-native-encoding=true",
+					"--spring.cloud.stream.bindings.piLogger-in-0.consumer.use-native-decoding=true",
+					"--spring.cloud.stream.pulsar.bindings.piLogger-in-0.consumer.schema-type=FLOAT",
+					"--spring.cloud.stream.pulsar.bindings.piSupplier-out-0.producer.schema-type=FLOAT",
+					"--spring.cloud.stream.pulsar.bindings.piLogger-in-0.consumer.subscription-name=pbit-float-sub2")) {
+				Awaitility.await().atMost(Duration.ofSeconds(10))
+						.until(() -> output.toString().contains("Hello binder: 3.14"));
+			}
 		}
+
+		@Test
+		void jsonTypeFoo(CapturedOutput output) {
+			SpringApplication app = new SpringApplication(JsonFooConfig.class);
+			app.setWebApplicationType(WebApplicationType.NONE);
+			try (ConfigurableApplicationContext ignored = app.run(
+					"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+					"--spring.cloud.function.definition=fooSupplier;fooLogger",
+					"--spring.cloud.stream.bindings.fooLogger-in-0.destination=fooSupplier-out-0",
+					"--spring.cloud.stream.bindings.fooSupplier-out-0.producer.use-native-encoding=true",
+					"--spring.cloud.stream.bindings.fooLogger-in-0.consumer.use-native-decoding=true",
+					"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.schema-type=JSON",
+					"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.message-type=" + Foo.class.getName(),
+					"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.subscription-name=pbit-foo-sub1",
+					"--spring.cloud.stream.pulsar.bindings.fooSupplier-out-0.producer.schema-type=JSON",
+					"--spring.cloud.stream.pulsar.bindings.fooSupplier-out-0.producer.message-type="
+							+ Foo.class.getName())) {
+				Awaitility.await().atMost(Duration.ofSeconds(10))
+						.until(() -> output.toString().contains("Hello binder: Foo[value=5150]"));
+			}
+		}
+
+		@Test
+		void jsonTypeFooWithNoSchemaTypeAndCustomFooMapping(CapturedOutput output) {
+			SpringApplication app = new SpringApplication(JsonFooWithCustomMappingConfig.class);
+			app.setWebApplicationType(WebApplicationType.NONE);
+			try (ConfigurableApplicationContext ignored = app.run(
+					"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+					"--spring.cloud.function.definition=fooSupplier;fooLogger",
+					"--spring.cloud.stream.bindings.fooSupplier-out-0.destination=foo-stream-2",
+					"--spring.cloud.stream.bindings.fooLogger-in-0.destination=foo-stream-2",
+					"--spring.cloud.stream.bindings.fooSupplier-out-0.producer.use-native-encoding=true",
+					"--spring.cloud.stream.bindings.fooLogger-in-0.consumer.use-native-decoding=true",
+					"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.message-type=" + Foo.class.getName(),
+					"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.subscription-name=pbit-foo-sub2",
+					"--spring.cloud.stream.pulsar.bindings.fooSupplier-out-0.producer.message-type="
+							+ Foo.class.getName())) {
+				Awaitility.await().atMost(Duration.ofSeconds(10))
+						.until(() -> output.toString().contains("Hello binder: Foo[value=5150]"));
+			}
+		}
+
+		@Test
+		void jsonTypeFooWithNoSchemaTypeAndNoCustomFooMapping(CapturedOutput output) {
+			SpringApplication app = new SpringApplication(JsonFooConfig.class);
+			app.setWebApplicationType(WebApplicationType.NONE);
+			try (ConfigurableApplicationContext ignored = app.run(
+					"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+					"--spring.cloud.function.definition=fooSupplier;fooLogger",
+					"--spring.cloud.stream.bindings.fooSupplier-out-0.destination=foo-stream-3",
+					"--spring.cloud.stream.bindings.fooLogger-in-0.destination=foo-stream-3",
+					"--spring.cloud.stream.bindings.fooSupplier-out-0.producer.use-native-encoding=true",
+					"--spring.cloud.stream.bindings.fooLogger-in-0.consumer.use-native-decoding=true",
+					"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.message-type=" + Foo.class.getName(),
+					"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.subscription-name=pbit-foo-sub3",
+					"--spring.cloud.stream.pulsar.bindings.fooSupplier-out-0.producer.message-type="
+							+ Foo.class.getName())) {
+				Awaitility.await().atMost(Duration.ofSeconds(10)).until(
+						() -> output.toString().contains("Could not determine producer schema for foo-stream-3"));
+			}
+		}
+
+		@Test
+		void jsonTypeFooConsumerWithNoSchemaTypeAndNoCustomFooMapping(CapturedOutput output) {
+			SpringApplication app = new SpringApplication(JsonFooConsumerConfig.class);
+			app.setWebApplicationType(WebApplicationType.NONE);
+			try (ConfigurableApplicationContext ignored = app.run(
+					"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+					"--spring.cloud.function.definition=fooSupplier;fooLogger",
+					"--spring.cloud.stream.bindings.fooSupplier-out-0.destination=foo-stream-4",
+					"--spring.cloud.stream.bindings.fooLogger-in-0.destination=foo-stream-4",
+					"--spring.cloud.stream.bindings.fooLogger-in-0.consumer.use-native-decoding=true",
+					"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.message-type=" + Foo.class.getName(),
+					"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.subscription-name=pbit-foo-sub4")) {
+				Awaitility.await().atMost(Duration.ofSeconds(10)).until(
+						() -> output.toString().contains("Could not determine consumer schema for foo-stream-4"));
+			}
+		}
+
+		@Test
+		void keyValueTypeWithCustomFooMapping(CapturedOutput output) {
+			SpringApplication app = new SpringApplication(KeyValueFooConfig.class);
+			app.setWebApplicationType(WebApplicationType.NONE);
+			try (ConfigurableApplicationContext ignored = app.run(
+					"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
+					"--spring.cloud.function.definition=fooSupplier;fooLogger",
+					"--spring.cloud.stream.bindings.fooSupplier-out-0.destination=kv-stream-1",
+					"--spring.cloud.stream.bindings.fooLogger-in-0.destination=kv-stream-1",
+					"--spring.cloud.stream.bindings.fooSupplier-out-0.producer.use-native-encoding=true",
+					"--spring.cloud.stream.bindings.fooLogger-in-0.consumer.use-native-decoding=true",
+					"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.schema-type=KEY_VALUE",
+					"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.message-key-type="
+							+ String.class.getName(),
+					"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.message-value-type="
+							+ Foo.class.getName(),
+					"--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.subscription-name=pbit-kv-sub1",
+					"--spring.cloud.stream.pulsar.bindings.fooSupplier-out-0.producer.schema-type=KEY_VALUE",
+					"--spring.cloud.stream.pulsar.bindings.fooSupplier-out-0.producer.message-key-type="
+							+ String.class.getName(),
+					"--spring.cloud.stream.pulsar.bindings.fooSupplier-out-0.producer.message-value-type="
+							+ Foo.class.getName())) {
+				Awaitility.await().atMost(Duration.ofSeconds(10))
+						.until(() -> output.toString().contains("Hello binder: 5150->Foo[value=5150]"));
+			}
+		}
+
 	}

 	@EnableAutoConfiguration
 	@SpringBootConfiguration
-	static class BasicScenarioConfig {
+	static class PrimitiveTextConfig {

-		private final Logger logger = LoggerFactory.getLogger(BasicScenarioConfig.class);
+		private final Logger logger = LoggerFactory.getLogger(getClass());

 		@Bean
 		public Supplier<String> textSupplier() {
@@ -109,9 +233,9 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {

 	@EnableAutoConfiguration
 	@SpringBootConfiguration
-	static class PiStreamConfig {
+	static class PrimitiveFloatConfig {

-		private final Logger logger = LoggerFactory.getLogger(BasicScenarioConfig.class);
+		private final Logger logger = LoggerFactory.getLogger(getClass());

 		@Bean
 		public Supplier<Float> piSupplier() {
@@ -125,4 +249,77 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {

 	}

+	@EnableAutoConfiguration
+	@SpringBootConfiguration
+	static class JsonFooConsumerConfig {
+
+		private final Logger logger = LoggerFactory.getLogger(getClass());
+
+		public Supplier<String> fooSupplier() {
+			return () -> "5150";
+		}
+
+		@Bean
+		public Consumer<Foo> fooLogger() {
+			return f -> this.logger.info("Hello binder: " + f);
+		}
+
+	}
+
+	@EnableAutoConfiguration
+	@SpringBootConfiguration
+	static class JsonFooConfig {
+
+		private final Logger logger = LoggerFactory.getLogger(getClass());
+
+		@Bean
+		public Supplier<Foo> fooSupplier() {
+			return () -> new Foo("5150");
+		}
+
+		@Bean
+		public Consumer<Foo> fooLogger() {
+			return f -> this.logger.info("Hello binder: " + f);
+		}
+
+	}
+
+	@EnableAutoConfiguration
+	@SpringBootConfiguration
+	@Import(JsonFooConfig.class)
+	static class JsonFooWithCustomMappingConfig {
+
+		@Bean
+		public SchemaResolverCustomizer<DefaultSchemaResolver> customMappings() {
+			return (resolver) -> resolver.addCustomSchemaMapping(Foo.class, JSONSchema.of(Foo.class));
+		}
+
+	}
+
+	@EnableAutoConfiguration
+	@SpringBootConfiguration
+	static class KeyValueFooConfig {
+
+		private final Logger logger = LoggerFactory.getLogger(getClass());
+
+		@Bean
+		public SchemaResolverCustomizer<DefaultSchemaResolver> customMappings() {
+			return (resolver) -> resolver.addCustomSchemaMapping(Foo.class, JSONSchema.of(Foo.class));
+		}
+
+		@Bean
+		public Supplier<KeyValue<String, Foo>> fooSupplier() {
+			return () -> new KeyValue<>("5150", new Foo("5150"));
+		}
+
+		@Bean
+		public Consumer<KeyValue<String, Foo>> fooLogger() {
+			return f -> this.logger.info("Hello binder: " + f.getKey() + "->" + f.getValue());
+		}
+
+	}
+
+	record Foo(String value) {
+	}
+
 }
--- a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinderResolveSchemaTests.java
+++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinderResolveSchemaTests.java
@@ -0,0 +1,137 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.pulsar.spring.cloud.stream.binder;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatExceptionOfType;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+import org.apache.pulsar.common.schema.KeyValue;
+import org.apache.pulsar.common.schema.SchemaType;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.EnumSource.Mode;
+
+import org.springframework.core.ResolvableType;
+import org.springframework.pulsar.core.PulsarConsumerFactory;
+import org.springframework.pulsar.core.PulsarTemplate;
+import org.springframework.pulsar.core.SchemaResolver;
+import org.springframework.pulsar.spring.cloud.stream.binder.provisioning.PulsarTopicProvisioner;
+
+/**
+ * Unit tests for {@link PulsarMessageChannelBinder#resolveSchema}.
+ *
+ * @author Chris Bono
+ */
+public class PulsarMessageChannelBinderResolveSchemaTests {
+
+	private SchemaResolver resolver = mock(SchemaResolver.class);
+
+	@SuppressWarnings("unchecked")
+	private PulsarMessageChannelBinder binder = new PulsarMessageChannelBinder(mock(PulsarTopicProvisioner.class),
+			mock(PulsarTemplate.class), mock(PulsarConsumerFactory.class), resolver);
+
+	@ParameterizedTest
+	@EnumSource(mode = Mode.MATCH_NONE, names = "^(AUTO.*|AVRO|JSON|KEY_VALUE|NONE|PROTOBUF.*)$")
+	void primitiveSchemaTypes(SchemaType schemaType) {
+		binder.resolveSchema(schemaType, null, null, null);
+		verify(resolver).getSchema(schemaType, null);
+	}
+
+	@ParameterizedTest
+	@EnumSource(mode = Mode.MATCH_ALL, names = "^(JSON|AVRO|PROTOBUF)$")
+	void structSchemaTypes(SchemaType schemaType) {
+		binder.resolveSchema(schemaType, Foo.class, null, null);
+		verify(resolver).getSchema(schemaType, ResolvableType.forClass(Foo.class));
+	}
+
+	@Test
+	void keyValueSchemaType() {
+		binder.resolveSchema(SchemaType.KEY_VALUE, null, Foo.class, Bar.class);
+		verify(resolver).getSchema(SchemaType.KEY_VALUE,
+				ResolvableType.forClassWithGenerics(KeyValue.class, Foo.class, Bar.class));
+	}
+
+	@ParameterizedTest
+	@EnumSource(mode = Mode.MATCH_ALL, names = "^(JSON|AVRO|PROTOBUF)$")
+	void structSchemaTypesRequireMessageType(SchemaType schemaType) {
+		assertThatExceptionOfType(NullPointerException.class)
+				.isThrownBy(() -> binder.resolveSchema(schemaType, null, null, null))
+				.withMessage("'message-type' required for 'schema-type' " + schemaType.name());
+	}
+
+	@Test
+	void keyValueSchemaTypeRequiresKeyAndValueTypes() {
+		assertThatExceptionOfType(NullPointerException.class)
+				.isThrownBy(() -> binder.resolveSchema(SchemaType.KEY_VALUE, null, null, null))
+				.withMessage("'message-key-type' required for 'schema-type' KEY_VALUE");
+		assertThatExceptionOfType(NullPointerException.class)
+				.isThrownBy(() -> binder.resolveSchema(SchemaType.KEY_VALUE, null, null, Bar.class))
+				.withMessage("'message-key-type' required for 'schema-type' KEY_VALUE");
+		assertThatExceptionOfType(NullPointerException.class)
+				.isThrownBy(() -> binder.resolveSchema(SchemaType.KEY_VALUE, null, Foo.class, null))
+				.withMessage("'message-value-type' required for 'schema-type' KEY_VALUE");
+	}
+
+	@Nested
+	class SchemaTypeNone {
+
+		@Test
+		void withMesssageType() {
+			binder.resolveSchema(SchemaType.NONE, Foo.class, null, null);
+			verify(resolver).getSchema(SchemaType.NONE, ResolvableType.forClass(Foo.class));
+		}
+
+		@Test
+		void withKeyAndValueTypes() {
+			binder.resolveSchema(SchemaType.NONE, null, Foo.class, Bar.class);
+			verify(resolver).getSchema(SchemaType.NONE,
+					ResolvableType.forClassWithGenerics(KeyValue.class, Foo.class, Bar.class));
+		}
+
+		@Test
+		void withMessageTypeAndKeyAndValueTypes() {
+			binder.resolveSchema(SchemaType.NONE, Foo.class, String.class, Bar.class);
+			verify(resolver).getSchema(SchemaType.NONE, ResolvableType.forClass(Foo.class));
+		}
+
+		@Test
+		void withOnlyKeyType() {
+			assertThatExceptionOfType(IllegalArgumentException.class)
+					.isThrownBy(() -> binder.resolveSchema(SchemaType.NONE, null, Foo.class, null)).withMessage(
+							"'message-type' OR ('message-key-type' AND 'message-value-type') required for 'schema-type' NONE");
+		}
+
+		@Test
+		void withOnlyValueType() {
+			assertThatExceptionOfType(IllegalArgumentException.class)
+					.isThrownBy(() -> binder.resolveSchema(SchemaType.NONE, null, null, Foo.class)).withMessage(
+							"'message-type' OR ('message-key-type' AND 'message-value-type') required for 'schema-type' NONE");
+
+		}
+
+	}
+
+	record Foo(String value) {
+	}
+
+	record Bar(String value) {
+	}
+
+}
--- a/spring-pulsar-spring-cloud-stream-binder/src/test/resources/logback-test.xml
+++ b/spring-pulsar-spring-cloud-stream-binder/src/test/resources/logback-test.xml
@@ -0,0 +1,13 @@
+<configuration>
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n</pattern>
+        </encoder>
+    </appender>
+    <root level="WARN">
+        <appender-ref ref="STDOUT"/>
+    </root>
+    <logger name="org.testcontainers" level="ERROR"/>
+    <logger name="com.github.dockerjava" level="ERROR"/>
+	<logger name="org.springframework.pulsar" level="INFO"/>
+</configuration>
--- a/src/checkstyle/checkstyle-suppressions.xml
+++ b/src/checkstyle/checkstyle-suppressions.xml
@@ -3,10 +3,11 @@
 		"-//Checkstyle//DTD SuppressionFilter Configuration 1.2//EN"
 		"https://checkstyle.org/dtds/suppressions_1_2.dtd">
 <suppressions>
-	<suppress files="package-info\.java" checks=".*"/>
-	<suppress files="[\\/]test[\\/]" checks="RequireThis"/>
-	<suppress files="[\\/]test[\\/]" checks="Javadoc*"/>
-	<suppress files="PulsarFunctionAdministrationIntegrationTests" checks="Regexp"/>
+	<suppress files="package-info\.java" checks=".*" />
+	<suppress files="[\\/]test[\\/]" checks="RequireThis" />
+	<suppress files="[\\/]test[\\/]" checks="Javadoc*" />
+	<suppress files="PulsarFunctionAdministrationIntegrationTests" checks="Regexp" />
+	<suppress files="PulsarMessageChannelBinderResolveSchemaTests" checks="AvoidStaticImport" />
 	<suppress files="DefaultSchemaResolverTests" checks="AvoidStaticImport|MethodParamPad" />
 	<suppress files="Proto" checks=".*"/>
 	<suppress files="ReactiveSpringPulsarBootApp" checks="HideUtilityClassConstructor"/>