[jira] [Created] (FLINK-17578) Union of 2 SideOutputs behaviour incorrect

["Tom Wells (Jira)" <ji...@apache.org>]

Tom Wells created FLINK-17578:
---------------------------------

             Summary: Union of 2 SideOutputs behaviour incorrect
                 Key: FLINK-17578
                 URL: https://issues.apache.org/jira/browse/FLINK-17578
             Project: Flink
          Issue Type: Bug
          Components: API / DataStream
    Affects Versions: 1.10.0
            Reporter: Tom Wells


Strange behaviour when using union() to merge outputs of 2 DataStreams, where both are sourced from SideOutputs.

See example code with comments demonstrating the issue:
{code:java}
// code placeholder
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val input = env.fromElements(1, 2, 3, 4)

    val oddTag = OutputTag[Int]("odds")
    val evenTag = OutputTag[Int]("even")

    val all =
      input.process {
        (value: Int, ctx: ProcessFunction[Int, Int]#Context, out: Collector[Int]) => {
          if (value % 2 != 0)
            ctx.output(oddTag, value)
          else
            ctx.output(evenTag, value)
        }
      }

    val odds = all.getSideOutput(oddTag)
    val evens = all.getSideOutput(evenTag)

    // These print correctly
    //
    odds.print                  // -> 1, 3
    evens.print                 // -> 2, 4

    // This prints incorrectly - BUG?
    //
    odds.union(evens).print       // -> 2, 2, 4, 4
    evens.union(odds).print       // -> 1, 1, 3, 3

    // Another test to understand normal behaviour of .union, using normal inputs
    //
    val odds1 = env.fromElements(1, 3)
    val evens1 = env.fromElements(2, 4)

    // Union of 2 normal inputs
    //
    odds1.union(evens1).print   // -> 1, 2, 3, 4

    // Union of a normal input plus an input from a sideoutput
    //
    odds.union(evens1).print    // -> 1, 2, 3, 4
    evens1.union(odds).print    // -> 1, 2, 3, 4

    //
    // So it seems that when both inputs are from sideoutputs that it behaves incorrectly... BUG?

    env.execute("Test job")
  }



{code}



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