Fix BEAM-359

Rather than throwing an exception when checking determinism of an erased
generic, record that as a non-determistic class and proceed.

sdks/java/core/src/main/java/org/apache/beam/sdk/coders/AvroCoder.java

@@ -475,6 +475,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:
@@ -695,7 +699,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,8 +713,7 @@ private static Field getField(Class<?> clazz, String name) {
         clazz = clazz.getSuperclass();
       }
 
-      throw new IllegalArgumentException(
-          "Unable to get field " + name + " from class " + clazz);
+      throw new IllegalArgumentException("Unable to get field " + name + " from " + originalClazz);
     }
   }
 }

sdks/java/core/src/main/java/org/apache/beam/sdk/values/TypeDescriptor.java

@@ -17,6 +17,8 @@
  */
 package org.apache.beam.sdk.values;
 
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+
 import com.google.common.collect.Lists;
 import com.google.common.reflect.Invokable;
 import com.google.common.reflect.Parameter;

sdks/java/core/src/test/java/org/apache/beam/sdk/coders/AvroCoderTest.java

@@ -755,4 +755,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;
+  }
 }