[jira] [Closed] (FLINK-6298) Local execution is not setting RuntimeContext for RichOutputFormat

["Greg Hogan (JIRA)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/FLINK-6298?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Greg Hogan closed FLINK-6298.
-----------------------------
    Resolution: Fixed

Fixed in 1.3.0: 1dde7327058cb973bd4ff94bf5531d5093fb60a2
and 1.2.2: 5061e7e28dd3688801406c3cb1ebb4bfac012729

> Local execution is not setting RuntimeContext for RichOutputFormat
> ------------------------------------------------------------------
>
>                 Key: FLINK-6298
>                 URL: https://issues.apache.org/jira/browse/FLINK-6298
>             Project: Flink
>          Issue Type: Bug
>          Components: DataStream API
>    Affects Versions: 1.1.0, 1.2.0
>            Reporter: Mateusz Zakarczemny
>            Assignee: Wenlong Lyu
>             Fix For: 1.3.0, 1.2.2
>
>
> RuntimeContext is never set in RichOutputFormat. I tested it in local execution. RichMapFunction is setup correctly. 
> Following code will never print "//////Context set in RichOutputFormat"
> {code}
> import org.apache.flink.api.common.functions.{RichMapFunction, RuntimeContext}
> import org.apache.flink.api.common.io.RichOutputFormat
> import org.apache.flink.api.scala._
> import org.apache.flink.configuration.Configuration
> import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
> object Startup {
>   def main(args: Array[String]): Unit = {
>     val mapFunction = new RichMapFunction[String, String] {
>       def open(taskNumber: Int, numTasks: Int) { getRuntimeContext }
>       def map(event: String) = { event }
>       override def setRuntimeContext(t: RuntimeContext) = {
>         println("//////Context set in RichMapFunction")
>         super.setRuntimeContext(t)
>       }
>     }
>     val outputFormat = new RichOutputFormat[String] {
>       override def setRuntimeContext(t: RuntimeContext) = {
>         println("//////Context set in RichOutputFormat")
>         super.setRuntimeContext(t)
>       }
>       def open(taskNumber: Int, numTasks: Int) {}
>       def writeRecord(event: String) {
>         println(event)
>       }
>       def configure(parameters: Configuration): Unit = {}
>       def close(): Unit = {}
>     }
>     val see = StreamExecutionEnvironment.getExecutionEnvironment
>     val eventsStream = see.fromElements[String]("A", "B", "C").map(mapFunction)
>     eventsStream.writeUsingOutputFormat(outputFormat)
>     see.execute("test-job")
>   }
> }
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)