Make TypeMapper trait serializable (#82)

* Make TypeMapper trait serializable

Author: buinauskas

src/main/scala/org/apache/flinkx/api/serializer/MappedSerializer.scala

@@ -37,7 +37,7 @@ case class MappedSerializer[A, B](mapper: TypeMapper[A, B], ser: TypeSerializer[
 }
 
 object MappedSerializer {
-  trait TypeMapper[A, B] {
+  trait TypeMapper[A, B] extends Serializable {
     def map(a: A): B
     def contramap(b: B): A
   }

src/test/scala/org/apache/flinkx/api/MappedTypeInfoTest.scala

@@ -11,9 +11,17 @@ import scala.reflect.ClassTag
 
 class MappedTypeInfoTest extends AnyFlatSpec with Matchers with TestUtils {
   import MappedTypeInfoTest._
-  it should "derive TI for non-serializeable classes" in {
+  it should "derive TI for non-serializable classes" in {
     drop(implicitly[TypeInformation[WrappedString]])
   }
+
+  it should "serialize type mappers across tasks" in {
+    val env         = StreamExecutionEnvironment.getExecutionEnvironment
+    val dataStream  = env.fromElements(Purchase(1, 1.0))
+    val purchase    = dataStream.keyBy(_.id).map(_.copy(price = 5.1)).executeAndCollect(1)
+
+    purchase.last should be (Purchase(1, 5.1))
+  }
 }
 
 object MappedTypeInfoTest {
@@ -40,4 +48,10 @@ object MappedTypeInfoTest {
       internal = value
     }
   }
+
+  case class Purchase(id: Int, price: BigDecimal)
+
+  object Purchase {
+    implicit val typeInfo: TypeInformation[Purchase] = deriveTypeInformation
+  }
 }