Merge pull request #61 from RADAR-base/release-0.11.0

Release 0.11.0

README.md

@@ -14,7 +14,7 @@ repositories {
 }
 
 dependencies {
- implementation group: 'org.radarcns', name: 'radar-commons', version: '0.10.1'
+ implementation group: 'org.radarcns', name: 'radar-commons', version: '0.11.0'
 }
 ```
 
@@ -26,7 +26,7 @@ repositories {
 }
 
 dependencies {
- implementation group: 'org.radarcns', name: 'radar-commons-server', version: '0.10.1'
+ implementation group: 'org.radarcns', name: 'radar-commons-server', version: '0.11.0'
 }
 ```
 
@@ -40,7 +40,7 @@ repositories {
 }
 
 dependencies {
- testImplementation group: 'org.radarcns', name: 'radar-commons-testing', version: '0.10.1'
+ testImplementation group: 'org.radarcns', name: 'radar-commons-testing', version: '0.11.0'
 }
 ```
 
@@ -53,7 +53,7 @@ repositories {
 }
 
 dependencies {
- runtimeOnly group: 'org.radarcns', name: 'radar-commons-unsafe', version: '0.10.1'
+ runtimeOnly group: 'org.radarcns', name: 'radar-commons-unsafe', version: '0.11.0'
 }
 ```
 
