This is an automated email from the ASF dual-hosted git repository. nielsbasjes pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/avro.git
commit 371e22468baa045b5eb5f1e05ee716d3996f520e Author: Oscar WvH-K <[email protected]> AuthorDate: Fri Jun 21 13:03:55 2019 +0200 AVRO-2440: Decode messages by their write schema This implements AVRO-2440 by allowing a null value for the readSchema field in the BinaryMessageDecoder. It is interpreted as "no preference", and makes the decoder use the write schema of an incoming buffer as its read schema. --- .../apache/avro/message/BinaryMessageDecoder.java | 27 ++++++++++++++-------- .../avro/message/TestBinaryMessageEncoding.java | 15 ++++++++++++ 2 files changed, 33 insertions(+), 9 deletions(-) diff --git a/lang/java/avro/src/main/java/org/apache/avro/message/BinaryMessageDecoder.java b/lang/java/avro/src/main/java/org/apache/avro/message/BinaryMessageDecoder.java index 362a8e8..61d8ef7 100644 --- a/lang/java/avro/src/main/java/org/apache/avro/message/BinaryMessageDecoder.java +++ b/lang/java/avro/src/main/java/org/apache/avro/message/BinaryMessageDecoder.java @@ -16,12 +16,12 @@ * specific language governing permissions and limitations * under the License. */ - package org.apache.avro.message; import org.apache.avro.Schema; import org.apache.avro.SchemaNormalization; import org.apache.avro.generic.GenericData; + import java.io.IOException; import java.io.InputStream; import java.nio.ByteBuffer; @@ -66,12 +66,15 @@ public class BinaryMessageDecoder<D> extends MessageDecoder.BaseDecoder<D> { * {@link Schema schema}. * <p> * The {@code readSchema} is as used the expected schema (read schema). Datum - * instances created by this class will are described by the expected schema. + * instances created by this class will be described by the expected schema. + * <p> + * If {@code readSchema} is {@code null}, the write schema of an incoming buffer + * is used as read schema for that datum instance. * <p> * The schema used to decode incoming buffers is determined by the schema * fingerprint encoded in the message header. This class can decode messages - * that were encoded using the {@code readSchema} and other schemas that are - * added using {@link #addSchema(Schema)}. + * that were encoded using the {@code readSchema} (if any) and other schemas + * that are added using {@link #addSchema(Schema)}. * * @param model the {@link GenericData data model} for datum instances * @param readSchema the {@link Schema} used to construct datum instances @@ -86,12 +89,15 @@ public class BinaryMessageDecoder<D> extends MessageDecoder.BaseDecoder<D> { * {@link Schema schema}. * <p> * The {@code readSchema} is used as the expected schema (read schema). Datum - * instances created by this class will are described by the expected schema. + * instances created by this class will be described by the expected schema. + * <p> + * If {@code readSchema} is {@code null}, the write schema of an incoming buffer + * is used as read schema for that datum instance. * <p> * The schema used to decode incoming buffers is determined by the schema * fingerprint encoded in the message header. This class can decode messages - * that were encoded using the {@code readSchema}, other schemas that are added - * using {@link #addSchema(Schema)}, or schemas returned by the + * that were encoded using the {@code readSchema} (if any), other schemas that + * are added using {@link #addSchema(Schema)}, or schemas returned by the * {@code resolver}. * * @param model the {@link GenericData data model} for datum instances @@ -102,7 +108,9 @@ public class BinaryMessageDecoder<D> extends MessageDecoder.BaseDecoder<D> { this.model = model; this.readSchema = readSchema; this.resolver = resolver; - addSchema(readSchema); + if (readSchema != null) { + addSchema(readSchema); + } } /** @@ -112,7 +120,8 @@ public class BinaryMessageDecoder<D> extends MessageDecoder.BaseDecoder<D> { */ public void addSchema(Schema writeSchema) { long fp = SchemaNormalization.parsingFingerprint64(writeSchema); - codecByFingerprint.put(fp, new RawMessageDecoder<>(model, writeSchema, readSchema)); + final Schema actualReadSchema = this.readSchema != null ? this.readSchema : writeSchema; + codecByFingerprint.put(fp, new RawMessageDecoder<D>(model, writeSchema, actualReadSchema)); } private RawMessageDecoder<D> getDecoder(long fp) { diff --git a/lang/java/avro/src/test/java/org/apache/avro/message/TestBinaryMessageEncoding.java b/lang/java/avro/src/test/java/org/apache/avro/message/TestBinaryMessageEncoding.java index c147d45..ba85d3b 100644 --- a/lang/java/avro/src/test/java/org/apache/avro/message/TestBinaryMessageEncoding.java +++ b/lang/java/avro/src/test/java/org/apache/avro/message/TestBinaryMessageEncoding.java @@ -139,6 +139,21 @@ public class TestBinaryMessageEncoding { } @Test + public void testIdenticalReadWithSchemaFromLookup() throws Exception { + MessageEncoder<Record> v1Encoder = new BinaryMessageEncoder<>(GenericData.get(), SCHEMA_V1); + + SchemaStore.Cache schemaCache = new SchemaStore.Cache(); + schemaCache.addSchema(SCHEMA_V1); + BinaryMessageDecoder<Record> genericDecoder = new BinaryMessageDecoder<>(GenericData.get(), null, schemaCache); + + ByteBuffer v1Buffer = v1Encoder.encode(V1_RECORDS.get(2)); + + Record record = genericDecoder.decode(v1Buffer); + + Assert.assertEquals(V1_RECORDS.get(2), record); + } + + @Test public void testBufferReuse() throws Exception { // This test depends on the serialized version of record 1 being smaller or // the same size as record 0 so that the reused ByteArrayOutputStream won't
