[jira] [Commented] (GROOVY-9320) Make lambda expression and method reference Serializable

["Daniel Sun (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/GROOVY-9320?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16983443#comment-16983443 ] 

Daniel Sun commented on GROOVY-9320:
------------------------------------

Here is the ideal code. Apart from the serializable issue, static groovy can not infer types properly...
{code:java}
@CompileStatic
class WordCount {
    public static final long TIME_WINDOW_SECONDS = (long) (WordProvider.PROVIDE_WORD_PERIOD / 1000) * 3

    static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment()
        DataStream<String> text = env.socketTextStream("localhost", WordProvider.PORT, "\n")
        DataStream<Tuple2<String, Integer>> windowCounts = text.flatMap(
                (String value, Collector<Tuple2<String, Integer>> out) -> {
                        value.split("\\s").each { String word ->
                            out.collect(Tuple2.of(word, 1))
                        }
                })
                .keyBy(0)
                .timeWindow(Time.seconds(TIME_WINDOW_SECONDS))
                .reduce((t1, t2) -> new Tuple2<>(t1.f0, t1.f1 + t2.f1))

        DataStreamSink<Tuple2<String, Integer>> print = windowCounts.print()
        print.setParallelism(1)

        env.execute "WordCount"
    }
}
{code}

> Make lambda expression and method reference Serializable
> --------------------------------------------------------
>
>                 Key: GROOVY-9320
>                 URL: https://issues.apache.org/jira/browse/GROOVY-9320
>             Project: Groovy
>          Issue Type: Improvement
>            Reporter: Daniel Sun
>            Priority: Major
>
> Here is a word count example for Flink.
> As groovy's lambda expression is not serializable, the following code has to be replaced with {{sum(1)}} as workaround, or {{java.io.NotSerializableException: Non-serializable lambda}} will be raised.
> https://github.com/danielsun1106/flink-wordcount/blob/0.1/src/main/groovy/me/sunlan/flinklabs/wordcount/WordCount.groovy#L49
> We need generate a synthetic method {{$deserializeLambda$}} in bytecode like the following:
> {code:java}
>   private static synthetic $deserializeLambda$(Ljava/lang/invoke/SerializedLambda;)Ljava/lang/Object;
>    L0
>     LINENUMBER 3 L0
>     ALOAD 0
>     INVOKEVIRTUAL java/lang/invoke/SerializedLambda.getImplMethodName ()Ljava/lang/String;
>     ASTORE 1
>     ICONST_M1
>     ISTORE 2
>     ALOAD 1
>     INVOKEVIRTUAL java/lang/String.hashCode ()I
>     LOOKUPSWITCH
>       -50212388: L1
>       default: L2
>    L1
>    FRAME APPEND [java/lang/String I]
>     ALOAD 1
>     LDC "lambda$main$9f099bdf$1"
>     INVOKEVIRTUAL java/lang/String.equals (Ljava/lang/Object;)Z
>     IFEQ L2
>     ICONST_0
>     ISTORE 2
>    L2
>    FRAME SAME
>     ILOAD 2
>     LOOKUPSWITCH
>       0: L3
>       default: L4
>    L3
>    FRAME SAME
>     ALOAD 0
>     INVOKEVIRTUAL java/lang/invoke/SerializedLambda.getImplMethodKind ()I
>     BIPUSH 6
>     IF_ICMPNE L4
>     ALOAD 0
>     INVOKEVIRTUAL java/lang/invoke/SerializedLambda.getFunctionalInterfaceClass ()Ljava/lang/String;
>     LDC "java/lang/Runnable"
>     INVOKEVIRTUAL java/lang/Object.equals (Ljava/lang/Object;)Z
>     IFEQ L4
>     ALOAD 0
>     INVOKEVIRTUAL java/lang/invoke/SerializedLambda.getFunctionalInterfaceMethodName ()Ljava/lang/String;
>     LDC "run"
>     INVOKEVIRTUAL java/lang/Object.equals (Ljava/lang/Object;)Z
>     IFEQ L4
>     ALOAD 0
>     INVOKEVIRTUAL java/lang/invoke/SerializedLambda.getFunctionalInterfaceMethodSignature ()Ljava/lang/String;
>     LDC "()V"
>     INVOKEVIRTUAL java/lang/Object.equals (Ljava/lang/Object;)Z
>     IFEQ L4
>     ALOAD 0
>     INVOKEVIRTUAL java/lang/invoke/SerializedLambda.getImplClass ()Ljava/lang/String;
>     LDC "Test"
>     INVOKEVIRTUAL java/lang/Object.equals (Ljava/lang/Object;)Z
>     IFEQ L4
>     ALOAD 0
>     INVOKEVIRTUAL java/lang/invoke/SerializedLambda.getImplMethodSignature ()Ljava/lang/String;
>     LDC "()V"
>     INVOKEVIRTUAL java/lang/Object.equals (Ljava/lang/Object;)Z
>     IFEQ L4
>     INVOKEDYNAMIC run()Ljava/lang/Runnable; [
>       // handle kind 0x6 : INVOKESTATIC
>       java/lang/invoke/LambdaMetafactory.altMetafactory(Ljava/lang/invoke/MethodHandles$Lookup;Ljava/lang/String;Ljava/lang/invoke/MethodType;[Ljava/lang/Object;)Ljava/lang/invoke/CallSite;
>       // arguments:
>       ()V, 
>       // handle kind 0x6 : INVOKESTATIC
>       Test.lambda$main$9f099bdf$1()V, 
>       ()V, 
>       5, 
>       0
>     ]
>     ARETURN
>    L4
>    FRAME CHOP 2
>     NEW java/lang/IllegalArgumentException
>     DUP
>     LDC "Invalid lambda deserialization"
>     INVOKESPECIAL java/lang/IllegalArgumentException.<init> (Ljava/lang/String;)V
>     ATHROW
>     MAXSTACK = 3
>     MAXLOCALS = 3
> {code}



--
This message was sent by Atlassian Jira
(v8.3.4#803005)