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

Fix brokent REST calls to Confluent Schema Registry

Browse Source 
The REST calls used to register and fetch schema were wrong in the controller.
Fixing those and the corresponding tests
This commit is contained in:
Soby Chacko
2019-09-19 10:29:37 -04:00
parent af81f8dd9e
commit 740c811d6c
2 changed files with 32 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<String, String> 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<String> request = new HttpEntity<>(payload, headers);
ResponseEntity<Map> response = this.template.exchange(this.endpoint,
ResponseEntity<Map> 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<List> 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");

17
spring-cloud-schema-registry-client/src/test/java/org/springframework/cloud/schema/avro/client/ConfluentSchemaRegistryClientTests.java
View File

@@ -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"))