[GH-48] rely on schema from response

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

@@ -21,24 +21,33 @@
  */
 public class SchemaRegistrationResponse {
 
-	private int id;
+    private int id;
 
-	private SchemaReference schemaReference;
+    private SchemaReference schemaReference;
 
-	public int getId() {
-		return this.id;
-	}
+    private String schema;
 
-	public void setId(int id) {
-		this.id = id;
-	}
+    public int getId() {
+        return this.id;
+    }
 
-	public SchemaReference getSchemaReference() {
-		return this.schemaReference;
-	}
+    public void setId(int id) {
+        this.id = id;
+    }
 
-	public void setSchemaReference(SchemaReference schemaReference) {
-		this.schemaReference = schemaReference;
-	}
+    public SchemaReference getSchemaReference() {
+        return this.schemaReference;
+    }
 
+    public void setSchemaReference(SchemaReference schemaReference) {
+        this.schemaReference = schemaReference;
+    }
+
+    public String getSchema() {
+        return schema;
+    }
+
+    public void setSchema(String schema) {
+        this.schema = schema;
+    }
 }

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

@@ -308,8 +308,13 @@ protected Schema resolveSchemaForWriting(Object payload, MessageHeaders headers,
 		if (parsedSchema.getRegistration() == null) {
 			SchemaRegistrationResponse response = this.schemaRegistryClient.register(toSubject(this.subjectNamePrefix, schema),
 					AVRO_FORMAT, parsedSchema.getRepresentation());
-			parsedSchema.setRegistration(response);
 
+			// rely on schema in response
+			schema = new Schema.Parser().parse(response.getSchema());
+			parsedSchema = new ParsedSchema(schema);
+			this.getCache(REFERENCE_CACHE_NAME).put(schema, parsedSchema);
+
+			parsedSchema.setRegistration(response);
 		}
 
 		SchemaReference schemaReference = parsedSchema.getRegistration().getSchemaReference();

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

@@ -83,6 +83,8 @@ public SchemaRegistrationResponse register(String subject, String format, String
 		String payload = null;
 		Map<String, String> maps = new HashMap<>();
 		maps.put("schema", schema);
+		SchemaRegistrationResponse schemaRegistrationResponse = new SchemaRegistrationResponse();
+		schemaRegistrationResponse.setSchema(schema);
 		try {
 			payload = this.mapper.writeValueAsString(maps);
 		}
@@ -107,14 +109,14 @@ public SchemaRegistrationResponse register(String subject, String format, String
 			final List body = response.getBody();
 			if (!CollectionUtils.isEmpty(body)) {
 				version = (Integer) body.get(body.size() - 1);
+				schemaRegistrationResponse.setSchema(fetch(new SchemaReference(subject, version, "avro")));
 			}
 		}
 		catch (HttpStatusCodeException httpException) {
 			throw new RuntimeException(String.format("Failed to register subject %s, server replied with status %d",
 					subject, httpException.getStatusCode().value()), httpException);
 		}
 
-		SchemaRegistrationResponse schemaRegistrationResponse = new SchemaRegistrationResponse();
 		schemaRegistrationResponse.setId(id);
 		schemaRegistrationResponse.setSchemaReference(new SchemaReference(subject, version, "avro"));
 		return schemaRegistrationResponse;

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

@@ -73,6 +73,7 @@ public SchemaRegistrationResponse register(String subject, String format, String
 			registrationResponse.setId((Integer) responseBody.get("id"));
 			registrationResponse.setSchemaReference(new SchemaReference(subject, (Integer) responseBody.get("version"),
 					responseBody.get("format").toString()));
+			registrationResponse.setSchema(schema);
 			return registrationResponse;
 		}
 		throw new RuntimeException(