title: Why this flink scala code is failed to overloaded method value map with alternatives?
body: <p>I tried this code in flink scala repl, and i build a RichMapFunction in java, but scala cannot work.
  anyone could give me about scala method overload strategy document link?</p>
  
  <h2>java rich map function</h2>
  
  <pre><code>public class RichMapFunctions {
  
      public static class CountRichMapFunction&lt;E, T&gt; extends RichMapFunction&lt;E, T&gt; {
  
          private static final long serialVersionUID = 1L;
  
          private transient Counter counter;
  
          @Override
          public void open(Configuration parameters) throws Exception {
              this.counter = getRuntimeContext().getMetricGroup().counter("demo");
          }
  
          @Override
          public T map(E value) throws Exception {
              this.counter.inc();
              return (T) value;
          }
  
      }
  
  }
  </code></pre>
  
  <h2>java FlinkIMain extend IMain imports</h2>
  
  <pre class="lang-java prettyprint-override"><code>    private static String[] PACKAGE_IMPORTS = new String[] { //
          "org.apache.flink.core.fs._", //
          "org.apache.flink.core.fs.local._", //
          "org.apache.flink.api.common.io._", //
          "org.apache.flink.api.common.aggregators._", //
          "org.apache.flink.api.common.accumulators._", //
          "org.apache.flink.api.common.distributions._", //
          "org.apache.flink.api.common.operators._", //
          "org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint", //
          "org.apache.flink.api.common.functions._", //
          "org.apache.flink.api.java.io._", //
          "org.apache.flink.api.java.aggregation._", //
          "org.apache.flink.api.java.functions._", //
          "org.apache.flink.api.java.operators._", //
          "org.apache.flink.api.java.sampling._", //
          "org.apache.flink.api.scala._", //
          "org.apache.flink.api.scala.utils._", //
          "org.apache.flink.streaming.api.scala._", //
          "org.apache.flink.streaming.api.windowing.time._", //
          "com.gaff.sappo.dag.streaming.interpreter.RichMapFunctions.CountRichMapFunction" //
      };
  </code></pre>
  
  <h2>scala script for repl run</h2>
  
  <pre class="lang-scala prettyprint-override"><code>val text = env.socketTextStream("localhost", 9999)
  
  val counts = text.flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } }
    .map { (_, 1) }
    .map(new CountRichMapFunction()) // this code is error
    .keyBy(0)
  val window = counts.timeWindow(Time.seconds(5)).sum(1)
  
  window.print()
  env.execute("Window Stream WordCount")
  </code></pre>
  
  <h2>the error json message</h2>
  
  <pre><code>{"data":"&lt;console&gt;:69: error: overloaded method value map with alternatives:"}
  {"data":"  [R](mapper: org.apache.flink.api.common.functions.MapFunction[(String, Int),R])(implicit evidence$9: org.apache.flink.api.common.typeinfo.TypeInformation[R])org.apache.flink.streaming.api.scala.DataStream[R] &lt;and&gt;"}
  {"data":"  [R](fun: ((String, Int)) =&gt; R)(implicit evidence$8: org.apache.flink.api.common.typeinfo.TypeInformation[R])org.apache.flink.streaming.api.scala.DataStream[R]"}
  {"data":" cannot be applied to (com.gaff.sappo.dag.streaming.interpreter.RichMapFunctions.CountRichMapFunction[Nothing,Nothing])"}
  {"data":"         .map(new CountRichMapFunction())"}
  {"data":"          ^"}
  </code></pre>