Scala：java.lang.ClassCastException：无法将java.lang.invoke.SerializedLambda的实例分配给scala.Function1类型的字段Child1.myfun

如何解决Scala：java.lang.ClassCastException：无法将java.lang.invoke.SerializedLambda的实例分配给scala.Function1类型的字段Child1.myfun

我想将函数名称作为父类中的参数，以便子类可以设置它。此变量将在父类的方法之一中使用。

abstract class Parent[T: TypeInformation] {
   val myfun: T => Unit

   // A different method uses myfun
}

class Child1 extends Parent[User] {
   val service = new Service()

   val myfun: User => Unit = service.callme
}

class Service {
   def callme(user: User) => Unit = {
      println("We are here for user")
   }
}

我是scala的新手，但这看起来还不错。虽然编译器没有抱怨，但我遇到了运行时异常，并且该作业无法启动：

org.apache.flink.streaming.runtime.tasks.StreamTaskException: Cannot instantiate user function.
    at org.apache.flink.streaming.api.graph.StreamConfig.getStreamOperatorFactory(StreamConfig.java:250)
    at org.apache.flink.streaming.runtime.tasks.OperatorChain.createChainedOperator(OperatorChain.java:427)
    at org.apache.flink.streaming.runtime.tasks.OperatorChain.createOutputCollector(OperatorChain.java:354)
    at org.apache.flink.streaming.runtime.tasks.OperatorChain.createChainedOperator(OperatorChain.java:418)
    at org.apache.flink.streaming.runtime.tasks.OperatorChain.createOutputCollector(OperatorChain.java:354)
    at org.apache.flink.streaming.runtime.tasks.OperatorChain.createChainedOperator(OperatorChain.java:418)
    at org.apache.flink.streaming.runtime.tasks.OperatorChain.createOutputCollector(OperatorChain.java:354)
    at org.apache.flink.streaming.runtime.tasks.OperatorChain.<init>(OperatorChain.java:144)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:393)
    at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:705)
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:530)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.ClassCastException: cannot assign instance of java.lang.invoke.SerializedLambda to field Child1.myfun of type scala.Function1 in instance of Child1
    at java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(ObjectStreamClass.java:2301)
    at java.io.ObjectStreamClass.setObjFieldValues(ObjectStreamClass.java:1431)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2372)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2290)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2148)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1647)
    at java.io.ObjectInputStream.readArray(ObjectInputStream.java:2054)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1635)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2366)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2290)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2148)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1647)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2366)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2290)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2148)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1647)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2366)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2290)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2148)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1647)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2366)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2290)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2148)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1647)
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:483)
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:441)
    at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:576)
    at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:562)
    at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:550)
    at org.apache.flink.util.InstantiationUtil.readObjectFromConfig(InstantiationUtil.java:511)
    at org.apache.flink.streaming.api.graph.StreamConfig.getStreamOperatorFactory(StreamConfig.java:235)
    ... 11 more

有什么主意，我该如何在子级中将函数名设置为变量值，以便父级可以使用它？

解决方法

您正在谈论传递函数名称，但是传递的是实际函数本身。似乎无法通过flink进行序列化。

一种替代编码可能根本不使用函数值，而是一种普通的旧方法：

abstract class Parent[T: TypeInformation] {
   def myfun(t: T): Unit

   // A different method uses myfun
}

class Child1 extends Parent[User] {
   val service = new Service()

   def myfun(t: User): Unit = service.callme(t)
}

class Service {
   def callme(user: User) => Unit = {
      println("We are here for user")
   }
}

如果无法立即使用，它至少应该为您提供更清晰的错误消息（可能与类路径上Service的可用性有关。

修复该问题之后，您可能可以返回到以前的编码方式