GH-897 fixed the NPE issue with CacheManager

- Added a new constructor which takes CacheManager as an argument
- Deprecated the old constructor and setter for CacheManager
- Added fallback to NoOpsCacheManager to the deprecatd construtor to avoid the possiblity of NPE.
- Added warning to notify user that caching is _effectively disabled_ if NoOpCacheManager is used.

Fix #897

spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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.
@@ -28,6 +28,7 @@ import org.apache.avro.reflect.ReflectData;
 import org.springframework.beans.factory.BeanInitializationException;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.cache.CacheManager;
+import org.springframework.cache.support.NoOpCacheManager;
 import org.springframework.cloud.stream.schema.ParsedSchema;
 import org.springframework.cloud.stream.schema.SchemaNotFoundException;
 import org.springframework.cloud.stream.schema.SchemaReference;
@@ -64,8 +65,10 @@ import org.springframework.util.ObjectUtils;
  *
  * When converting from a message, the converter will parse the content-type and use it to
  * fetch and cache the writer schema using the provided {@link SchemaRegistryClient}.
+ *
  * @author Marius Bogoevici
  * @author Vinicius Carvalho
+ * @author Oleg Zhurakousky
  */
 public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessageConverter
 		implements InitializingBean {
@@ -98,15 +101,27 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag
 	private String prefix = "vnd";
 
 	/**
-	 * Creates a new instance, configuring it with a {@link SchemaRegistryClient}.
+	 * @deprecated as of release 1.2.2 in favor of
+	 * {@link #AvroSchemaRegistryClientMessageConverter(SchemaRegistryClient, CacheManager)}
+	 */
+	@Deprecated
+	public AvroSchemaRegistryClientMessageConverter(SchemaRegistryClient schemaRegistryClient) {
+		this(schemaRegistryClient, new NoOpCacheManager());
+	}
+
+	/**
+	 * Creates a new instance, configuring it with {@link SchemaRegistryClient} and {@link CacheManager}.
 	 * @param schemaRegistryClient the {@link SchemaRegistryClient} used to interact with
 	 * the schema registry server.
+	 * @param cacheManager instance of {@link CacheManager} to cache parsed schemas. If caching
+	 * is not required use {@link NoOpCacheManager}
 	 */
-	public AvroSchemaRegistryClientMessageConverter(
-			SchemaRegistryClient schemaRegistryClient) {
+	public AvroSchemaRegistryClientMessageConverter(SchemaRegistryClient schemaRegistryClient, CacheManager cacheManager) {
 		super(Arrays.asList(new MimeType("application", "*+avro")));
 		Assert.notNull(schemaRegistryClient, "cannot be null");
+		Assert.notNull(cacheManager, "'cacheManager' cannot be null");
 		this.schemaRegistryClient = schemaRegistryClient;
+		this.cacheManager = cacheManager;
 	}
 
 	public boolean isDynamicSchemaGenerationEnabled() {
@@ -179,6 +194,12 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag
 				}
 			}
 		}
+		if (this.cacheManager instanceof NoOpCacheManager){
+			logger.warn("Schema caching is effectively disabled "
+					+ "since configured cache manager is a NoOpCacheManager. If this was not "
+					+ "the intention, please provide the appropriate instance of CacheManager "
+					+ "(i.e., ConcurrentMapCacheManager).");
+		}
 	}
 
 	protected String toSubject(Schema schema) {
@@ -316,7 +337,13 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag
 		return schema;
 	}
 
+
+	/**
+	 * @deprecated as of release 1.0.4. Please use the constructor to inject CacheManager
+	 */
+	@Deprecated
 	public void setCacheManager(CacheManager cacheManager) {
+		Assert.notNull(cacheManager, "'cacheManager' cannot be null");
 		this.cacheManager = cacheManager;
 	}
 }

spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/AvroSchemaRegistryClientMessageConverterTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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.
@@ -23,24 +23,30 @@ import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
 
+import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.cache.support.NoOpCacheManager;
 import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.annotation.StreamListener;
 import org.springframework.cloud.stream.messaging.Sink;
 import org.springframework.cloud.stream.messaging.Source;
+import org.springframework.cloud.stream.schema.avro.AvroSchemaRegistryClientMessageConverter;
+import org.springframework.cloud.stream.schema.client.DefaultSchemaRegistryClient;
 import org.springframework.cloud.stream.schema.client.EnableSchemaRegistryClient;
 import org.springframework.cloud.stream.schema.client.SchemaRegistryClient;
 import org.springframework.cloud.stream.schema.server.SchemaRegistryServerApplication;
 import org.springframework.cloud.stream.test.binder.MessageCollector;
 import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.MessageBuilder;
 
 import static org.assertj.core.api.Assertions.assertThat;
-
 /**
  * @author Marius Bogoevici
+ * @author Oleg Zhurakousky
  */
 public class AvroSchemaRegistryClientMessageConverterTests {
 
@@ -142,4 +148,21 @@ public class AvroSchemaRegistryClientMessageConverterTests {
 		}
 
 	}
+
+	@Test
+	public void testNoCacheConfiguration (){
+		ConfigurableApplicationContext sourceContext = SpringApplication.run(NoCacheConfiguration.class, "--spring.main.web-environment=false");
+		AvroSchemaRegistryClientMessageConverter converter = sourceContext.getBean(AvroSchemaRegistryClientMessageConverter.class);
+		DirectFieldAccessor accessor = new DirectFieldAccessor(converter);
+		assertThat(accessor.getPropertyValue("cacheManager")).isInstanceOf(NoOpCacheManager.class);
+	}
+
+	@Configuration
+	public static class NoCacheConfiguration {
+		@SuppressWarnings("deprecation")
+		@Bean
+		AvroSchemaRegistryClientMessageConverter avroSchemaRegistryClientMessageConverter() {
+			return new AvroSchemaRegistryClientMessageConverter(new DefaultSchemaRegistryClient());
+		}
+	}
 }