Hi everyone, I am trying to write an integration test for topologies we have. I am struggling to find a use case, when I could test the following scenario:

source -> stream1.connect(stream2).process(statefulFunction) -> sink

where statefulFunction has ValueState inside:

KeyedCoProcessFunction {
  ValueState state;
  processElement1() {
    if (state.value() != null) {
      // do something
    }
  }
  processElement2() {
    state.update()
  }
}

There is no guarantee what event first lands in statefulFunction, if they emitted almost at the same time. Our design presumes the events for processElement2 are passed first , landing in the state before processElement1 events starts flowing in. I wrote a test to emit events one by one. They wait each other before being emitted. But I still

sleep(5)

in controllable Source to give enough time for events to land on the right side. Maybe someone know better way of doing that, thank you! Let me know if I need to write the question in more details.

Normally you’d test this via a unit test, leveraging Flink’s test harness to precisely control the arrival of events to the operator. Why does it have to be an integration test?

Unit testing with test harness covers only operators and does not cover topology itself. The idea is to see how only a part of topology behaves in isolation without jumping into heavyweight e2e tests. Topology may grow quite complex. Maybe I am missing something and best practice is to write small topologies (2-3 DataStreams) and source/sink from/into persistence layer, like Kafka?

You said:

Our design presumes the events for processElement2 are passed first , landing in the state before processElement1 events starts flowing in.

Normally that’s not how you’d design a Flink workflow, as there isn’t any guarantee that’s (by default) provided by Flink as to the ordering of events. I would probably code up the

KeyedCoProcessFunction

to handle any order (using state to save incoming records that arrive at

processElement1()

before their corresponding record arrives at

processElement2()

.