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

Fix URLs used by ConfluentSchemaRegistryClient

Browse Source 
Fix #845

Fixed resource path for fetch by SchemaReference in the ConfluentSchemaRegistryClient.
This commit is contained in:
Julian Hanhart
2017-03-06 18:57:16 +01:00
committed by Vinicius Carvalho
parent d78adea895
commit d250022250
1 changed files with 33 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -23,13 +23,16 @@ import java.util.Map;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.cloud.stream.schema.SchemaNotFoundException;
import org.springframework.cloud.stream.schema.SchemaReference;
import org.springframework.cloud.stream.schema.SchemaRegistrationResponse;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
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.RestTemplate;
/**
@@ -81,16 +84,28 @@ public class ConfluentSchemaRegistryClient implements SchemaRegistryClient {
@Override
public String fetch(SchemaReference schemaReference) {
String path = String.format("/schemas/ids/%d", schemaReference.getVersion());
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.add("Content-Type", "application/vnd.schemaregistry.v1+json");
HttpEntity<String> request = new HttpEntity<>("", headers);
ResponseEntity<Map> response = this.template.exchange(this.endpoint + path, HttpMethod.GET, request, Map
.class);
return (String) response.getBody().get("schema");
try {
ResponseEntity<Map> response = this.template.exchange(this.endpoint + path, HttpMethod.GET, request,
Map.class);
return (String) response.getBody().get("schema");
}
catch (HttpClientErrorException e) {
if (e.getStatusCode() == HttpStatus.NOT_FOUND) {
throw new SchemaNotFoundException(
String.format("Could not find schema for reference: %s", schemaReference));
}
else {
throw e;
}
}
}
@Override
@@ -102,8 +117,19 @@ public class ConfluentSchemaRegistryClient implements SchemaRegistryClient {
"application/json"));
headers.add("Content-Type", "application/vnd.schemaregistry.v1+json");
HttpEntity<String> request = new HttpEntity<>("", headers);
ResponseEntity<Map> response = this.template.exchange(this.endpoint + path, HttpMethod.GET, request, Map
.class);
return (String) response.getBody().get("schema");
try {
ResponseEntity<Map> response = this.template.exchange(this.endpoint + path, HttpMethod.GET, request,
Map.class);
return (String) response.getBody().get("schema");
}
catch (HttpClientErrorException e) {
if (e.getStatusCode() == HttpStatus.NOT_FOUND) {
throw new SchemaNotFoundException(
String.format("Could not find schema with id: %s", id));
}
else {
throw e;
}
}
}
}