def execute()

in ride-cleansing/src/main/scala/org/apache/flink/training/exercises/ridecleansing/scala/RideCleansingExercise.scala [50:61]

• 8 lines of code
• 1 McCabe index (conditional complexity)

    def execute(): JobExecutionResult = {
      val env = StreamExecutionEnvironment.getExecutionEnvironment

      // set up the pipeline
      env
        .addSource(source)
        .filter(new NYCFilter())
        .addSink(sink)

      // execute the pipeline and return the result
      env.execute()
    }