79098029

Date: 2024-10-17 12:21:46
Score: 0.5
Natty:
Report link

We were able to find the underlying issue why the state was not restored properly.

In the code block where both operators and the sink are joined by the CoProcessFunction, I only provided a uid/name to the sink and not to the function itself. Since the graph built for the initializer application differs from the graph of the one that should take over, flink was not able to map the state properly.

Using env.disableOperatorChaining(); provided some extra insight to the situation by displaying every node separately.

sensorStreamOperator
    .keyBy(SensorEvent::getMachineId)
    .connect(configStreamOperator.keyBy(MachineConfig::getMachineId))
    .process(new HandleEventDataCoProcess(outputTag)
    .uid("CoProcessUID")  // Provide separate uid to the processor!
    .name("CoProcessFunction")
    .getSideOutput(outputTag)
    .sinkTo(outlierSink)
    .uid("OutlierSinkUID")
    .name("OutlierSink");
Reasons:
  • Long answer (-0.5):
  • Has code block (-0.5):
  • Self-answer (0.5):
  • Low reputation (1):
Posted by: mauam