beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From echauc...@apache.org
Subject [beam] 14/37: Fix beam coder lazy init using reflexion: use .clas + try catch + cast
Date Thu, 24 Oct 2019 10:15:23 GMT
This is an automated email from the ASF dual-hosted git repository.

echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git

commit 0cf2c8759a64c81c1d4f83f74a759ae3dafd1f83
Author: Etienne Chauchot <echauchot@apache.org>
AuthorDate: Thu Sep 5 10:07:32 2019 +0200

    Fix beam coder lazy init using reflexion: use .clas + try catch + cast
---
 .../translation/helpers/EncoderHelpers.java           | 19 ++++++++++++-------
 1 file changed, 12 insertions(+), 7 deletions(-)

diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
index a452da0..0751c4c 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/helpers/EncoderHelpers.java
@@ -26,6 +26,7 @@ import java.util.List;
 import java.util.Objects;
 import org.apache.beam.runners.spark.structuredstreaming.translation.SchemaHelpers;
 import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
 import org.apache.beam.sdk.util.WindowedValue;
 import org.apache.beam.sdk.values.KV;
 import org.apache.spark.sql.Encoder;
@@ -388,18 +389,22 @@ case class DecodeUsingSerializer[T](child: Expression, tag: ClassTag[T],
kryo: B
     ctx.addImmutableStateIfNotExists(coderClass.getName(), beamCoderInstance, func(v1 ->
{
       /*
     CODE GENERATED
-    v = (coderClass) coderClass.getDeclaredConstructor().newInstance();
+    try {
+    v1 = coderClass.class.getDeclaredConstructor().newInstance();
+    } catch (Exception e) {
+      throw new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));
+    }
      */
-        List<String> parts = new ArrayList<>();
-        parts.add("");
+      List<String> parts = new ArrayList<>();
+        parts.add("try {");
         parts.add(" = (");
-        parts.add(") ");
-        parts.add(".getDeclaredConstructor().newInstance();");
+      parts.add(") ");
+      parts.add(".class.getDeclaredConstructor().newInstance();} catch (Exception e) {throw
new RuntimeException(org.apache.beam.sdk.util.UserCodeException.wrap(e));}");
         StringContext sc = new StringContext(JavaConversions.collectionAsScalaIterable(parts).toSeq());
         List<Object> args = new ArrayList<>();
         args.add(v1);
-        args.add(coderClass.getName());
-        args.add(coderClass.getName());
+      args.add(coderClass.getName());
+      args.add(coderClass.getName());
         return sc.s(JavaConversions.collectionAsScalaIterable(args).toSeq());
       }));
     return beamCoderInstance;


Mime
View raw message