This is an automated email from the ASF dual-hosted git repository.
nielsbasjes pushed a commit to branch branch-1.9
in repository https://gitbox.apache.org/repos/asf/avro.git
The following commit(s) were added to refs/heads/branch-1.9 by this push:
new c3e7936 AVRO-2440: Decode messages by their write schema
c3e7936 is described below
commit c3e79364b905c5414e65118ad218850ec76f204c
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.
(cherry picked from commit 371e22468baa045b5eb5f1e05ee716d3996f520e)
---
.../apache/avro/message/BinaryMessageDecoder.java | 27 ++++++++++++++--------
.../avro/message/TestBinaryMessageEncoding.java | 17 ++++++++++++++
2 files changed, 35 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 3a78424..5ef3e3c 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 f8c2082..f5608db 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,23 @@ 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);
+ // The null readSchema should not throw an NPE, but trigger the
+ // BinaryMessageEncoder to use the write schema as read schema
+ 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
