AvroCoder extends CustomCoder<T> { public static <T> AvroCoder<T> of(TypeDescriptor<T> type) { @SuppressWarnings("unchecked") Class<T> clazz = (Class<T>) type.getRawType(); return of(clazz); } protected  AvroCoder(Class<T> type, Schema schema); static AvroCoder<T> of(TypeDescriptor<T> type); static AvroCoder<T> of(Class<T> clazz); static AvroGenericCoder of(Schema schema); static AvroCoder<T> of(Class<T> type, Schema schema); @SuppressWarnings("unused") static CoderProvider getCoderProvider(); Class<T> getType(); @Override void encode(T value, OutputStream outStream); @Override T decode(InputStream inStream); @Override void verifyDeterministic(); Schema getSchema(); @Override TypeDescriptor<T> getEncodedTypeDescriptor(); @Override boolean equals(@Nullable Object other); @Override int hashCode();  }
@Test public void testAvroCoderEncoding() throws Exception { AvroCoder<Pojo> coder = AvroCoder.of(Pojo.class); CoderProperties.coderSerializable(coder); AvroCoder<Pojo> copy = SerializableUtils.clone(coder); Pojo pojo = new Pojo("foo", 3, DATETIME_A); Pojo equalPojo = new Pojo("foo", 3, DATETIME_A); Pojo otherPojo = new Pojo("bar", -19, DATETIME_B); CoderProperties.coderConsistentWithEquals(coder, pojo, equalPojo); CoderProperties.coderConsistentWithEquals(copy, pojo, equalPojo); CoderProperties.coderConsistentWithEquals(coder, pojo, otherPojo); CoderProperties.coderConsistentWithEquals(copy, pojo, otherPojo); }
@Test public void testTwoClassLoaders() throws Exception { ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader(); ClassLoader loader1 = new InterceptingUrlClassLoader(contextClassLoader, AvroCoderTestPojo.class.getName()); ClassLoader loader2 = new InterceptingUrlClassLoader(contextClassLoader, AvroCoderTestPojo.class.getName()); Class<?> pojoClass1 = loader1.loadClass(AvroCoderTestPojo.class.getName()); Class<?> pojoClass2 = loader2.loadClass(AvroCoderTestPojo.class.getName()); Object pojo1 = InstanceBuilder.ofType(pojoClass1).withArg(String.class, "hello").build(); Object pojo2 = InstanceBuilder.ofType(pojoClass2).withArg(String.class, "goodbye").build(); try { pojoClass2.cast(pojo1); fail("Expected ClassCastException; without it, this test is vacuous"); } catch (ClassCastException e) { } AvroCoder<Object> avroCoder1 = (AvroCoder) AvroCoder.of(pojoClass1); AvroCoder<Object> avroCoder2 = (AvroCoder) AvroCoder.of(pojoClass2); Object cloned1 = CoderUtils.clone(avroCoder1, pojo1); Object cloned2 = CoderUtils.clone(avroCoder2, pojo2); pojoClass1.cast(cloned1); pojoClass2.cast(cloned2); }
@Test public void testTransientFieldInitialization() throws Exception { Pojo value = new Pojo("Hello", 42, DATETIME_A); AvroCoder<Pojo> coder = AvroCoder.of(Pojo.class); ByteArrayOutputStream bos = new ByteArrayOutputStream(); ObjectOutputStream out = new ObjectOutputStream(bos); out.writeObject(coder); ByteArrayInputStream bis = new ByteArrayInputStream(bos.toByteArray()); ObjectInputStream in = new ObjectInputStream(bis); AvroCoder<Pojo> copied = (AvroCoder<Pojo>) in.readObject(); CoderProperties.coderDecodeEncodeEqual(copied, value); }
@Test public void testKryoSerialization() throws Exception { Pojo value = new Pojo("Hello", 42, DATETIME_A); AvroCoder<Pojo> coder = AvroCoder.of(Pojo.class); Kryo kryo = new Kryo(); kryo.setInstantiatorStrategy(new StdInstantiatorStrategy()); ByteArrayOutputStream coderWithoutMemoizationBos = new ByteArrayOutputStream(); try (Output output = new Output(coderWithoutMemoizationBos)) { kryo.writeObject(output, coder); } CoderProperties.coderDecodeEncodeEqual(coder, value); ByteArrayOutputStream coderWithMemoizationBos = new ByteArrayOutputStream(); try (Output output = new Output(coderWithMemoizationBos)) { kryo.writeObject(output, coder); } ByteArrayInputStream bisWithoutMemoization = new ByteArrayInputStream(coderWithoutMemoizationBos.toByteArray()); AvroCoder<Pojo> copiedWithoutMemoization = (AvroCoder<Pojo>) kryo.readObject(new Input(bisWithoutMemoization), AvroCoder.class); ByteArrayInputStream bisWithMemoization = new ByteArrayInputStream(coderWithMemoizationBos.toByteArray()); AvroCoder<Pojo> copiedWithMemoization = (AvroCoder<Pojo>) kryo.readObject(new Input(bisWithMemoization), AvroCoder.class); CoderProperties.coderDecodeEncodeEqual(copiedWithoutMemoization, value); CoderProperties.coderDecodeEncodeEqual(copiedWithMemoization, value); }
@Test public void testPojoEncoding() throws Exception { Pojo value = new Pojo("Hello", 42, DATETIME_A); AvroCoder<Pojo> coder = AvroCoder.of(Pojo.class); CoderProperties.coderDecodeEncodeEqual(coder, value); }
@Test @Category(NeedsRunner.class) public void testDefaultCoder() throws Exception { PCollection<String> output = pipeline .apply(Create.of(new Pojo("hello", 1, DATETIME_A), new Pojo("world", 2, DATETIME_B))) .apply(ParDo.of(new GetTextFn())); PAssert.that(output).containsInAnyOrder("hello", "world"); pipeline.run(); }
@Test public void testAvroCoderIsSerializable() throws Exception { AvroCoder<Pojo> coder = AvroCoder.of(Pojo.class); SerializableUtils.ensureSerializable(coder); }
@Test public void testDeterministicInteger() { assertDeterministic(AvroCoder.of(Integer.class)); }
@Test public void testDeterministicInt() { assertDeterministic(AvroCoder.of(int.class)); }
@Test public void testDeterministicSimple() { assertDeterministic(AvroCoder.of(SimpleDeterministicClass.class)); }
@Test public void testDeterministicUnorderedMap() { assertNonDeterministic( AvroCoder.of(UnorderedMapClass.class), reasonField( UnorderedMapClass.class, "mapField", "java.util.Map<java.lang.String, java.lang.String> " + "may not be deterministically ordered")); }
@Test public void testDeterministicNonDeterministicArray() { assertNonDeterministic( AvroCoder.of(NonDeterministicArray.class), reasonField( UnorderedMapClass.class, "mapField", "java.util.Map<java.lang.String, java.lang.String>" + " may not be deterministically ordered")); }
@Test public void testDeterministicNonDeterministicChild() { assertNonDeterministic( AvroCoder.of(SubclassOfUnorderedMapClass.class), reasonField(UnorderedMapClass.class, "mapField", "may not be deterministically ordered")); }
@Test public void testDeterministicWithAvroName() { assertDeterministic(AvroCoder.of(FieldWithAvroName.class)); }
@Test public void testDeterminismSortedMap() { assertDeterministic(AvroCoder.of(StringSortedMapField.class)); }
@Test public void testDeterminismTreeMapValue() { assertNonDeterministic( AvroCoder.of(TreeMapNonDetValue.class), reasonField( UnorderedMapClass.class, "mapField", "java.util.Map<java.lang.String, java.lang.String> " + "may not be deterministically ordered")); }
@Test public void testDeterminismUnorderedMap() { assertNonDeterministic( AvroCoder.of(LinkedHashMapField.class), reasonField( LinkedHashMapField.class, "nonDeterministicMap", "java.util.LinkedHashMap<java.lang.String, java.lang.String> " + "may not be deterministically ordered")); }
@Test public void testDeterminismCollection() { assertNonDeterministic( AvroCoder.of(StringCollection.class), reasonField( StringCollection.class, "stringCollection", "java.util.Collection<java.lang.String> may not be deterministically ordered")); }
@Test public void testDeterminismList() { assertDeterministic(AvroCoder.of(StringList.class)); assertDeterministic(AvroCoder.of(StringArrayList.class)); }
@Test public void testDeterminismSet() { assertDeterministic(AvroCoder.of(StringSortedSet.class)); assertDeterministic(AvroCoder.of(StringTreeSet.class)); assertNonDeterministic( AvroCoder.of(StringHashSet.class), reasonField( StringHashSet.class, "stringCollection", "java.util.HashSet<java.lang.String> may not be deterministically ordered")); }
@Test public void testDeterminismCollectionValue() { assertNonDeterministic( AvroCoder.of(OrderedSetOfNonDetValues.class), reasonField(UnorderedMapClass.class, "mapField", "may not be deterministically ordered")); assertNonDeterministic( AvroCoder.of(ListOfNonDetValues.class), reasonField(UnorderedMapClass.class, "mapField", "may not be deterministically ordered")); }
@Test public void testDeterminismUnion() { assertDeterministic(AvroCoder.of(DeterministicUnionBase.class)); assertNonDeterministic( AvroCoder.of(NonDeterministicUnionBase.class), reasonField(UnionCase3.class, "mapField", "may not be deterministically ordered")); }
@Test public void testDeterminismStringable() { assertDeterministic(AvroCoder.of(String.class)); assertNonDeterministic( AvroCoder.of(StringableClass.class), reasonClass(StringableClass.class, "may not have deterministic #toString()")); }
@Test public void testDeterminismCyclicClass() { assertNonDeterministic( AvroCoder.of(Cyclic.class), reasonField(Cyclic.class, "cyclicField", "appears recursively")); assertNonDeterministic( AvroCoder.of(CyclicField.class), reasonField(Cyclic.class, "cyclicField", Cyclic.class.getName() + " appears recursively")); assertNonDeterministic( AvroCoder.of(IndirectCycle1.class), reasonField( IndirectCycle2.class, "field2", IndirectCycle1.class.getName() + " appears recursively")); }
@Test public void testDeterminismHasGenericRecord() { assertDeterministic(AvroCoder.of(HasGenericRecord.class)); }
@Test public void testDeterminismHasCustomSchema() { assertNonDeterministic( AvroCoder.of(HasCustomSchema.class), reasonField( HasCustomSchema.class, "withCustomSchema", "Custom schemas are only supported for subtypes of IndexedRecord.")); }
@Test public void testAvroCoderSimpleSchemaDeterminism() { assertDeterministic(AvroCoder.of(SchemaBuilder.record("someRecord").fields().endRecord())); assertDeterministic( AvroCoder.of( SchemaBuilder.record("someRecord") .fields() .name("int") .type() .intType() .noDefault() .endRecord())); assertDeterministic( AvroCoder.of( SchemaBuilder.record("someRecord") .fields() .name("string") .type() .stringType() .noDefault() .endRecord())); assertNonDeterministic( AvroCoder.of( SchemaBuilder.record("someRecord") .fields() .name("map") .type() .map() .values() .stringType() .noDefault() .endRecord()), reason("someRecord.map", "HashMap to represent MAPs")); assertDeterministic( AvroCoder.of( SchemaBuilder.record("someRecord") .fields() .name("array") .type() .array() .items() .stringType() .noDefault() .endRecord())); assertDeterministic( AvroCoder.of( SchemaBuilder.record("someRecord") .fields() .name("enum") .type() .enumeration("anEnum") .symbols("s1", "s2") .enumDefault("s1") .endRecord())); assertDeterministic( AvroCoder.of( SchemaBuilder.unionOf() .intType() .and() .record("someRecord") .fields() .nullableString("someField", "") .endRecord() .endUnion())); }
@Test public void testAvroCoderStrings() { assertDeterministic( AvroCoder.of( SchemaBuilder.record("someRecord") .fields() .name("string") .prop(SpecificData.CLASS_PROP, "java.lang.String") .type() .stringType() .noDefault() .endRecord())); assertNonDeterministic( AvroCoder.of( SchemaBuilder.record("someRecord") .fields() .name("string") .prop(SpecificData.CLASS_PROP, "unknownString") .type() .stringType() .noDefault() .endRecord()), reason("someRecord.string", "unknownString is not known to be deterministic")); assertNonDeterministic( AvroCoder.of( SchemaBuilder.unionOf() .intType() .and() .record("someRecord") .fields() .name("someField") .prop(SpecificData.CLASS_PROP, "unknownString") .type() .stringType() .noDefault() .endRecord() .endUnion()), reason("someRecord.someField", "unknownString is not known to be deterministic")); }
@Test public void testAvroCoderNestedRecords() { assertDeterministic( AvroCoder.of( SchemaBuilder.record("nestedRecord") .fields() .name("subRecord") .type() .record("subRecord") .fields() .name("innerField") .type() .stringType() .noDefault() .endRecord() .noDefault() .endRecord())); }
@Test public void testAvroCoderCyclicRecords() { assertNonDeterministic( AvroCoder.of( SchemaBuilder.record("cyclicRecord") .fields() .name("cycle") .type("cyclicRecord") .noDefault() .endRecord()), reason("cyclicRecord.cycle", "cyclicRecord appears recursively")); }
@Test public void testNullableField() { assertDeterministic(AvroCoder.of(NullableField.class)); }
@Test public void testNullableNonDeterministicField() { assertNonDeterministic( AvroCoder.of(NullableCyclic.class), reasonField( NullableCyclic.class, "nullableNullableCyclicField", NullableCyclic.class.getName() + " appears recursively")); assertNonDeterministic( AvroCoder.of(NullableCyclicField.class), reasonField(Cyclic.class, "cyclicField", Cyclic.class.getName() + " appears recursively")); assertNonDeterministic( AvroCoder.of(NullableNonDeterministicField.class), reasonField(UnorderedMapClass.class, "mapField", " may not be deterministically ordered")); }