title: Apache Flink - can't compile KeyedOneInputStreamOperatorTestHarness with Int or Long generic parameters
body: <p>The bellow mentioned Apache Flink test code doesn't compile in Scala when I want to use KeyedOneInputStreamOperatorTestHarness[K, IN, OUT] with Int or Long as the key (K) parameter. The same code can be compiled with generic parameters of type STRING, INTEGER or java.lang.Long:</p>
  <pre><code>import org.apache.flink.api.java.functions.KeySelector
  import org.apache.flink.api.scala.typeutils.Types
  import org.apache.flink.streaming.api.functions.KeyedProcessFunction
  import org.apache.flink.streaming.api.operators.KeyedProcessOperator
  import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness
  import org.apache.flink.streaming.api.scala._
  import org.junit.Test
  
  case class TaxiRide(rideId: Int)
  
  class RideKeySelector extends KeySelector[TaxiRide, Int] {
    override def getKey(in: TaxiRide): Int = in.rideId
  }
  
  class RideTests {
    private def setupHarness(function: KeyedProcessFunction[Int, TaxiRide, Int]): KeyedOneInputStreamOperatorTestHarness[Int, TaxiRide, Int] = {
      val operator: KeyedProcessOperator[Int, TaxiRide, Int] = new KeyedProcessOperator(function)
  
      val testHarness: KeyedOneInputStreamOperatorTestHarness[Int, TaxiRide, Int] =
        new KeyedOneInputStreamOperatorTestHarness(operator, new RideKeySelector(), Types.INT)
  
      testHarness.setup()
      testHarness.open()
  
      testHarness
    }
  }
  </code></pre>
  <p><strong>Build error:</strong></p>
  <pre><code>type mismatch;
   found   : org.example.RideKeySelector
   required: org.apache.flink.api.java.functions.KeySelector[org.example.TaxiRide,Int]
        new KeyedOneInputStreamOperatorTestHarness(operator, new RideKeySelector(), Types.INT)
  </code></pre>
  <p>I created the project from the following Maven archetype:</p>
  <pre><code>&lt;dependency&gt;
      &lt;groupId&gt;org.apache.flink&lt;/groupId&gt;
      &lt;artifactId&gt;flink-quickstart-scala&lt;/artifactId&gt;
      &lt;version&gt;1.14.2&lt;/version&gt;
  &lt;/dependency&gt;
  </code></pre>
  <p>I added the following test dependencies to pom.xml in IntelliJ IDEA:</p>
  <pre><code>&lt;dependencies&gt;
      &lt;dependency&gt;
          &lt;groupId&gt;org.apache.flink&lt;/groupId&gt;
          &lt;artifactId&gt;flink-test-utils_2.11&lt;/artifactId&gt;
          &lt;version&gt;1.14.2&lt;/version&gt;    
          &lt;scope&gt;test&lt;/scope&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
          &lt;groupId&gt;org.apache.flink&lt;/groupId&gt;
          &lt;artifactId&gt;flink-runtime&lt;/artifactId&gt;
          &lt;version&gt;1.14.2&lt;/version&gt;    
          &lt;scope&gt;test&lt;/scope&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
          &lt;groupId&gt;org.apache.flink&lt;/groupId&gt;
          &lt;artifactId&gt;flink-streaming-java_2.11&lt;/artifactId&gt;
          &lt;version&gt;1.14.2&lt;/version&gt;    
          &lt;scope&gt;test&lt;/scope&gt;    
          &lt;classifier&gt;tests&lt;/classifier&gt;
      &lt;/dependency&gt;
      &lt;dependency&gt;
          &lt;groupId&gt;org.apache.flink&lt;/groupId&gt;
          &lt;artifactId&gt;flink-streaming-scala_2.11&lt;/artifactId&gt;
          &lt;version&gt;1.14.2&lt;/version&gt;
          &lt;scope&gt;test&lt;/scope&gt;
          &lt;classifier&gt;tests&lt;/classifier&gt;
      &lt;/dependency&gt;
  &lt;/dependencies&gt;
  
  </code></pre>
  <p>I tried to replace the key selector with a lambda:</p>
  <pre><code>val testHarness = new KeyedOneInputStreamOperatorTestHarness(operator, (ride: TaxiRide) =&gt; ride.rideId, Types.INT)
  </code></pre>
  <p>The <strong>build error</strong> changes to:</p>
  <pre><code>overloaded method constructor KeyedOneInputStreamOperatorTestHarness with alternatives:
    (x$1: org.apache.flink.streaming.api.operators.OneInputStreamOperator[IN,OUT],x$2: org.apache.flink.api.java.functions.KeySelector[IN,K],x$3: org.apache.flink.api.common.typeinfo.TypeInformation[K],x$4: org.apache.flink.runtime.operators.testutils.MockEnvironment)org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness[K,IN,OUT] &lt;and&gt;
    (x$1: org.apache.flink.streaming.api.operators.OneInputStreamOperator[IN,OUT],x$2: org.apache.flink.api.java.functions.KeySelector[IN,K],x$3: org.apache.flink.api.common.typeinfo.TypeInformation[K])org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness[K,IN,OUT] &lt;and&gt;
    (x$1: org.apache.flink.streaming.api.operators.StreamOperatorFactory[OUT],x$2: org.apache.flink.api.java.functions.KeySelector[IN,K],x$3: org.apache.flink.api.common.typeinfo.TypeInformation[K])org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness[K,IN,OUT] &lt;and&gt;
    (x$1: org.apache.flink.streaming.api.operators.StreamOperatorFactory[OUT],x$2: org.apache.flink.api.java.functions.KeySelector[IN,K],x$3: org.apache.flink.api.common.typeinfo.TypeInformation[K],x$4: Int,x$5: Int,x$6: Int)org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness[K,IN,OUT] &lt;and&gt;
    (x$1: org.apache.flink.streaming.api.operators.OneInputStreamOperator[IN,OUT],x$2: org.apache.flink.api.java.functions.KeySelector[IN,K],x$3: org.apache.flink.api.common.typeinfo.TypeInformation[K],x$4: Int,x$5: Int,x$6: Int)org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness[K,IN,OUT]
   cannot be applied to (org.apache.flink.streaming.api.operators.KeyedProcessOperator[Int,org.example.TaxiRide,Int], org.example.TaxiRide =&gt; Integer, org.apache.flink.api.common.typeinfo.TypeInformation[Integer])
       testHarness = new KeyedOneInputStreamOperatorTestHarness(operator, (ride: TaxiRide) =&gt; ride.rideId, Types.INT)
  </code></pre>
  <p>It looks like that it cannot accept Scala types, why is it so?</p>