[jira] [Closed] (FLINK-18960) flink sideoutput union

["Dawid Wysakowicz (Jira)" <ji...@apache.org>]

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

Dawid Wysakowicz closed FLINK-18960.
------------------------------------
    Fix Version/s: 1.10.2
       Resolution: Duplicate

> flink sideoutput union
> ----------------------
>
>                 Key: FLINK-18960
>                 URL: https://issues.apache.org/jira/browse/FLINK-18960
>             Project: Flink
>          Issue Type: Bug
>          Components: API / DataStream
>    Affects Versions: 1.10.1
>            Reporter: xiaohang.li
>            Priority: Minor
>             Fix For: 1.10.2
>
>
> Flink sideoutput union seems not works right. If we union the sideoutput from the same operator, the output is the result of last side output times by the number of unions, which is not expected. For example,
> {code:java}
> val side = new OutputTag[String]("side")
>  val side2 = new OutputTag[String]("side2")
>  val side3 = new OutputTag[String]("side3")
>  val ds = env.socketTextStream("master",9001)
>  val res = ds.process(new ProcessFunction[String,String] {
>  override def processElement(value: String, ctx: ProcessFunction[String, String]#Context, out: Collector[String]): Unit = {
>  if(value.contains("hello"))
> { ctx.output(side,value) }
> else if(value.contains("world"))
> { ctx.output(side2,value) }
> else if(value.contains("flink"))
> { ctx.output(side3,value) }
> out.collect(value)
>  }
>  })
> val res1 = res.getSideOutput(side)
>  val res2 = res.getSideOutput(side2)
>  val res3 = res.getSideOutput(side3)
> println( "====>"+res1.getClass)
>  println( "====>"+res2.getClass)
> res1.print("res1")
>  res2.print("res2")
>  res3.print("res3")
> res2.union(res1).union(res3).print("all")
> {code}
>  
>  If we input 
> {code:java}
> hello
> world
> flink
> {code}
> The output will be 
>  
> {code:java}
> res1> hello
>  res2> world
>  res3> flink
>  all> flink
>  all> flink
>  all> flink
> {code}
>  
> But the expected output would be 
> {code:java}
> res1> hello
> res2> world
> res3> flink
> all> hello 
> all> world 
> all> flink
> {code}
>  
>  
> if we add a _map_ after the sideoutput and then union them, the output would be right, but adding map should be not needed. 
>  



--
This message was sent by Atlassian Jira
(v8.3.4#803005)