Intent to backport to 1.4, 1.2 : OAK-4668

[Chetan Mehrotra]

Need to backport OAK-4668 [1] to older branches. This is required to
prevent loss of checkpoint in unstable topology

Chetan Mehrotra
[1] https://issues.apache.org/jira/browse/OAK-4668