@@ -78,7 +78,7 @@ configurations.all {
 }
 
 dependencies {
- compile group: 'org.radarcns', name: 'radar-commons', version: '0.10.2-SNAPSHOT', changing: true
+ compile group: 'org.radarcns', name: 'radar-commons', version: '0.11.1-SNAPSHOT', changing: true
 }
 ```
 

build.gradle

@@ -36,7 +36,7 @@ subprojects {
  // Configuration //
  //---------------------------------------------------------------------------//
 
- version = '0.10.1'
+ version = '0.11.0'
  group = 'org.radarcns'
  ext.githubRepoName = 'RADAR-CNS/RADAR-Commons'
 
@@ -179,5 +179,5 @@ subprojects {
 }
 
 wrapper {
- gradleVersion '4.9'
+ gradleVersion '4.10.2'
 }

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-4.9-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-4.10.2-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists

radar-commons/src/main/java/org/radarcns/data/SpecificRecordDecoder.java → radar-commons/src/main/java/org/radarcns/data/AvroDatumDecoder.java

@@ -19,29 +19,36 @@
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
 import org.apache.avro.io.BinaryDecoder;
 import org.apache.avro.io.DatumReader;
 import org.apache.avro.io.Decoder;
 import org.apache.avro.io.DecoderFactory;
-import org.apache.avro.specific.SpecificDatumReader;
-import org.apache.avro.specific.SpecificRecord;
 
 /** An AvroDecoder to decode known SpecificRecord classes. */
-public class SpecificRecordDecoder implements AvroDecoder {
+public class AvroDatumDecoder implements AvroDecoder {
  private final DecoderFactory decoderFactory;
  private final boolean binary;
+ private final GenericData genericData;
 
- public SpecificRecordDecoder(boolean binary) {
+ /**
+ * Decoder for Avro data.
+ * @param genericData instance of GenericData or SpecificData that should implement
+ * {@link GenericData#createDatumReader(Schema)}.
+ * @param binary true if the read data has Avro binary encoding, false if it has Avro JSON
+ * encoding.
+ */
+ public AvroDatumDecoder(GenericData genericData, boolean binary) {
+ this.genericData = genericData;
  this.decoderFactory = DecoderFactory.get();
  this.binary = binary;
  }
 
  @Override
  public <T> AvroReader<T> reader(Schema schema, Class<? extends T> clazz) {
- if (!SpecificRecord.class.isAssignableFrom(clazz)) {
- throw new IllegalArgumentException("Can only create readers for SpecificRecords.");
- }
- return new AvroRecordReader<>(schema, new SpecificDatumReader<T>(schema));
+ @SuppressWarnings("unchecked")
+ DatumReader<T> reader = genericData.createDatumReader(schema);
+ return new AvroRecordReader<>(schema, reader);
  }
 
  private class AvroRecordReader<T> implements AvroReader<T> {

radar-commons/src/main/java/org/radarcns/data/SpecificRecordEncoder.java → radar-commons/src/main/java/org/radarcns/data/AvroDatumEncoder.java

@@ -18,30 +18,30 @@
 
 import java.io.IOException;
 import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.io.DatumWriter;
 import org.apache.avro.io.EncoderFactory;
-import org.apache.avro.specific.SpecificDatumWriter;
-import org.apache.avro.specific.SpecificRecord;
 
 /** An AvroEncoder to encode known SpecificRecord classes. */
-public class SpecificRecordEncoder implements AvroEncoder {
+public class AvroDatumEncoder implements AvroEncoder {
  private final EncoderFactory encoderFactory;
  private final boolean binary;
+ private final GenericData genericData;
 
  /**
  * Create a SpecificRecordEncoder.
  * @param binary whether to use binary encoding or JSON.
  */
- public SpecificRecordEncoder(boolean binary) {
+ public AvroDatumEncoder(GenericData genericData, boolean binary) {
+ this.genericData = genericData;
  this.encoderFactory = EncoderFactory.get();
  this.binary = binary;
  }
 
  @Override
  public <T> AvroWriter<T> writer(Schema schema, Class<? extends T> clazz) throws IOException {
- if (!SpecificRecord.class.isAssignableFrom(clazz)) {
- throw new IllegalArgumentException("Can only create writers for SpecificRecords.");
- }
- return new AvroRecordWriter<>(encoderFactory, schema, new SpecificDatumWriter<T>(schema),
- binary);
+ @SuppressWarnings("unchecked")
+ DatumWriter<T> writer = (DatumWriter<T>)genericData.createDatumWriter(schema);
+ return new AvroRecordWriter<>(encoderFactory, schema, writer, binary);
  }
 }

radar-commons/src/main/java/org/radarcns/data/AvroEncoder.java

@@ -18,14 +18,36 @@
 
 import java.io.IOException;
 import org.apache.avro.Schema;
+import org.apache.avro.SchemaValidationException;
+import org.radarcns.producer.rest.ParsedSchemaMetadata;
 
-/** Encode Avro values with a given encoder. */
+/** Encode Avro values with a given encoder. The encoder may take into account the schema
+ * that the schema registry has listed for a given topic. */
 public interface AvroEncoder {
  /** Create a new writer. This method is thread-safe, but the class it returns is not. */
  <T> AvroWriter<T> writer(Schema schema, Class<? extends T> clazz) throws IOException;
 
  interface AvroWriter<T> {
- /** Encode an object. This method is not thread-safe. */
+ /**
+ * Encode an object. This method is not thread-safe. Call
+ * {@link #setReaderSchema(ParsedSchemaMetadata)} before calling encode.
+ * @param object object to encode
+ * @return byte array with serialized object.
+ */
  byte[] encode(T object) throws IOException;
+
+ /**
+ * Update the schema that the server is lists for the current topic.
+ * @param readerSchema schema listed by the schema registry.
+ * @throws SchemaValidationException if the server schema is incompatible with the writer
+ * schema.
+ */
+ void setReaderSchema(ParsedSchemaMetadata readerSchema) throws SchemaValidationException;
+
+ /**
+ * Get the schema that the server lists.
+ * @return schema as set by setReaderSchema or null if not called yet.
+ */
+ ParsedSchemaMetadata getReaderSchema();
  }
 }

radar-commons/src/main/java/org/radarcns/data/AvroRecordData.java

@@ -1,12 +1,8 @@
 package org.radarcns.data;
 
-import java.io.IOException;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Objects;
-import org.apache.avro.Schema;
-import org.apache.avro.generic.GenericRecord;
-import org.apache.avro.specific.SpecificRecord;
 import org.radarcns.topic.AvroTopic;
 
 /**
@@ -53,30 +49,6 @@ public Iterator<V> iterator() {
  return records.iterator();
  }
 
- /**
- * Get an Avro encoder for given settings. This only works for
- * {@link org.apache.avro.generic.IndexedRecord} instances.
- * @param schema schema to encode with.
- * @param cls class type to encode.
- * @param binary true if the converter should yield binary data, false otherwise.
- * @param <T> type of data
- * @return new Avro writer.
- * @throws IOException if the record converter could not be created.
- * @throws IllegalArgumentException if the supplied class is not an IndexedRecord.
- */
- public static <T> AvroEncoder.AvroWriter<T> getEncoder(
- Schema schema, Class<? extends T> cls, boolean binary) throws IOException {
- AvroEncoder encoder;
- if (SpecificRecord.class.isAssignableFrom(cls)) {
- encoder = new SpecificRecordEncoder(binary);
- } else if (GenericRecord.class.isAssignableFrom(cls)) {
- encoder = new GenericRecordEncoder(binary);
- } else {
- throw new IllegalArgumentException("Cannot get encoder for non-avro records");
- }
- return encoder.writer(schema, cls);
- }
-
  @Override
  public boolean isEmpty() {
  return records.isEmpty();

radar-commons/src/main/java/org/radarcns/data/AvroRecordWriter.java

@@ -23,6 +23,7 @@
 import org.apache.avro.io.Encoder;
 import org.apache.avro.io.EncoderFactory;
 import org.radarcns.data.AvroEncoder.AvroWriter;
+import org.radarcns.producer.rest.ParsedSchemaMetadata;
 
 /**
  * Encodes Avro records to bytes.
@@ -31,6 +32,7 @@ public class AvroRecordWriter<T> implements AvroWriter<T> {
  private final Encoder encoder;
  private final ByteArrayOutputStream out;
  private final DatumWriter<T> writer;
+ private ParsedSchemaMetadata serverSchema;
 
  /**
  * Writer for a given encoder, schema and writer.
@@ -62,4 +64,14 @@ public byte[] encode(T record) throws IOException {
  out.reset();
  }
  }
+
+ @Override
+ public void setReaderSchema(ParsedSchemaMetadata readerSchema) {
+ this.serverSchema = readerSchema;
+ }
+
+ @Override
+ public ParsedSchemaMetadata getReaderSchema() {
+ return serverSchema;
+ }
 }