fix: Pipeline state during disconnects

[sakoush]

What this PR does / why we need it:

This PR fixes issues with regards to pipeline state inconsistency in cases of components failures (e.g. dataflow-engine, scheduler).

Which issue(s) this PR fixes:

Fixes INFRA-716 (internal)

Changes:

• Do not remove a pipeline from the scheduler local persistence store upon the user deleting this particular pipeline. This is required because if the scheduler restarts and before the pipeline delete is propagated and acknowledged by the different services (e.g. dataflow-engine and controller) then there is a risk that this state will become inconsistent. This has the drawback though that the internal state of the scheduler indefinitely increasing so we need to recycle it in a follow up work. For now we prioritise consistency over storage as we expect storage of protobuf control plan messages of these pipelines to be low.
• Fix a bug in dataflow-engine to report success when the pipeline topology is already running.
• Fix a bug in controller to not remove the finaliser of a pipeline if its state is PipelineTerminate (i.e. still to terminate).
• Return PipelineTerminating pipelines when calling GetAllRunningPipelineVersions, which allow us to handle the following case.
• For PipelineTerminating pipelines and no currently available dataflow-engines, set them to PipelineTerminated.
• If receiving an event for a pipeline with state PipelineTermiante or PipelineTerminating and no currently available dataflow-engines, set them to PipelineTerminated.
• We always send all pipelines (event deleted ones) when the controller connects as it might be that some messages have been missed during the disconnect.

Testing

• Induced a 2 min delay in datafllow pipeline create and delete (to allow time to kill components).

    private suspend fun handleDelete(metadata: PipelineMetadata) {
        logger.info("Delete pipeline ${metadata.name} version: ${metadata.version} id: ${metadata.id}")
        Thread.sleep(120_000)

    private suspend fun handleCreate(
        metadata: PipelineMetadata,
        steps: List<PipelineStepUpdate>,
        kafkaConsumerGroupIdPrefix: String,
        namespace: String,
    ) {
        logger.info("Create pipeline ${metadata.name} version: ${metadata.version} id: ${metadata.id}")
        Thread.sleep(120_000)

• Killing dataflow-engine, controller and scheduler pods, them make sure state is eventually consistent.

Special notes for your reviewer:

[lc525]

do we deal with the network glitch case?

[sakoush]

I think in the case of a networking glitch, then no the pipelines are going to remain in dataflow engine and not removed. We could repay all pipeline control plane messages up to a specific time and therefore could deal with glitches but will leave it to a follow up PR.

[lc525]

That sounds like a good potential solution, and agreed, to be dealt with in another PR

[lc525]

lgtm; left some minor observations/comments