From 740c811d6ced20f2d6668282d17c25ea7f71f256 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Thu, 19 Sep 2019 10:29:37 -0400 Subject: [PATCH] Fix brokent REST calls to Confluent Schema Registry The REST calls used to register and fetch schema were wrong in the controller. Fixing those and the corresponding tests --- .../client/ConfluentSchemaRegistryClient.java | 28 ++++++++++++++----- .../ConfluentSchemaRegistryClientTests.java | 17 +++++++---- 2 files changed, 32 insertions(+), 13 deletions(-) diff --git a/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/client/ConfluentSchemaRegistryClient.java b/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/client/ConfluentSchemaRegistryClient.java index 52ee3de..c52131e 100644 --- a/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/client/ConfluentSchemaRegistryClient.java +++ b/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/client/ConfluentSchemaRegistryClient.java @@ -33,6 +33,7 @@ import org.springframework.http.HttpMethod; import org.springframework.http.HttpStatus; import org.springframework.http.ResponseEntity; import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; import org.springframework.web.client.HttpStatusCodeException; import org.springframework.web.client.RestTemplate; @@ -82,9 +83,7 @@ public class ConfluentSchemaRegistryClient implements SchemaRegistryClient { Integer id = null; String payload = null; Map maps = new HashMap<>(); - maps.put("subject", subject); - maps.put("format", format); - maps.put("definition", schema); + maps.put("schema", schema); try { payload = this.mapper.writeValueAsString(maps); } @@ -93,16 +92,31 @@ public class ConfluentSchemaRegistryClient implements SchemaRegistryClient { } try { HttpEntity request = new HttpEntity<>(payload, headers); - ResponseEntity response = this.template.exchange(this.endpoint, + ResponseEntity response = this.template.exchange(this.endpoint + "/subjects/" + subject + "/versions", HttpMethod.POST, request, Map.class); id = (Integer) response.getBody().get("id"); - version = (Integer) ((Map) response.getBody()).get("version"); } catch (HttpStatusCodeException httpException) { throw new RuntimeException(String.format( "Failed to register subject %s, server replied with status %d", subject, httpException.getStatusCode().value()), httpException); } + + try { + ResponseEntity response = this.template.getForEntity(this.endpoint + "/subjects/" + subject + "/versions", + List.class); + + final List body = response.getBody(); + if (!CollectionUtils.isEmpty(body)) { + version = (Integer) body.get(body.size() - 1); + } + } + 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 @@ -112,8 +126,8 @@ public class ConfluentSchemaRegistryClient implements SchemaRegistryClient { @Override public String fetch(SchemaReference schemaReference) { - String path = String.format("/%s/%s/v%d", - schemaReference.getSubject(), schemaReference.getFormat(), schemaReference.getVersion()); + String path = String.format("/subjects/%s/versions/%d", + schemaReference.getSubject(), schemaReference.getVersion()); HttpHeaders headers = new HttpHeaders(); headers.put("Accept", ACCEPT_HEADERS); headers.add("Content-Type", "application/vnd.schemaregistry.v1+json"); diff --git a/spring-cloud-schema-registry-client/src/test/java/org/springframework/cloud/schema/avro/client/ConfluentSchemaRegistryClientTests.java b/spring-cloud-schema-registry-client/src/test/java/org/springframework/cloud/schema/avro/client/ConfluentSchemaRegistryClientTests.java index dfd1618..5dbb771 100644 --- a/spring-cloud-schema-registry-client/src/test/java/org/springframework/cloud/schema/avro/client/ConfluentSchemaRegistryClientTests.java +++ b/spring-cloud-schema-registry-client/src/test/java/org/springframework/cloud/schema/avro/client/ConfluentSchemaRegistryClientTests.java @@ -58,12 +58,17 @@ public class ConfluentSchemaRegistryClientTests { @Test public void registerSchema() throws Exception { this.mockRestServiceServer - .expect(requestTo("http://localhost:8081")) + .expect(requestTo("http://localhost:8081/subjects/user/versions")) .andExpect(method(HttpMethod.POST)) .andExpect(header("Content-Type", "application/json")) .andExpect(header("Accept", "application/vnd.schemaregistry.v1+json")) .andRespond(withSuccess("{\"id\":101,\"version\":1}", MediaType.APPLICATION_JSON)); + this.mockRestServiceServer + .expect(requestTo("http://localhost:8081/subjects/user/versions")) + .andExpect(method(HttpMethod.GET)) + .andRespond((withSuccess("[1]", MediaType.APPLICATION_JSON))); + ConfluentSchemaRegistryClient client = new ConfluentSchemaRegistryClient( this.restTemplate); SchemaRegistrationResponse response = client.register("user", "avro", "{}"); @@ -75,7 +80,7 @@ public class ConfluentSchemaRegistryClientTests { @Test(expected = RuntimeException.class) public void registerWithInvalidJson() { this.mockRestServiceServer - .expect(requestTo("http://localhost:8081")) + .expect(requestTo("http://localhost:8081/subjects/user/versions")) .andExpect(method(HttpMethod.POST)) .andExpect(header("Content-Type", "application/json")) .andExpect(header("Accept", "application/vnd.schemaregistry.v1+json")) @@ -88,7 +93,7 @@ public class ConfluentSchemaRegistryClientTests { @Test public void registerIncompatibleSchema() { this.mockRestServiceServer - .expect(requestTo("http://localhost:8081")) + .expect(requestTo("http://localhost:8081/subjects/user/versions")) .andExpect(method(HttpMethod.POST)) .andExpect(header("Content-Type", "application/json")) .andExpect(header("Accept", "application/vnd.schemaregistry.v1+json")) @@ -110,7 +115,7 @@ public class ConfluentSchemaRegistryClientTests { @Test public void findByReference() { this.mockRestServiceServer - .expect(requestTo("http://localhost:8081/user/avro/v1")) + .expect(requestTo("http://localhost:8081/subjects/user/versions/1")) .andExpect(method(HttpMethod.GET)) .andExpect( header("Content-Type", "application/vnd.schemaregistry.v1+json")) @@ -127,7 +132,7 @@ public class ConfluentSchemaRegistryClientTests { @Test(expected = SchemaNotFoundException.class) public void schemaNotFound() { this.mockRestServiceServer - .expect(requestTo("http://localhost:8081/user/avro/v1")) + .expect(requestTo("http://localhost:8081/subjects/user/versions/1")) .andExpect(method(HttpMethod.GET)) .andExpect( header("Content-Type", "application/vnd.schemaregistry.v1+json")) @@ -142,7 +147,7 @@ public class ConfluentSchemaRegistryClientTests { @Test public void responseErrorFetch() { this.mockRestServiceServer - .expect(requestTo("http://localhost:8081")) + .expect(requestTo("http://localhost:8081/subjects/user/versions")) .andExpect(method(HttpMethod.POST)) .andExpect(header("Content-Type", "application/json")) .andExpect(header("Accept", "application/vnd.schemaregistry.v1+json"))