From 8f627f041d3f230e70700944cf979e48aa97c0b1 Mon Sep 17 00:00:00 2001 From: nicolas Date: Wed, 4 Nov 2020 23:49:10 -0500 Subject: [PATCH 1/2] [GH-48] rely on schema from response --- .../registry/SchemaRegistrationResponse.java | 37 ++++++++++++------- ...oSchemaRegistryClientMessageConverter.java | 7 +++- .../client/ConfluentSchemaRegistryClient.java | 4 +- .../client/DefaultSchemaRegistryClient.java | 1 + 4 files changed, 33 insertions(+), 16 deletions(-) diff --git a/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/SchemaRegistrationResponse.java b/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/SchemaRegistrationResponse.java index 200b5a0..abe4135 100644 --- a/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/SchemaRegistrationResponse.java +++ b/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/SchemaRegistrationResponse.java @@ -21,24 +21,33 @@ */ public class SchemaRegistrationResponse { - private int id; + private int id; - private SchemaReference schemaReference; + private SchemaReference schemaReference; - public int getId() { - return this.id; - } + private String schema; - public void setId(int id) { - this.id = id; - } + public int getId() { + return this.id; + } - public SchemaReference getSchemaReference() { - return this.schemaReference; - } + public void setId(int id) { + this.id = id; + } - public void setSchemaReference(SchemaReference schemaReference) { - this.schemaReference = schemaReference; - } + public SchemaReference getSchemaReference() { + return this.schemaReference; + } + public void setSchemaReference(SchemaReference schemaReference) { + this.schemaReference = schemaReference; + } + + public String getSchema() { + return schema; + } + + public void setSchema(String schema) { + this.schema = schema; + } } diff --git a/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/avro/AvroSchemaRegistryClientMessageConverter.java b/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/avro/AvroSchemaRegistryClientMessageConverter.java index 3bf9cab..39df7ef 100644 --- a/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/avro/AvroSchemaRegistryClientMessageConverter.java +++ b/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/avro/AvroSchemaRegistryClientMessageConverter.java @@ -308,8 +308,13 @@ protected Schema resolveSchemaForWriting(Object payload, MessageHeaders headers, if (parsedSchema.getRegistration() == null) { SchemaRegistrationResponse response = this.schemaRegistryClient.register(toSubject(this.subjectNamePrefix, schema), AVRO_FORMAT, parsedSchema.getRepresentation()); - parsedSchema.setRegistration(response); + // rely on schema in response + schema = new Schema.Parser().parse(response.getSchema()); + parsedSchema = new ParsedSchema(schema); + this.getCache(REFERENCE_CACHE_NAME).putIfAbsent(schema, parsedSchema); + + parsedSchema.setRegistration(response); } SchemaReference schemaReference = parsedSchema.getRegistration().getSchemaReference(); diff --git a/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/client/ConfluentSchemaRegistryClient.java b/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/client/ConfluentSchemaRegistryClient.java index bce8f06..9d3efbe 100644 --- a/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/client/ConfluentSchemaRegistryClient.java +++ b/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/client/ConfluentSchemaRegistryClient.java @@ -83,6 +83,8 @@ public SchemaRegistrationResponse register(String subject, String format, String String payload = null; Map maps = new HashMap<>(); maps.put("schema", schema); + SchemaRegistrationResponse schemaRegistrationResponse = new SchemaRegistrationResponse(); + schemaRegistrationResponse.setSchema(schema); try { payload = this.mapper.writeValueAsString(maps); } @@ -107,6 +109,7 @@ public SchemaRegistrationResponse register(String subject, String format, String final List body = response.getBody(); if (!CollectionUtils.isEmpty(body)) { version = (Integer) body.get(body.size() - 1); + schemaRegistrationResponse.setSchema(fetch(new SchemaReference(subject, version, "avro"))); } } catch (HttpStatusCodeException httpException) { @@ -114,7 +117,6 @@ public SchemaRegistrationResponse register(String subject, String format, String subject, httpException.getStatusCode().value()), httpException); } - SchemaRegistrationResponse schemaRegistrationResponse = new SchemaRegistrationResponse(); schemaRegistrationResponse.setId(id); schemaRegistrationResponse.setSchemaReference(new SchemaReference(subject, version, "avro")); return schemaRegistrationResponse; diff --git a/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/client/DefaultSchemaRegistryClient.java b/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/client/DefaultSchemaRegistryClient.java index e401b3d..4256d46 100644 --- a/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/client/DefaultSchemaRegistryClient.java +++ b/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/client/DefaultSchemaRegistryClient.java @@ -73,6 +73,7 @@ public SchemaRegistrationResponse register(String subject, String format, String registrationResponse.setId((Integer) responseBody.get("id")); registrationResponse.setSchemaReference(new SchemaReference(subject, (Integer) responseBody.get("version"), responseBody.get("format").toString())); + registrationResponse.setSchema(schema); return registrationResponse; } throw new RuntimeException( From d582b58627181e0f2336c89d54e8e880f42370d1 Mon Sep 17 00:00:00 2001 From: nicolas Date: Wed, 4 Nov 2020 23:53:34 -0500 Subject: [PATCH 2/2] override previously cached value --- .../registry/avro/AvroSchemaRegistryClientMessageConverter.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/avro/AvroSchemaRegistryClientMessageConverter.java b/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/avro/AvroSchemaRegistryClientMessageConverter.java index 39df7ef..5643b99 100644 --- a/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/avro/AvroSchemaRegistryClientMessageConverter.java +++ b/spring-cloud-schema-registry-client/src/main/java/org/springframework/cloud/schema/registry/avro/AvroSchemaRegistryClientMessageConverter.java @@ -312,7 +312,7 @@ protected Schema resolveSchemaForWriting(Object payload, MessageHeaders headers, // rely on schema in response schema = new Schema.Parser().parse(response.getSchema()); parsedSchema = new ParsedSchema(schema); - this.getCache(REFERENCE_CACHE_NAME).putIfAbsent(schema, parsedSchema); + this.getCache(REFERENCE_CACHE_NAME).put(schema, parsedSchema); parsedSchema.setRegistration(response); }