[BEAM-359] Treat erased type variables as non-deterministic in AvroCoder (#531)

Author: ehlyzov

sdk/src/main/java/com/google/cloud/dataflow/sdk/coders/AvroCoder.java

@@ -474,6 +474,10 @@ private void doCheck(String context, TypeDescriptor<?> type, Schema schema) {
           checkMap(context, type, schema);
           break;
         case RECORD:
+          if (!(type.getType() instanceof Class)) {
+            reportError(context, "Cannot determine type from generic %s due to erasure", type);
+            return;
+          }
           checkRecord(type, schema);
           break;
         case UNION:
@@ -694,7 +698,8 @@ private void checkArray(String context, TypeDescriptor<?> type, Schema schema) {
      * Extract a field from a class. We need to look at the declared fields so that we can
      * see private fields. We may need to walk up to the parent to get classes from the parent.
      */
-    private static Field getField(Class<?> clazz, String name) {
+    private static Field getField(Class<?> originalClazz, String name) {
+      Class<?> clazz = originalClazz;
       while (clazz != null) {
         for (Field field : clazz.getDeclaredFields()) {
           AvroName avroName = field.getAnnotation(AvroName.class);
@@ -708,7 +713,7 @@ private static Field getField(Class<?> clazz, String name) {
       }
 
       throw new IllegalArgumentException(
-          "Unable to get field " + name + " from class " + clazz);
+          "Unable to get field " + name + " from class " + originalClazz);
     }
   }
 }

sdk/src/test/java/com/google/cloud/dataflow/sdk/coders/AvroCoderTest.java

@@ -751,4 +751,30 @@ public int hashCode() {
       return Objects.hash(getClass(), onlySomeTypesAllowed);
     }
   }
+
+  @Test
+  public void testAvroCoderForGenerics() throws Exception {
+    Schema fooSchema = AvroCoder.of(Foo.class).getSchema();
+    Schema schema = new Schema.Parser().parse("{"
+            + "\"type\":\"record\","
+            + "\"name\":\"SomeGeneric\","
+            + "\"namespace\":\"ns\","
+            + "\"fields\":["
+            + "  {\"name\":\"foo\", \"type\":" + fooSchema.toString() + "}"
+            + "]}");
+    @SuppressWarnings("rawtypes")
+    AvroCoder<SomeGeneric> coder = AvroCoder.of(SomeGeneric.class, schema);
+
+    assertNonDeterministic(coder,
+            reasonField(SomeGeneric.class, "foo", "erasure"));
+  }
+
+  private static class SomeGeneric<T> {
+    @SuppressWarnings("unused")
+    private T foo;
+  }
+  private static class Foo {
+    @SuppressWarnings("unused")
+    String id;
+  }
 }