SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix Confluent Schema Registry registration

Browse Source 
Changing internal registration logic

Adding some tests

Resolves #985 #1014
This commit is contained in:
Vinicius Carvalho
2017-07-11 19:29:57 -04:00
committed by Ilayaperumal Gopinathan
parent 965945c275
commit 661f8494a3
2 changed files with 241 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

111
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/ConfluentSchemaRegistryClient.java
View File

@@ -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<String> request = new HttpEntity<>(payload, headers);
ResponseEntity<Map> 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<String> request = new HttpEntity<>(payload, headers);
ResponseEntity<Map> response = this.template.exchange(this.endpoint + path, HttpMethod.POST, request,
Map.class);
ResponseEntity<Map> 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<String> request = new HttpEntity<>("", headers);
try {
ResponseEntity<Map> response = this.template.exchange(this.endpoint + path, HttpMethod.GET, request,
Map.class);
ResponseEntity<Map> 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<String> request = new HttpEntity<>("", headers);
try {
ResponseEntity<Map> response = this.template.exchange(this.endpoint + path, HttpMethod.GET, request,
Map.class);
ResponseEntity<Map> 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));

160
spring-cloud-stream-schema/src/test/java/org/springframework/cloud/schema/avro/client/ConfluentSchemaRegistryClientTests.java Normal file
View File

@@ -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);
}
}