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

Migrating PR from core Spring Cloud Stream

Browse Source 
0ad15247f6
This commit is contained in:
Soby Chacko
2019-08-19 19:08:09 -04:00
parent 9a0307a7e0
commit af81f8dd9e
2 changed files with 527 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
spring-cloud-schema-registry-core/src/main/java/org/springframework/cloud/schema/registry/controllers/ServerController.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-2019 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.
@@ -31,9 +31,11 @@ import org.springframework.cloud.schema.registry.support.UnsupportedFormatExcept
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.lang.NonNull;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
@@ -42,9 +44,12 @@ import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.util.UriComponentsBuilder;
import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;
/**
* @author Vinicius Carvalho
* @author Ilayaperumal Gopinathan
* @author Jeff Maxwell
*/
@RestController
@RequestMapping(path = "${spring.cloud.stream.schema.server.path:}")
@@ -57,8 +62,8 @@ public class ServerController {
private final SchemaServerProperties schemaServerProperties;
public ServerController(SchemaRepository repository,
Map<String, SchemaValidator> validators,
SchemaServerProperties schemaServerProperties) {
Map<String, SchemaValidator> validators,
SchemaServerProperties schemaServerProperties) {
Assert.notNull(repository, "cannot be null");
Assert.notEmpty(validators, "cannot be empty");
this.repository = repository;
@@ -68,7 +73,7 @@ public class ServerController {
@RequestMapping(method = RequestMethod.POST, path = "/", consumes = "application/json", produces = "application/json")
public synchronized ResponseEntity<Schema> register(@RequestBody Schema schema,
UriComponentsBuilder builder) {
UriComponentsBuilder builder) {
SchemaValidator validator = this.validators.get(schema.getFormat());
if (validator == null) {
@@ -85,7 +90,7 @@ public class ServerController {
List<Schema> registeredEntities = this.repository
.findBySubjectAndFormatOrderByVersion(schema.getSubject(),
schema.getFormat());
if (registeredEntities == null || registeredEntities.size() == 0) {
if (registeredEntities.isEmpty()) {
schema.setVersion(1);
result = this.repository.save(schema);
}
@@ -115,8 +120,8 @@ public class ServerController {
@RequestMapping(method = RequestMethod.GET, produces = "application/json", path = "/{subject}/{format}/v{version}")
public ResponseEntity<Schema> findOne(@PathVariable("subject") String subject,
@PathVariable("format") String format,
@PathVariable("version") Integer version) {
@PathVariable("format") String format,
@PathVariable("version") Integer version) {
Schema schema = this.repository.findOneBySubjectAndFormatAndVersion(subject,
format, version);
if (schema == null) {
@@ -134,23 +139,51 @@ public class ServerController {
return new ResponseEntity<>(schema.get(), HttpStatus.OK);
}
@RequestMapping(method = RequestMethod.GET, produces = "application/json", path = "/{subject}/{format}")
public ResponseEntity<List<Schema>> findBySubjectAndVersion(
@PathVariable("subject") String subject,
@PathVariable("format") String format) {
List<Schema> schemas = this.repository
.findBySubjectAndFormatOrderByVersion(subject, format);
if (schemas == null || schemas.size() == 0) {
throw new SchemaNotFoundException(String.format(
"No schemas found for subject %s and format %s", subject, format));
}
return new ResponseEntity<List<Schema>>(schemas, HttpStatus.OK);
/**
* <p>
* Find by {@link Schema#getSubject() subject} and {@link Schema#getFormat() format}.
*
* @param subject the {@link Schema#getSubject() subject}, must not be
* {@literal null}.
* @param format the {@link Schema#getFormat() format}, must not be {@literal null}.
* @return An {@link HttpStatus#OK} response populated with the list of {@link Schema
* Schemas}, in ascending order by {@link Schema#getVersion() version}, that matched
* the supplied {@link Schema#getSubject() subject} and {@link Schema#getFormat()
* format}.
* @deprecated use {@link #findBySubjectAndFormat(String, String)}
* @see <a href=
* "https://github.com/spring-cloud/spring-cloud-stream/issues/1760">GH-1760</a>
*/
@Deprecated
public ResponseEntity<List<Schema>> findBySubjectAndVersion(@PathVariable("subject") String subject,
@PathVariable("format") String format) {
return findBySubjectAndFormatOrderByVersionAsc(subject, format);
}
/**
* Find by {@link Schema#getSubject() subject} and {@link Schema#getFormat() format}.
*
* @param subject the {@link Schema#getSubject() subject}, must not be
* {@literal null}.
* @param format the {@link Schema#getFormat() format}, must not be {@literal null}.
* @return An {@link HttpStatus#OK} response populated with the list of {@link Schema
* Schemas}, in ascending order by {@link Schema#getVersion() version}, that matched
* the supplied {@link Schema#getSubject() subject} and {@link Schema#getFormat()
* format}.
*
* @since 3.0.0
*/
@GetMapping(produces = APPLICATION_JSON_VALUE, path = "/{subject}/{format}")
@NonNull
public ResponseEntity<List<Schema>> findBySubjectAndFormat(@NonNull @PathVariable("subject") final String subject,
@NonNull @PathVariable("format") final String format) {
return findBySubjectAndFormatOrderByVersionAsc(subject, format);
}
@RequestMapping(value = "/{subject}/{format}/v{version}", method = RequestMethod.DELETE)
public void delete(@PathVariable("subject") String subject,
@PathVariable("format") String format,
@PathVariable("version") Integer version) {
@PathVariable("format") String format,
@PathVariable("version") Integer version) {
if (this.schemaServerProperties.isAllowSchemaDeletion()) {
Schema schema = this.repository.findOneBySubjectAndFormatAndVersion(subject,
format, version);
@@ -190,6 +223,17 @@ public class ServerController {
}
@NonNull
public final ResponseEntity<List<Schema>> findBySubjectAndFormatOrderByVersionAsc(@NonNull final String subject,
@NonNull final String format) {
List<Schema> schemas = this.repository.findBySubjectAndFormatOrderByVersion(subject, format);
if (schemas.isEmpty()) {
throw new SchemaNotFoundException(
String.format("No schemas found for subject %s and format %s", subject, format));
}
return new ResponseEntity<>(schemas, HttpStatus.OK);
}
private void deleteSchema(Schema schema) {
if (schema == null) {
throw new SchemaNotFoundException("Could not find Schema");

615
spring-cloud-schema-registry-server/src/test/java/org/springframework/cloud/schema/registry/server/SchemaRegistryServerAvroTests.java
View File

@@ -16,25 +16,46 @@
package org.springframework.cloud.schema.registry.server;
import java.net.URI;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.stream.Stream;
import org.assertj.core.api.Assertions;
import org.apache.avro.Schema.Parser;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.web.ServerProperties;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.web.client.TestRestTemplate;
import org.springframework.boot.web.server.Ssl;
import org.springframework.cloud.schema.registry.config.SchemaServerProperties;
import org.springframework.cloud.schema.registry.controllers.ServerController;
import org.springframework.cloud.schema.registry.model.Schema;
import org.springframework.cloud.schema.registry.support.SchemaNotFoundException;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.RequestEntity;
import org.springframework.http.ResponseEntity;
import org.springframework.lang.NonNull;
import org.springframework.lang.Nullable;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.web.context.WebApplicationContext;
import org.springframework.web.util.UriComponentsBuilder;
import static java.util.stream.Collectors.toList;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.springframework.test.annotation.DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD;
/**
* @author Vinicius Carvalho
@@ -44,22 +65,49 @@ import org.springframework.web.context.WebApplicationContext;
// @checkstyle:off
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.DEFINED_PORT, properties = "spring.main.allow-bean-definition-overriding=true")
// @checkstyle:on
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
@DirtiesContext(classMode = AFTER_EACH_TEST_METHOD)
public class SchemaRegistryServerAvroTests {
final String USER_SCHEMA_V1 = "{\"namespace\": \"example.avro\",\n"
private static final String AVRO_FORMAT_NAME = "avro";
private static final String AVRO_USER_DEFINITION_SCHEMA_V1 = "{\"namespace\": \"example.avro\",\n"
+ " \"type\": \"record\",\n" + " \"name\": \"User\",\n" + " \"fields\": [\n"
+ " {\"name\": \"name\", \"type\": \"string\"},\n"
+ " {\"name\": \"favorite_number\", \"type\": [\"int\", \"null\"]}\n"
+ " ]\n" + "}";
final String USER_SCHEMA_V2 = "{\"namespace\": \"example.avro\",\n"
private static final String AVRO_USER_DEFINTITION_SCHEMA_V2 = "{\"namespace\": \"example.avro\",\n"
+ " \"type\": \"record\",\n" + " \"name\": \"User\",\n" + " \"fields\": [\n"
+ " {\"name\": \"name\", \"type\": \"string\"},\n"
+ " {\"name\": \"favorite_number\", \"type\": [\"int\", \"null\"]},\n"
+ " {\"name\": \"favorite_color\", \"type\": [\"string\", \"null\"]}\n"
+ " ]\n" + "}";
private static final org.apache.avro.Schema AVRO_USER_AVRO_SCHEMA_V1 = new Parser()
.parse(AVRO_USER_DEFINITION_SCHEMA_V1);
private static final org.apache.avro.Schema AVRO_USER_AVRO_SCHEMA_V2 = new Parser()
.parse(AVRO_USER_DEFINTITION_SCHEMA_V2);
private static final String AVRO_USER_SCHEMA_DEFAULT_NAME_STRATEGY_SUBJECT = AVRO_USER_AVRO_SCHEMA_V1.getName()
.toLowerCase();
private static final String AVRO_USER_SCHEMA_QUALIFED_NAME_STRATEGY_SUBJECT = AVRO_USER_AVRO_SCHEMA_V1
.getFullName()
.toLowerCase();
private static final Schema AVRO_USER_REGISTRY_SCHEMA_V1 = toSchema(
AVRO_USER_SCHEMA_DEFAULT_NAME_STRATEGY_SUBJECT,
AVRO_FORMAT_NAME, AVRO_USER_AVRO_SCHEMA_V1.toString());
private static final Schema AVRO_USER_REGISTRY_SCHEMA_V2 = toSchema(
AVRO_USER_SCHEMA_DEFAULT_NAME_STRATEGY_SUBJECT,
AVRO_FORMAT_NAME, AVRO_USER_AVRO_SCHEMA_V2.toString());
private static final Schema AAVRO_USER_REGISTRY_SCHEMA_V1_WITH_QUAL_SUBJECT = toSchema(
AVRO_USER_SCHEMA_QUALIFED_NAME_STRATEGY_SUBJECT,
AVRO_FORMAT_NAME, AVRO_USER_AVRO_SCHEMA_V1.toString());
@Autowired
private TestRestTemplate client;
@@ -67,7 +115,38 @@ public class SchemaRegistryServerAvroTests {
private SchemaServerProperties schemaServerProperties;
@Autowired
private WebApplicationContext wac;
private ServerController serverController;
@Autowired
private ServerProperties serverProperties;
private URI serverControllerUri;
@Before
public void setUp() {
String scheme = Optional.ofNullable(this.serverProperties.getSsl())
.filter(Ssl::isEnabled)
.map(ssl -> "https").orElse("http");
Integer port = this.serverProperties.getPort();
String contextPath = this.serverProperties.getServlet().getContextPath();
this.serverControllerUri = UriComponentsBuilder.newInstance().scheme(scheme)
.host("localhost")
.port(port)
.path(contextPath).build().toUri();
}
@NonNull
static Schema toSchema(String subject, String format, String definition) {
Schema schema = new Schema();
schema.setSubject(subject);
schema.setFormat(format);
schema.setDefinition(definition);
return schema;
}
@Test
public void testUnsupportedFormat() throws Exception {
@@ -75,83 +154,84 @@ public class SchemaRegistryServerAvroTests {
schema.setFormat("spring");
schema.setSubject("boot");
ResponseEntity<Schema> response = this.client
.postForEntity("http://localhost:8990/", schema, Schema.class);
Assertions.assertThat(response.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST);
.postForEntity(this.serverControllerUri, schema, Schema.class);
assertThat(response.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST);
}
@Test
public void testInvalidSchema() throws Exception {
Schema schema = new Schema();
schema.setFormat("avro");
schema.setFormat(AVRO_FORMAT_NAME);
schema.setSubject("boot");
schema.setDefinition("{}");
ResponseEntity<Schema> response = this.client
.postForEntity("http://localhost:8990/", schema, Schema.class);
Assertions.assertThat(response.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST);
.postForEntity(this.serverControllerUri, schema, Schema.class);
assertThat(response.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST);
}
@Test
public void testUserSchemaV1() throws Exception {
public void testRegister1AvroSchema() {
Schema schema = new Schema();
schema.setFormat("avro");
schema.setFormat(AVRO_FORMAT_NAME);
schema.setSubject("org.springframework.cloud.stream.schema.User");
schema.setDefinition(this.USER_SCHEMA_V1);
ResponseEntity<Schema> response = this.client
.postForEntity("http://localhost:8990/", schema, Schema.class);
Assertions.assertThat(response.getStatusCode().is2xxSuccessful()).isTrue();
Assertions.assertThat(response.getBody().getVersion()).isEqualTo(new Integer(1));
List<String> location = response.getHeaders().get(HttpHeaders.LOCATION);
Assertions.assertThat(location).isNotNull();
ResponseEntity<Schema> persistedSchema = this.client.getForEntity(location.get(0),
Schema.class);
Assertions.assertThat(persistedSchema.getBody().getId())
.isEqualTo(response.getBody().getId());
schema.setDefinition(SchemaRegistryServerAvroTests.AVRO_USER_DEFINITION_SCHEMA_V1);
registerSchemaAndAssertSuccess(schema, 1, 1);
}
@Test
public void testUserSchemaV2() throws Exception {
Schema schema = new Schema();
schema.setFormat("avro");
schema.setSubject("org.springframework.cloud.stream.schema.User");
schema.setDefinition(this.USER_SCHEMA_V1);
public void testFindByIdFound() {
Schema schema2 = new Schema();
schema2.setFormat("avro");
schema2.setSubject("org.springframework.cloud.stream.schema.User");
schema2.setDefinition(this.USER_SCHEMA_V2);
ResponseEntity<Schema> registerSchemaReponse = registerSchemaAndAssertSuccess(
AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1);
ResponseEntity<Schema> response = this.client
.postForEntity("http://localhost:8990/", schema, Schema.class);
Assertions.assertThat(response.getStatusCode().is2xxSuccessful()).isTrue();
Assertions.assertThat(response.getBody().getVersion()).isEqualTo(new Integer(1));
List<String> location = response.getHeaders().get(HttpHeaders.LOCATION);
Assertions.assertThat(location).isNotNull();
Schema registeredSchema = registerSchemaReponse.getBody();
ResponseEntity<Schema> response2 = this.client
.postForEntity("http://localhost:8990/", schema2, Schema.class);
Assertions.assertThat(response.getStatusCode().is2xxSuccessful()).isTrue();
Assertions.assertThat(response2.getBody().getVersion()).isEqualTo(new Integer(2));
List<String> location2 = response2.getHeaders().get(HttpHeaders.LOCATION);
Assertions.assertThat(location2).isNotNull();
URI findByIdUriId1 = this.serverControllerUri.resolve("/schemas/" + registeredSchema.getId());
ResponseEntity<Schema> findByIdResponse = this.client
.getForEntity(findByIdUriId1, Schema.class);
assertThat(findByIdResponse.getStatusCode().is2xxSuccessful()).isTrue();
Schema actual = findByIdResponse.getBody();
assertSchema(registeredSchema, actual);
}
@Test
public void testIdempotentRegistration() throws Exception {
Schema schema = new Schema();
schema.setFormat("avro");
schema.setSubject("org.springframework.cloud.stream.schema.User");
schema.setDefinition(this.USER_SCHEMA_V1);
public void testFindByIdNotFound() {
registerSchemaAndAssertSuccess(AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1);
URI findByIdUriId1 = this.serverControllerUri.resolve("/schemas/" + 2);
ResponseEntity<Schema> response = this.client
.postForEntity("http://localhost:8990/", schema, Schema.class);
Assertions.assertThat(response.getStatusCode().is2xxSuccessful()).isTrue();
Assertions.assertThat(response.getBody().getVersion()).isEqualTo(new Integer(1));
List<String> location = response.getHeaders().get(HttpHeaders.LOCATION);
Assertions.assertThat(location).isNotNull();
ResponseEntity<Schema> response2 = this.client
.postForEntity("http://localhost:8990/", schema, Schema.class);
Assertions.assertThat(response2.getBody().getId()).isEqualTo(response.getBody().getId());
.getForEntity(findByIdUriId1, Schema.class);
final HttpStatus statusCode = response.getStatusCode();
assertThat(statusCode).isEqualTo(HttpStatus.NOT_FOUND);
}
@Test
public void testUserSchemaV2() {
registerSchemasAndAssertSuccess(
AVRO_USER_REGISTRY_SCHEMA_V1,
AVRO_USER_REGISTRY_SCHEMA_V2);
}
@Test
public void testIdempotentRegistration() {
registerSchemaAndAssertSuccess(
AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1);
registerSchemaAndAssertSuccess(
AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1);
}
@@ -159,126 +239,357 @@ public class SchemaRegistryServerAvroTests {
public void testSchemaNotfound() throws Exception {
ResponseEntity<Schema> response = this.client
.getForEntity("http://localhost:8990/foo/avro/v42", Schema.class);
Assertions.assertThat(response.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND);
assertThat(response.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND);
}
@Test
public void testSchemaDeletionBySubjectFormatVersion() throws Exception {
Schema schema = new Schema();
schema.setFormat("avro");
schema.setSubject("test");
schema.setDefinition(this.USER_SCHEMA_V1);
ResponseEntity<Schema> response1 = this.client
.postForEntity("http://localhost:8990/", schema, Schema.class);
Assertions.assertThat(response1.getStatusCode().is2xxSuccessful()).isTrue();
ResponseEntity<Schema> registerSchemaAndAssertSuccess = registerSchemaAndAssertSuccess(
AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1);
this.schemaServerProperties.setAllowSchemaDeletion(true);
this.client.delete("http://localhost:8990/test/avro/v1");
ResponseEntity<Schema> response2 = this.client
.getForEntity("http://localhost:8990/test/avro/v1", Schema.class);
Assertions.assertThat(response2.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND);
URI subjectFormatVersionUri = this.serverControllerUri
.resolve(registerSchemaAndAssertSuccess.getHeaders().getLocation());
ResponseEntity<Void> deleteResponse = this.client.exchange(
new RequestEntity<>(HttpMethod.DELETE, subjectFormatVersionUri),
Void.class);
assertThat(deleteResponse.getStatusCode()).isEqualTo(HttpStatus.OK);
ResponseEntity<Schema> findBySubjectFormatVersionUriResponse = this.client
.getForEntity(subjectFormatVersionUri, Schema.class);
assertThat(findBySubjectFormatVersionUriResponse.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND);
}
@Test
public void testSchemaDeletionBySubjectFormatVersionNotFound() throws Exception {
ResponseEntity<Schema> registerSchemaAndAssertSuccess = registerSchemaAndAssertSuccess(
AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1);
this.schemaServerProperties.setAllowSchemaDeletion(true);
URI subjectFormatVersionUri = this.serverControllerUri
.resolve(registerSchemaAndAssertSuccess.getHeaders().getLocation().toString().replace("v1", "v100"));
ResponseEntity<Void> deleteResponse = this.client.exchange(
new RequestEntity<>(HttpMethod.DELETE, subjectFormatVersionUri),
Void.class);
assertThat(deleteResponse.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND);
}
@Test
public void testSchemaDeletionBySubjectFormatVersionNotAllowed() throws Exception {
ResponseEntity<Schema> registerSchemaAndAssertSuccess = registerSchemaAndAssertSuccess(
AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1);
URI versionUri = this.serverControllerUri
.resolve(registerSchemaAndAssertSuccess.getHeaders().getLocation());
ResponseEntity<Void> deleteResponse = this.client.exchange(new RequestEntity<>(HttpMethod.DELETE, versionUri),
Void.class);
assertThat(deleteResponse.getStatusCode()).isEqualTo(HttpStatus.METHOD_NOT_ALLOWED);
}
@Test
public void testSchemaDeletionById() throws Exception {
Schema schema = new Schema();
schema.setFormat("avro");
schema.setSubject("test");
schema.setDefinition(this.USER_SCHEMA_V1);
ResponseEntity<Schema> response1 = this.client
.postForEntity("http://localhost:8990/", schema, Schema.class);
Assertions.assertThat(response1.getStatusCode().is2xxSuccessful()).isTrue();
ResponseEntity<Schema> response2 = this.client
.getForEntity("http://localhost:8990/test/avro/v1", Schema.class);
Assertions.assertThat(response2.getStatusCode()).isEqualTo(HttpStatus.OK);
ResponseEntity<Schema> registerSchemaAndAssertSuccess = registerSchemaAndAssertSuccess(
AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1);
this.schemaServerProperties.setAllowSchemaDeletion(true);
this.client.delete("http://localhost:8990/schemas/1");
this.client.delete(this.serverControllerUri
.resolve("/schemas/" + registerSchemaAndAssertSuccess.getBody().getVersion()));
ResponseEntity<Schema> response3 = this.client
.getForEntity("http://localhost:8990/test/avro/1", Schema.class);
Assertions.assertThat(response3.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND);
.getForEntity(registerSchemaAndAssertSuccess.getHeaders().getLocation(), Schema.class);
assertThat(response3.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND);
}
@Test
public void testSchemaDeletionBySubject() throws Exception {
Schema schema1 = new Schema();
schema1.setFormat("avro");
schema1.setSubject("test");
schema1.setDefinition(this.USER_SCHEMA_V1);
ResponseEntity<Schema> response1 = this.client
.postForEntity("http://localhost:8990/", schema1, Schema.class);
Assertions.assertThat(response1.getStatusCode().is2xxSuccessful()).isTrue();
Assertions.assertThat(this.client
.getForEntity("http://localhost:8990/test/avro/v1", Schema.class)
.getStatusCode()).isEqualTo(HttpStatus.OK);
this.client.getForEntity("http://localhost:8990/test/avro/1", Schema.class);
Schema schema2 = new Schema();
schema2.setFormat("avro");
schema2.setSubject("test");
schema2.setDefinition(this.USER_SCHEMA_V2);
ResponseEntity<Schema> response2 = this.client
.postForEntity("http://localhost:8990/", schema2, Schema.class);
Assertions.assertThat(response2.getStatusCode().is2xxSuccessful()).isTrue();
Assertions.assertThat(this.client
.getForEntity("http://localhost:8990/test/avro/v2", Schema.class)
.getStatusCode()).isEqualTo(HttpStatus.OK);
public void testSchemaDeletionByIdNotFound() throws Exception {
registerSchemaAndAssertSuccess(
AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1);
this.schemaServerProperties.setAllowSchemaDeletion(true);
this.client.delete("http://localhost:8990/test");
ResponseEntity<Schema> response4 = this.client
.getForEntity("http://localhost:8990/test/avro/v1", Schema.class);
Assertions.assertThat(response4.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND);
ResponseEntity<Schema> response5 = this.client
.getForEntity("http://localhost:8990/test/avro/v2", Schema.class);
Assertions.assertThat(response5.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND);
ResponseEntity<Void> deleteByIdResponse = this.client.exchange(
new RequestEntity<>(HttpMethod.DELETE, this.serverControllerUri
.resolve("/schemas/" + 2)),
Void.class);
assertThat(deleteByIdResponse.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND);
}
@Test
public void testSchemaDeletionNotAllowed() throws Exception {
Schema schema = new Schema();
schema.setFormat("avro");
schema.setSubject("test");
schema.setDefinition(this.USER_SCHEMA_V1);
ResponseEntity<Schema> response1 = this.client
.postForEntity("http://localhost:8990/", schema, Schema.class);
Assertions.assertThat(response1.getStatusCode().is2xxSuccessful()).isTrue();
ResponseEntity<Object> deleteBySubjectFormatVersion = this.client.exchange(
"http://localhost:8990/test/avro/v1", HttpMethod.DELETE, null,
Object.class);
Assertions.assertThat(deleteBySubjectFormatVersion.getStatusCode())
.isEqualTo(HttpStatus.METHOD_NOT_ALLOWED);
ResponseEntity<Object> deleteBySubject = this.client.exchange(
"http://localhost:8990/test", HttpMethod.DELETE, null, Object.class);
Assertions.assertThat(deleteBySubject.getStatusCode())
.isEqualTo(HttpStatus.METHOD_NOT_ALLOWED);
ResponseEntity<Object> deleteById = this.client.exchange(
"http://localhost:8990/schemas/1", HttpMethod.DELETE, null, Object.class);
Assertions.assertThat(deleteById.getStatusCode()).isEqualTo(HttpStatus.METHOD_NOT_ALLOWED);
public void testSchemaDeletionByIdNotAllowed() throws Exception {
ResponseEntity<Schema> registerSchemaAndAssertSuccess = registerSchemaAndAssertSuccess(
AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1);
URI schemaIdUri = this.serverControllerUri
.resolve(this.serverControllerUri
.resolve("/schemas/" + registerSchemaAndAssertSuccess.getBody().getVersion()));
ResponseEntity<Void> exchange = this.client.exchange(new RequestEntity<>(HttpMethod.DELETE, schemaIdUri),
Void.class);
assertThat(exchange.getStatusCode()).isEqualTo(HttpStatus.METHOD_NOT_ALLOWED);
}
@Test
public void testFindSchemasBySubjectAndVersion() throws Exception {
Schema v1 = new Schema();
v1.setFormat("avro");
v1.setSubject("test");
v1.setDefinition(this.USER_SCHEMA_V1);
ResponseEntity<Schema> response1 = this.client
.postForEntity("http://localhost:8990/", v1, Schema.class);
Assertions.assertThat(response1.getStatusCode().is2xxSuccessful()).isTrue();
public void testSchemaDeletionBySubject() {
Map<String, Map<String, List<ResponseEntity<Schema>>>> registerSchemaResponsesByFormatBySubject = registerSchemasAndAssertSuccess(
AVRO_USER_REGISTRY_SCHEMA_V1,
AVRO_USER_REGISTRY_SCHEMA_V2, AAVRO_USER_REGISTRY_SCHEMA_V1_WITH_QUAL_SUBJECT);
Schema v2 = new Schema();
v2.setFormat("avro");
v2.setSubject("test");
v2.setDefinition(this.USER_SCHEMA_V2);
this.schemaServerProperties.setAllowSchemaDeletion(true);
ResponseEntity<Schema> response2 = this.client
.postForEntity("http://localhost:8990/", v2, Schema.class);
Assertions.assertThat(response2.getStatusCode().is2xxSuccessful()).isTrue();
registerSchemaResponsesByFormatBySubject.forEach((subject, registerSchemaResponsesByFormat) -> {
assertThat(registerSchemaResponsesByFormat).isNotEmpty();
ResponseEntity<Void> deleteBySubject = this.client.exchange(
new RequestEntity<>(HttpMethod.DELETE, this.serverControllerUri
.resolve("/" + subject)),
Void.class);
assertThat(deleteBySubject.getStatusCode()).isEqualTo(HttpStatus.OK);
registerSchemaResponsesByFormat.forEach((format, registerSchemaResponses) -> {
assertThat(registerSchemaResponses).isNotEmpty();
registerSchemaResponses.forEach(registerSchemaResponse -> {
ResponseEntity<Schema> shouldBe404Response = this.client.getForEntity(
registerSchemaResponse.getHeaders().getLocation(),
Schema.class);
assertThat(shouldBe404Response.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND);
ResponseEntity<List<Schema>> schemaResponse = this.client.exchange(
"http://localhost:8990/test/avro", HttpMethod.GET, null,
new ParameterizedTypeReference<List<Schema>>() {
});
});
});
Assertions.assertThat(schemaResponse.getStatusCode().is2xxSuccessful()).isTrue();
Assertions.assertThat(schemaResponse.getBody().size()).isEqualTo(2);
}
@Test
public void testSchemaDeletionBySubjectNotFound() throws Exception {
registerSchemaAndAssertSuccess(
AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1);
this.schemaServerProperties.setAllowSchemaDeletion(true);
ResponseEntity<Void> deleteBySubject = this.client.exchange(
new RequestEntity<>(HttpMethod.DELETE, this.serverControllerUri
.resolve("/foo")),
Void.class);
assertThat(deleteBySubject.getStatusCode())
.isEqualTo(HttpStatus.OK);
}
@Test
public void testSchemaDeletionBySubjectNotAllowed() throws Exception {
ResponseEntity<Schema> registerSchemaAndAssertSuccess = registerSchemaAndAssertSuccess(
AVRO_USER_REGISTRY_SCHEMA_V1, 1, 1);
Schema schema = registerSchemaAndAssertSuccess.getBody();
ResponseEntity<Void> deleteBySubject = this.client.exchange(
new RequestEntity<>(HttpMethod.DELETE, this.serverControllerUri
.resolve("/" + schema.getSubject())),
Void.class);
assertThat(deleteBySubject.getStatusCode())
.isEqualTo(HttpStatus.METHOD_NOT_ALLOWED);
}
@Test
public void testFindSchemasBySubjectAndVersion() {
Map<String, Map<String, List<ResponseEntity<Schema>>>> registerSchemaResponsesByFormatBySubject = registerSchemasAndAssertSuccess(
AVRO_USER_REGISTRY_SCHEMA_V1,
AVRO_USER_REGISTRY_SCHEMA_V2);
registerSchemaResponsesByFormatBySubject.forEach((subject, schemasByFormat) -> {
assertThat(schemasByFormat).hasSize(1);
schemasByFormat.forEach((format, schemas) -> {
assertThat(schemas).hasSize(2);
@SuppressWarnings("deprecation")
final ResponseEntity<List<Schema>> findBySubjectAndVersionResponseEntity = this.serverController
.findBySubjectAndVersion(subject, format);
assertThat(findBySubjectAndVersionResponseEntity.getStatusCode().is2xxSuccessful()).isTrue();
final List<Schema> schemaResponseBody = findBySubjectAndVersionResponseEntity.getBody();
assertThat(schemaResponseBody)
.<Schema>zipSatisfy(schemas.stream().map(ResponseEntity::getBody)
.collect(toList()), this::assertSchema);
});
});
}
@Test
public void testFindBySubjectAndFormatOrderByVersionAscNoMatch() {
String subject = "test";
String format = AVRO_FORMAT_NAME;
assertThatExceptionOfType(SchemaNotFoundException.class).isThrownBy(() -> this.serverController
.findBySubjectAndFormatOrderByVersionAsc(subject, format))
.withMessage("No schemas found for subject %s and format %s", subject, format)
.withNoCause();
}
@Test
public void testFindSchemasBySubjectAndFormat() {
Map<String, Map<String, List<ResponseEntity<Schema>>>> registerSchemaResponsesByFormatBySubject = registerSchemasAndAssertSuccess(
AVRO_USER_REGISTRY_SCHEMA_V1,
AVRO_USER_REGISTRY_SCHEMA_V2);
registerSchemaResponsesByFormatBySubject.forEach((subject, schemasByFormat) -> {
assertThat(schemasByFormat).hasSize(1);
schemasByFormat.forEach((format, schemas) -> {
assertThat(schemas).hasSize(2);
ResponseEntity<List<Schema>> findBySubjectFormatResponse = this.client.exchange(
this.serverControllerUri.resolve("/" + subject + "/" + format), HttpMethod.GET, null,
new ParameterizedTypeReference<List<Schema>>() {
});
assertThat(findBySubjectFormatResponse.getStatusCode().is2xxSuccessful()).isTrue();
final List<Schema> schemaResponseBody = findBySubjectFormatResponse.getBody();
assertThat(schemaResponseBody)
.<Schema>zipSatisfy(schemas.stream().map(ResponseEntity::getBody)
.collect(toList()), this::assertSchema);
});
});
}
private Map<String, Map<String, List<ResponseEntity<Schema>>>> registerSchemasAndAssertSuccess(
@NonNull Schema... schemas) {
Map<String, Map<String, Integer>> versionsByFormatAndSubject = new HashMap<>();
Map<String, Map<String, List<ResponseEntity<Schema>>>> result = new HashMap<>();
int numOfSchemas = schemas.length;
int id = 0;
for (int i = 0; i < numOfSchemas; i++) {
Schema schema = schemas[i];
id++;
String format = schema.getFormat();
String subject = schema.getSubject();
Integer version = versionsByFormatAndSubject
.compute(subject,
(_subject, currentValue) -> currentValue == null ? new HashMap<>() : currentValue)
.merge(format, 1, Integer::sum);
ResponseEntity<Schema> registerSchemaResponse = registerSchemaAndAssertSuccess(schema, version, id);
result.compute(subject,
(_subject, currentValue) -> currentValue == null ? new HashMap<>() : currentValue)
.compute(format, (_format, currentValue) -> {
List<ResponseEntity<Schema>> value = currentValue == null ? new ArrayList<>() : currentValue;
value.add(registerSchemaResponse);
return value;
});
}
Stream<ResponseEntity<Schema>> asStream = result.entrySet().stream()
.map(Entry::getValue)
.map(Map::entrySet)
.flatMap(Collection::stream)
.map(Entry::getValue)
.flatMap(Collection::stream);
assertThat(asStream).hasSize(numOfSchemas);
return result;
}
@NonNull
private ResponseEntity<Schema> registerSchemaAndAssertSuccess(@NonNull Schema schema,
@Nullable Integer expectedVersion,
@Nullable Integer expectedId) {
ResponseEntity<Schema> registerReponse = this.client
.postForEntity(this.serverControllerUri, schema, Schema.class);
HttpStatus statusCode = registerReponse.getStatusCode();
assertThat(statusCode.is2xxSuccessful()).isTrue();
Schema registeredSchema = registerReponse.getBody();
assertSchema(schema, expectedVersion, expectedId, registeredSchema);
HttpHeaders headers = registerReponse.getHeaders();
assertLocation(headers, registeredSchema);
return registerReponse;
}
private void assertLocation(HttpHeaders headers, Schema registeredSchema) {
URI location = headers.getLocation();
assertThat(location).isNotNull();
assertPersisted(location, registeredSchema);
}
private void assertPersisted(URI location, Schema registeredSchema) {
ResponseEntity<Schema> findOneResponse = this.client.getForEntity(location,
Schema.class);
HttpStatus statusCode = findOneResponse.getStatusCode();
assertThat(statusCode.is2xxSuccessful()).isTrue();
Schema actual = findOneResponse.getBody();
assertSchema(registeredSchema, registeredSchema.getVersion(), registeredSchema.getId(), actual);
}
private void assertSchema(@NonNull Schema expected, @NonNull Schema actual) {
assertSchema(expected, expected.getVersion(), expected.getId(), actual);
}
private void assertSchema(@NonNull Schema expected, Integer expectedVersion, Integer expectedId,
@NonNull Schema actual) {
assertThat(actual).isEqualToIgnoringGivenFields(expected, "version", "id");
if (expectedVersion != null) {
assertThat(actual.getVersion()).isEqualTo(expectedVersion);
}
if (expectedId != null) {
assertThat(actual.getId()).isEqualTo(expectedId);
}
}
}