App.scala

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

import scala.beans.BeanProperty

object App {
  case class Test(
                   @BeanProperty var a: String,
                   @BeanProperty var b: Int,
                   @BeanProperty var c: String,
                   @BeanProperty var d: Long) {
    def this() = {
      this(null, 0, null, 0)
    }
  }

  class MapFunc extends RichMapFunction[String, /*Test*/Any] {
    override def map(value: String): /*Test*/Any = {
      Test("aaa", 1, "ccc", 2L)
    }
  }

  def main(args: Array[String]): Unit = {
    implicit val typeInfo = TypeInformation.of(classOf[/*Test*/Any])

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val stream = env.socketTextStream("localhost", 9999)
    println(stream.map(new MapFunc).keyBy("a", "c"))
    //org.apache.flink.api.common.InvalidProgramException: This type (GenericType<java.lang.Object>) cannot be used as key.
  }
}