Add prerix based subject naming strategy

spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/avro/AvroMessageConverterAutoConfiguration.java

@@ -81,6 +81,7 @@ public class AvroMessageConverterAutoConfiguration {
 			throw new IllegalStateException("Unable to create SubjectNamingStrategy "
 					+ avroMessageConverterProperties.getSubjectNamingStrategy().toString(), ex);
 		}
+		avroSchemaRegistryClientMessageConverter.setSubjectNamePrefix(avroMessageConverterProperties.getSubjectNamePrefix());
 
 		return avroSchemaRegistryClientMessageConverter;
 	}

spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/avro/AvroMessageConverterProperties.java

@@ -50,6 +50,8 @@ public class AvroMessageConverterProperties {
 
 	private String prefix = "vnd";
 
+	private String subjectNamePrefix;
+
 	private Class<? extends SubjectNamingStrategy> subjectNamingStrategy = DefaultSubjectNamingStrategy.class;
 
 	public Resource getReaderSchema() {
@@ -103,4 +105,11 @@ public class AvroMessageConverterProperties {
 		this.schemaImports = schemaImports;
 	}
 
+	public String getSubjectNamePrefix() {
+		return subjectNamePrefix;
+	}
+
+	public void setSubjectNamePrefix(String subjectNamePrefix) {
+		this.subjectNamePrefix = subjectNamePrefix;
+	}
 }

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

@@ -132,6 +132,8 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag
 
 	private String prefix = "vnd";
 
+	private String subjectNamePrefix;
+
 	private SubjectNamingStrategy subjectNamingStrategy;
 
 	/**
@@ -227,6 +229,10 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag
 		this.subjectNamingStrategy = subjectNamingStrategy;
 	}
 
+	public void setSubjectNamePrefix(String subjectNamePrefix) {
+		this.subjectNamePrefix = subjectNamePrefix;
+	}
+
 	@Override
 	public void afterPropertiesSet() {
 		this.versionedSchema = Pattern.compile("application/" + this.prefix
@@ -267,8 +273,8 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag
 		}
 	}
 
-	protected String toSubject(Schema schema) {
-		return this.subjectNamingStrategy.toSubject(schema);
+	protected String toSubject(String subjectNamePrefix, Schema schema) {
+		return this.subjectNamingStrategy.toSubject(subjectNamePrefix, schema);
 	}
 
 	@Override
@@ -300,7 +306,7 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag
 		}
 
 		if (parsedSchema.getRegistration() == null) {
-			SchemaRegistrationResponse response = this.schemaRegistryClient.register(toSubject(schema),
+			SchemaRegistrationResponse response = this.schemaRegistryClient.register(toSubject(this.subjectNamePrefix, schema),
 					AVRO_FORMAT, parsedSchema.getRepresentation());
 			parsedSchema.setRegistration(response);
 
@@ -376,7 +382,7 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag
 					+ schema.getNamespace() + "." + schema.getName());
 		}
 
-		this.schemaRegistryClient.register(toSubject(schema), AVRO_FORMAT, schema.toString());
+		this.schemaRegistryClient.register(toSubject(this.subjectNamePrefix, schema), AVRO_FORMAT, schema.toString());
 
 		if (this.logger.isInfoEnabled()) {
 			this.logger.info("Schema " + schema.getName() + " registered with id " + schema);

spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/avro/DefaultSubjectNamingStrategy.java

@@ -18,14 +18,18 @@ package org.springframework.cloud.schema.registry.avro;
 
 import org.apache.avro.Schema;
 
+import org.springframework.util.StringUtils;
+
 /**
  * @author David Kalosi
  */
 public class DefaultSubjectNamingStrategy implements SubjectNamingStrategy {
 
 	@Override
-	public String toSubject(Schema schema) {
-		return schema.getName().toLowerCase();
+	public String toSubject(String subjectNamePrefix, Schema schema) {
+		return StringUtils.hasText(subjectNamePrefix) ?
+				subjectNamePrefix + "-" + schema.getName().toLowerCase() :
+				schema.getName().toLowerCase();
 	}
 
 }

spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/avro/QualifiedSubjectNamingStrategy.java

@@ -18,6 +18,8 @@ package org.springframework.cloud.schema.registry.avro;
 
 import org.apache.avro.Schema;
 
+import org.springframework.util.StringUtils;
+
 /**
  * @author José A. Íñigo
  * @since 2.2.0
@@ -25,8 +27,10 @@ import org.apache.avro.Schema;
 public class QualifiedSubjectNamingStrategy implements SubjectNamingStrategy {
 
 	@Override
-	public String toSubject(Schema schema) {
-		return schema.getFullName().toLowerCase();
+	public String toSubject(String subjectNamePrefix, Schema schema) {
+		return StringUtils.hasText(subjectNamePrefix) ?
+				subjectNamePrefix + "-" + schema.getFullName().toLowerCase() :
+				schema.getFullName().toLowerCase();
 	}
 
 }

spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/avro/SubjectNamingStrategy.java

@@ -28,9 +28,10 @@ public interface SubjectNamingStrategy {
 	/**
 	 * Takes the Avro schema on input and returns the generated subject under which the
 	 * schema should be registered.
+	 * @param subjectNamePrefix optional subject name prefix
 	 * @param schema schema to register
 	 * @return subject name
 	 */
-	String toSubject(Schema schema);
+	String toSubject(String subjectNamePrefix, Schema schema);
 
 }

spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/avro/SubjectPrefixOnlyNamingStrategy.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2020-2020 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.cloud.schema.registry.avro;
+
+import org.apache.avro.Schema;
+
+/**
+ * @author Christian Tzolov
+ */
+public class SubjectPrefixOnlyNamingStrategy implements SubjectNamingStrategy {
+
+	@Override
+	public String toSubject(String subjectNamePrefix, Schema schema) {
+		return subjectNamePrefix;
+	}
+
+}

spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/client/ConfluentSchemaRegistryClient.java

@@ -163,5 +163,4 @@ public class ConfluentSchemaRegistryClient implements SchemaRegistryClient {
 			}
 		}
 	}
-
 }