[FLINK-5985] Report no task states for stateless tasks on checkpointing

[StefanRRichter]

This PR fixes [FLINK-5985]. The solution is based on acknowledging null instead of some empty SubtaskState to CheckpointCoordinator#acknowledgeCheckpoint(...), so that no TaskState is registered under the JobVertexID of a stateless task in the checkpoint.

[StefanRRichter]

CC @gyfora @uce

[gyfora]

The changes look reasonable :)

[gyfora]

I could only try the backported version on the topology that caused the problem initally (that is running 1.2.0)

[StefanRRichter]

@gyfora if the effort is reasonable, it would be great to try this out on your topology. As soon as you give your +1, I could merge this change :-)

[gyfora]

Im gonna try to cherry-pick this on 1.2 and run it today

[StefanRRichter]

Great, thanks!

[gyfora]

There seems to have been some changes in the StreamTask and some tests so I couldn't rebase this nicely. Do you have a minute to take a look and maybe push a branch with the backport please? That would help me a lot.

[StefanRRichter]

Sure, I just quickly prepared a backport here:

https://github.com/StefanRRichter/flink/tree/FLINK-5985-backport-to-1.2

[gyfora]

Hm, doesnt seem to work for the first try. What I did is I updated the client with the new jar based on your backport branch. Redeployed the job with a savepoint (to get the new Flink version), took a savepoint and tried to redeploy with the changed topology.

I still seem to get the same error.

Is it possible that the previous checkpoints have an effect on this? In any case I will double check tomorrow morning and try to do the test again.

[gyfora]

It also doesnt seem to work starting from a clean state and then savepoint redeploy with changed topology so maybe I am really screwing up something

[gyfora]

@StefanRRichter It seems to work correctly locally, I am trying to see what went wrong with my yarn tests, but this shouldnt block you

[gyfora]

Ah, the reason is probably that I didnt change my job jar, and this relies on changes in the rocks backend

[StefanRRichter]

Ok, then the mystery is finally solved :-) Thanks again for reporting this problem and your additional testing efforts!

[gyfora]

On a second thought, shouldnt the older rocks backend version still work? (I guess that still returns a Done future with null value)

[StefanRRichter]

Yes, it should still work because the changes on RocksDBKeyedStateBackend are purely cosmetical without changing any functionality.

[StefanRRichter]

Hm, one potential pitfall that I see is operator chaining, in case your stateless operators are chained together with stateful ones. But then again, you said it works locally?

[StephanEwen]

I think this fix is good.
The StreamTask could use some cleanup to make tests easier (less whiteboxing needed). We should do that in a separate refactoring, either as a followup to this, or as a preparation for a modified version of this.

What do you think?

[StephanEwen]

I think the fact that this test requires extensive whiteboxing means we should move the whole CheckpointOperation to a separate class and make it work independent of StreamTask.

[StefanRRichter]

Yes, we could separate the classes. CheckpointOperation is already a static inner class anyways. I would suggest to do this in a followup.

[StephanEwen]

I think that verifying the possibility to reconfigure a job with respect to stateless operators warrants an ITCase. Can we extend the SavepointITCase for that?

[StefanRRichter]

@StephanEwen, I have added the IT case. Please have another look.

[StephanEwen]

Can we slightly adapt the test to target more the typical use case:

• Original job has some stateless ops (no uid), and some stateful ones (with uid)
• Create a modified job that has the same stateful ones (same uids) but different stateless ones

Otherwise this looks good.

[StefanRRichter]

Thanks for the review @StephanEwen. I updated the test as suggested. Merging this now.