GH-3103 Fix the retrieval of the schema version registered on serializing an Avro message with the ConfluentSchemaRegistryClient

Signed-off-by: Andy Palmer <[email protected]>

Resolves #3103
Resolves #3107

Author: Andy Palmer

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

@@ -16,14 +16,8 @@
 
 package org.springframework.cloud.stream.schema.registry.client;
 
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
-
 import org.springframework.cloud.stream.schema.registry.SchemaNotFoundException;
 import org.springframework.cloud.stream.schema.registry.SchemaReference;
 import org.springframework.cloud.stream.schema.registry.SchemaRegistrationResponse;
@@ -37,6 +31,11 @@
 import org.springframework.web.client.HttpStatusCodeException;
 import org.springframework.web.client.RestTemplate;
 
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
 /**
  * @author Vinicius Carvalho
  * @author Marius Bogoevici
@@ -102,11 +101,12 @@ public SchemaRegistrationResponse register(String subject, String format, String
 
 		try {
 			ResponseEntity<List> response = this.template.getForEntity(
-					this.endpoint + "/subjects/" + subject + "/versions", List.class);
+					this.endpoint + "/schemas/ids/" + id + "/versions", List.class);
 
 			final List body = response.getBody();
 			if (!CollectionUtils.isEmpty(body)) {
-				version = (Integer) body.get(body.size() - 1);
+				// Assume only a single version is registered for this ID
+				version = (Integer) ((Map<String, Object>) body.get(0)).get("version");
 			}
 		}
 		catch (HttpStatusCodeException httpException) {

schema-registry/spring-cloud-stream-schema-registry-client/src/test/java/org/springframework/cloud/stream/schema/avro/client/ConfluentSchemaRegistryClientTests.java

@@ -66,9 +66,9 @@ public void registerSchema() throws Exception {
 				.andRespond(withSuccess("{\"id\":101,\"version\":1}", MediaType.APPLICATION_JSON));
 
 		this.mockRestServiceServer
-				.expect(requestTo("http://localhost:8081/subjects/user/versions"))
+				.expect(requestTo("http://localhost:8081/schemas/ids/101/versions"))
 				.andExpect(method(HttpMethod.GET))
-				.andRespond((withSuccess("[1]", MediaType.APPLICATION_JSON)));
+				.andRespond((withSuccess("[{\"subject\":\"user\",\"version\":1}]", MediaType.APPLICATION_JSON)));
 
 		ConfluentSchemaRegistryClient client = new ConfluentSchemaRegistryClient(
 				this.restTemplate);