diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/ConfluentSchemaRegistryClient.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/ConfluentSchemaRegistryClient.java index 956c03b93..0698d796e 100644 --- a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/ConfluentSchemaRegistryClient.java +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/ConfluentSchemaRegistryClient.java @@ -32,7 +32,7 @@ import org.springframework.http.HttpMethod; import org.springframework.http.HttpStatus; import org.springframework.http.ResponseEntity; import org.springframework.util.Assert; -import org.springframework.web.client.HttpClientErrorException; +import org.springframework.web.client.HttpStatusCodeException; import org.springframework.web.client.RestTemplate; /** @@ -49,8 +49,16 @@ public class ConfluentSchemaRegistryClient implements SchemaRegistryClient { private ObjectMapper mapper; public ConfluentSchemaRegistryClient() { - this.template = new RestTemplate(); - this.mapper = new ObjectMapper(); + this(new RestTemplate()); + } + + public ConfluentSchemaRegistryClient(RestTemplate template) { + this(template,new ObjectMapper()); + } + + public ConfluentSchemaRegistryClient(RestTemplate template, ObjectMapper mapper) { + this.template = template; + this.mapper = mapper; } public void setEndpoint(String endpoint) { @@ -58,29 +66,74 @@ public class ConfluentSchemaRegistryClient implements SchemaRegistryClient { } @Override - public SchemaRegistrationResponse register(String subject, String format, String schema) { + public SchemaRegistrationResponse register(String subject, String format, + String schema) { Assert.isTrue("avro".equals(format), "Only Avro is supported"); + String path = String.format("/subjects/%s/versions", subject); + HttpHeaders headers = new HttpHeaders(); + headers.put("Accept", Arrays.asList("application/vnd.schemaregistry.v1+json", + "application/vnd.schemaregistry+json", "application/json")); + headers.add("Content-Type", "application/json"); + Integer version = null; + Integer id = null; + String payload = null; + try { + payload = this.mapper + .writeValueAsString(Collections.singletonMap("schema", schema)); + } + catch (JsonProcessingException e) { + throw new RuntimeException("Could not parse schema, invalid JSON format", e); + } + try { + HttpEntity request = new HttpEntity<>(payload, headers); + ResponseEntity response = this.template.exchange(this.endpoint + path, + HttpMethod.POST, request, Map.class); + id = (Integer) response.getBody().get("id"); + version = getSubjectVersion(subject, payload); + } + 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; + } + + /** + * Confluent register API returns the id, but we need the version of a given schema + * subject. After a successful registration we can inquire the server to get the + * version of a schema + * @param subject + * @return + */ + private Integer getSubjectVersion(String subject, String payload) { String path = String.format("/subjects/%s", subject); HttpHeaders headers = new HttpHeaders(); - headers.put("Accept", - Arrays.asList("application/vnd.schemaregistry.v1+json", "application/vnd.schemaregistry+json", - "application/json")); + headers.put("Accept", Arrays.asList("application/vnd.schemaregistry.v1+json", + "application/vnd.schemaregistry+json", "application/json")); headers.add("Content-Type", "application/json"); Integer version = null; try { - String payload = this.mapper.writeValueAsString(Collections.singletonMap("schema", schema)); + HttpEntity request = new HttpEntity<>(payload, headers); - ResponseEntity response = this.template.exchange(this.endpoint + path, HttpMethod.POST, request, - Map.class); + ResponseEntity response = this.template.exchange(this.endpoint + path, + HttpMethod.POST, request, Map.class); version = (Integer) response.getBody().get("version"); } - catch (JsonProcessingException e) { - e.printStackTrace(); + 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(version); - schemaRegistrationResponse.setSchemaReference(new SchemaReference(subject, version, "avro")); - return schemaRegistrationResponse; + return version; } @Override @@ -88,20 +141,19 @@ public class ConfluentSchemaRegistryClient implements SchemaRegistryClient { String path = String.format("/subjects/%s/versions/%d", schemaReference.getSubject(), schemaReference.getVersion()); HttpHeaders headers = new HttpHeaders(); - headers.put("Accept", - Arrays.asList("application/vnd.schemaregistry.v1+json", "application/vnd.schemaregistry+json", - "application/json")); + headers.put("Accept", Arrays.asList("application/vnd.schemaregistry.v1+json", + "application/vnd.schemaregistry+json", "application/json")); headers.add("Content-Type", "application/vnd.schemaregistry.v1+json"); HttpEntity request = new HttpEntity<>("", headers); try { - ResponseEntity response = this.template.exchange(this.endpoint + path, HttpMethod.GET, request, - Map.class); + ResponseEntity response = this.template.exchange(this.endpoint + path, + HttpMethod.GET, request, Map.class); return (String) response.getBody().get("schema"); } - catch (HttpClientErrorException e) { + catch (HttpStatusCodeException e) { if (e.getStatusCode() == HttpStatus.NOT_FOUND) { - throw new SchemaNotFoundException( - String.format("Could not find schema for reference: %s", schemaReference)); + throw new SchemaNotFoundException(String.format( + "Could not find schema for reference: %s", schemaReference)); } else { throw e; @@ -113,17 +165,16 @@ public class ConfluentSchemaRegistryClient implements SchemaRegistryClient { public String fetch(int id) { String path = String.format("/schemas/ids/%d", id); HttpHeaders headers = new HttpHeaders(); - headers.put("Accept", - Arrays.asList("application/vnd.schemaregistry.v1+json", "application/vnd.schemaregistry+json", - "application/json")); + headers.put("Accept", Arrays.asList("application/vnd.schemaregistry.v1+json", + "application/vnd.schemaregistry+json", "application/json")); headers.add("Content-Type", "application/vnd.schemaregistry.v1+json"); HttpEntity request = new HttpEntity<>("", headers); try { - ResponseEntity response = this.template.exchange(this.endpoint + path, HttpMethod.GET, request, - Map.class); + ResponseEntity response = this.template.exchange(this.endpoint + path, + HttpMethod.GET, request, Map.class); return (String) response.getBody().get("schema"); } - catch (HttpClientErrorException e) { + catch (HttpStatusCodeException e) { if (e.getStatusCode() == HttpStatus.NOT_FOUND) { throw new SchemaNotFoundException( String.format("Could not find schema with id: %s", id)); diff --git a/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/client/ConfluentSchemaRegistryClientTests.java b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/client/ConfluentSchemaRegistryClientTests.java new file mode 100644 index 000000000..da5e10259 --- /dev/null +++ b/spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/client/ConfluentSchemaRegistryClientTests.java @@ -0,0 +1,160 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://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.avro.client; + +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import org.springframework.cloud.stream.schema.SchemaNotFoundException; +import org.springframework.cloud.stream.schema.SchemaReference; +import org.springframework.cloud.stream.schema.SchemaRegistrationResponse; +import org.springframework.cloud.stream.schema.client.ConfluentSchemaRegistryClient; +import org.springframework.http.HttpMethod; +import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; +import org.springframework.test.web.client.MockRestServiceServer; +import org.springframework.web.client.HttpStatusCodeException; +import org.springframework.web.client.RestTemplate; + +import static org.springframework.test.web.client.match.MockRestRequestMatchers.header; +import static org.springframework.test.web.client.match.MockRestRequestMatchers.method; +import static org.springframework.test.web.client.match.MockRestRequestMatchers.requestTo; +import static org.springframework.test.web.client.response.MockRestResponseCreators.withBadRequest; +import static org.springframework.test.web.client.response.MockRestResponseCreators.withStatus; +import static org.springframework.test.web.client.response.MockRestResponseCreators.withSuccess; + + +/** + * @author Vinicius Carvalho + */ +public class ConfluentSchemaRegistryClientTests { + + private RestTemplate restTemplate; + private MockRestServiceServer mockRestServiceServer; + + @Before + public void setup(){ + this.restTemplate = new RestTemplate(); + this.mockRestServiceServer = MockRestServiceServer.createServer(restTemplate); + } + + @Test + public void registerSchema() throws Exception{ + this.mockRestServiceServer.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}", MediaType.APPLICATION_JSON)); + + this.mockRestServiceServer.expect(requestTo("http://localhost:8081/subjects/user")) + .andExpect(method(HttpMethod.POST)) + .andExpect(header("Content-Type","application/json")) + .andExpect(header("Accept","application/vnd.schemaregistry.v1+json")) + .andRespond(withSuccess("{\"version\":1}", MediaType.APPLICATION_JSON)); + + ConfluentSchemaRegistryClient client = new ConfluentSchemaRegistryClient(this.restTemplate); + SchemaRegistrationResponse response = client.register("user","avro","{}"); + Assert.assertEquals(1,response.getSchemaReference().getVersion()); + Assert.assertEquals(101,response.getId()); + this.mockRestServiceServer.verify(); + } + + @Test(expected = RuntimeException.class) + public void registerWithInvalidJson() { + this.mockRestServiceServer.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(withBadRequest()); + ConfluentSchemaRegistryClient client = new ConfluentSchemaRegistryClient(this.restTemplate); + SchemaRegistrationResponse response = client.register("user","avro","<>"); + } + + @Test + public void registerIncompatibleSchema() { + this.mockRestServiceServer.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(withStatus(HttpStatus.CONFLICT)); + ConfluentSchemaRegistryClient client = new ConfluentSchemaRegistryClient(this.restTemplate); + Exception expected = null; + try { + SchemaRegistrationResponse response = client.register("user","avro","{}"); + } + catch (Exception e) { + expected = e; + } + Assert.assertTrue(expected instanceof RuntimeException); + Assert.assertTrue(expected.getCause() instanceof HttpStatusCodeException); + this.mockRestServiceServer.verify(); + } + + @Test + public void responseErrorFetch() { + this.mockRestServiceServer.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}", MediaType.APPLICATION_JSON)); + + this.mockRestServiceServer.expect(requestTo("http://localhost:8081/subjects/user")) + .andExpect(method(HttpMethod.POST)) + .andExpect(header("Content-Type","application/json")) + .andExpect(header("Accept","application/vnd.schemaregistry.v1+json")) + .andRespond(withBadRequest()); + ConfluentSchemaRegistryClient client = new ConfluentSchemaRegistryClient(this.restTemplate); + Exception expected = null; + try { + SchemaRegistrationResponse response = client.register("user","avro","{}"); + } + catch (Exception e) { + expected = e; + } + Assert.assertTrue(expected instanceof RuntimeException); + Assert.assertTrue(expected.getCause() instanceof HttpStatusCodeException); + this.mockRestServiceServer.verify(); + } + + @Test + public void findByReference(){ + this.mockRestServiceServer.expect(requestTo("http://localhost:8081/subjects/user/versions/1")) + .andExpect(method(HttpMethod.GET)) + .andExpect(header("Content-Type","application/vnd.schemaregistry.v1+json")) + .andExpect(header("Accept","application/vnd.schemaregistry.v1+json")) + .andRespond(withSuccess("{\"schema\":\"\"}", MediaType.APPLICATION_JSON)); + ConfluentSchemaRegistryClient client = new ConfluentSchemaRegistryClient(this.restTemplate); + SchemaReference reference = new SchemaReference("user",1,"avro"); + String schema = client.fetch(reference); + Assert.assertEquals("",schema); + this.mockRestServiceServer.verify(); + } + + @Test(expected = SchemaNotFoundException.class) + public void schemaNotFound(){ + this.mockRestServiceServer.expect(requestTo("http://localhost:8081/subjects/user/versions/1")) + .andExpect(method(HttpMethod.GET)) + .andExpect(header("Content-Type","application/vnd.schemaregistry.v1+json")) + .andExpect(header("Accept","application/vnd.schemaregistry.v1+json")) + .andRespond(withStatus(HttpStatus.NOT_FOUND)); + ConfluentSchemaRegistryClient client = new ConfluentSchemaRegistryClient(this.restTemplate); + SchemaReference reference = new SchemaReference("user",1,"avro"); + String schema = client.fetch(reference); + } + +}