From b1285c3d4aa994ae65094c5dac1aca894d7438a5 Mon Sep 17 00:00:00 2001 From: Jon Archer Date: Fri, 5 May 2017 14:52:19 -0600 Subject: [PATCH] Extract schema version not id in register method. Fix #965 Original implementation was getting the schema id and using it as the version. This made deserialization fail since it would endeavor to retrieve the schema by version using the id value. --- .../client/ConfluentSchemaRegistryClient.java | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/ConfluentSchemaRegistryClient.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/ConfluentSchemaRegistryClient.java index 14858cfc5..956c03b93 100644 --- a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/ConfluentSchemaRegistryClient.java +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/ConfluentSchemaRegistryClient.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-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. @@ -38,6 +38,7 @@ import org.springframework.web.client.RestTemplate; /** * @author Vinicius Carvalho * @author Marius Bogoevici + * @author Jon Archer */ public class ConfluentSchemaRegistryClient implements SchemaRegistryClient { @@ -59,26 +60,26 @@ public class ConfluentSchemaRegistryClient implements SchemaRegistryClient { @Override 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); + 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.add("Content-Type", "application/json"); - Integer id = null; + Integer version = null; try { String payload = this.mapper.writeValueAsString(Collections.singletonMap("schema", schema)); HttpEntity request = new HttpEntity<>(payload, headers); ResponseEntity response = this.template.exchange(this.endpoint + path, HttpMethod.POST, request, Map.class); - id = (Integer) response.getBody().get("id"); + version = (Integer) response.getBody().get("version"); } catch (JsonProcessingException e) { e.printStackTrace(); } SchemaRegistrationResponse schemaRegistrationResponse = new SchemaRegistrationResponse(); - schemaRegistrationResponse.setId(id); - schemaRegistrationResponse.setSchemaReference(new SchemaReference(subject, id, "avro")); + schemaRegistrationResponse.setId(version); + schemaRegistrationResponse.setSchemaReference(new SchemaReference(subject, version, "avro")); return schemaRegistrationResponse; }