Default to BYTES schema if JSON cannot be constructed for the message type

Author: cbornet

spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolver.java

@@ -39,6 +39,7 @@
 import org.apache.pulsar.common.schema.SchemaType;
 
 import org.springframework.core.ResolvableType;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.lang.Nullable;
 
 import com.google.protobuf.GeneratedMessageV3;
@@ -56,6 +57,8 @@
  */
 public class DefaultSchemaResolver implements SchemaResolver {
 
+	private final LogAccessor logger = new LogAccessor(this.getClass());
+
 	private static final Map<Class<?>, Schema<?>> BASE_SCHEMA_MAPPINGS = new HashMap<>();
 	static {
 		BASE_SCHEMA_MAPPINGS.put(byte[].class, Schema.BYTES);
@@ -129,10 +132,20 @@ public <T> Schema<T> getSchema(Class<?> messageClass, boolean returnDefault) {
 	}
 
 	@Nullable
-	private Schema<?> getCustomSchemaOrMaybeDefault(Class<?> messageClass, boolean returnDefault) {
+	protected Schema<?> getCustomSchemaOrMaybeDefault(Class<?> messageClass, boolean returnDefault) {
 		Schema<?> schema = this.customSchemaMappings.get(messageClass);
 		if (schema == null && returnDefault) {
-			return messageClass != null ? Schema.JSON(messageClass) : Schema.BYTES;
+			if (messageClass != null) {
+				try {
+					return Schema.JSON(messageClass);
+				}
+				catch (Exception e) {
+					if (logger.isDebugEnabled()) {
+						logger.debug(e, "Failed to create JSON schema for " + messageClass.getName());
+					}
+				}
+			}
+			return Schema.BYTES;
 		}
 		return schema;
 	}