Flink ProcessWindowFunction Compilation Error

Question

import com.typesafe.config.ConfigFactory
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.assigners._
import org.apache.flink.streaming.api.windowing.evictors.CountEvictor
import org.apache.flink.streaming.api.windowing.time.Time
import org.project.async.core.job.FlinkKafkaConnector
import org.project.functions.TestWindowFunction
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction // USED SCALA API
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)

    val source = env.fromElements(("hello", 1), ("hello", 2))

    val window1 = source
      .keyBy(_._1)
      .window(EventTimeSessionWindows.withGap(Time.seconds(1)))
      .evictor(CountEvictor.of(2))
      .process(new TestWindowFunction)




====TestWindowFunction.scala=====

class TestWindowFunction
  extends ProcessWindowFunction[(String, Int), (String, String, Int), String, TimeWindow] {

  override def process(
                        key: String,
                        window: Context,
                        input: Iterable[(String, Int)],
                        out: Collector[(String, String, Int)]): Unit = {

    input.foreach(e => out.collect((e._1, e._1, e._2)))
  }
}


=================ERROR===============

Task.scala:47: error: type mismatch;
[ERROR]  found   : org.project.functions.TestWindowFunction
[ERROR]  required: org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction[(String, Int),?,String,org.apache.flink.streaming.api.windowing.windows.TimeWindow]
[ERROR]       .process(new TestWindowFunction)
[ERROR]                ^
[ERROR] one error found

I'm trying to use window functionality to collect the events for some certain interval of time but I'm getting this compilation error and I could not able to under this issue also I refer this below issue still I could not able to solve it.

Link: Apache Flink: ProcessWindowFunction implementation

What could be wrong in this code?

Flink version: <flink.version>1.10.0</flink.version>

Answer 1

You need these imports for Scala

import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

and you should not use

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment

When in doubt in a situation like this, my solution is to delete all the imports and then have IntelliJ help me add them back (carefully).