From 22e87daf4c8a5fe520a3808186b9a50aadb7eebb Mon Sep 17 00:00:00 2001
From: Thomas Groh <tgroh@google.com>
Date: Thu, 11 May 2017 17:16:53 -0700
Subject: [PATCH] Enable SerializableCoder to Serialize with Generic Types

A TypeToken that contains generics is not serializable. However, the
TypeDescriptor does not need to be transmitted via the serialized form,
so mark it as transient.
---
 .../src/main/resources/beam/findbugs-filter.xml        | 10 ++++++++++
 .../org/apache/beam/sdk/coders/SerializableCoder.java  |  5 ++++-
 .../apache/beam/sdk/coders/SerializableCoderTest.java  |  8 +++++++-
 3 files changed, 21 insertions(+), 2 deletions(-)

diff --git a/sdks/java/build-tools/src/main/resources/beam/findbugs-filter.xml b/sdks/java/build-tools/src/main/resources/beam/findbugs-filter.xml
index 1db0e86d6afc..8ff0cb02eb3c 100644
--- a/sdks/java/build-tools/src/main/resources/beam/findbugs-filter.xml
+++ b/sdks/java/build-tools/src/main/resources/beam/findbugs-filter.xml
@@ -73,6 +73,16 @@
     -->
   </Match>
 
+  <Match>
+    <Class name="org.apache.beam.sdk.coders.SerializableCoder"/>
+    <Field name="typeDescriptor"/>
+    <Bug pattern="SE_TRANSIENT_FIELD_NOT_RESTORED"/>
+    <!--
+    the field is used only in getEncodedTypeDescriptor, where it is restored if it is not present due to
+    serialization
+    -->
+  </Match>
+
   <Match>
     <Class name="org.apache.beam.sdk.io.jms.JmsRecord"/>
     <Field name="jmsDestination"/>
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/SerializableCoder.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/SerializableCoder.java
index 9aa8493e5e71..66918768818f 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/SerializableCoder.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/SerializableCoder.java
@@ -107,7 +107,7 @@ public <T> Coder<T> coderFor(TypeDescriptor<T> typeDescriptor,
   }
 
   private final Class<T> type;
-  private final TypeDescriptor<T> typeDescriptor;
+  private transient TypeDescriptor<T> typeDescriptor;
 
   protected SerializableCoder(Class<T> type, TypeDescriptor<T> typeDescriptor) {
     this.type = type;
@@ -166,6 +166,9 @@ public int hashCode() {
 
   @Override
   public TypeDescriptor<T> getEncodedTypeDescriptor() {
+    if (typeDescriptor == null) {
+      typeDescriptor = TypeDescriptor.of(type);
+    }
     return typeDescriptor;
   }
 
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/SerializableCoderTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/SerializableCoderTest.java
index adb665286eee..dd4f6ca345cd 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/SerializableCoderTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/SerializableCoderTest.java
@@ -125,11 +125,17 @@ public void testSerializableCoderConstruction() throws Exception {
     assertEquals(coder.getRecordType(), MyRecord.class);
     CoderProperties.coderSerializable(coder);
 
-
     SerializableCoder<?> decoded = SerializableUtils.clone(coder);
     assertThat(decoded.getRecordType(), Matchers.<Object>equalTo(MyRecord.class));
   }
 
+  @Test
+  public <T extends Serializable> void testSerializableCoderIsSerializableWithGenericTypeToken()
+  throws Exception {
+    SerializableCoder<T> coder = SerializableCoder.of(new TypeDescriptor<T>() {});
+    CoderProperties.coderSerializable(coder);
+  }
+
   @Test
   public void testNullEquals() {
     SerializableCoder<MyRecord> coder = SerializableCoder.of(MyRecord.class);