KAFKA-20330: Ack handling improvement on broker restart

[AndrewJSchofield]

A share consumer uses a share session to keep track of its acquired
records with each share-partition leader it is talking to. When the
connection breaks, the share session is lost and acknowledgements fail.

When the connection to a share-partition leader failed while there was
an outstanding request, the share consumer noticed the disconnection and
failed the acknowledgements as expected. Also, if the leader changed to
a different broker, again the share consumer noticed the leadership
change and failed the acknowledgements as expected.

In the situation where the share-partition leadership did not change
when the broker restarted AND there was no in-flight request, the share
consumer did not notice the disconnection and would try to continue the
share session. There were also a few situations in which
acknowledgements could be lost and not notified to the acknowledgement
commit callback to do with this kind of leadership non-transition.

This PR improves the situation by using a consistent exception
NotLeaderOrFollowerException regardless of when the disconnection was
noticed. It also makes sure that acknowledgements which cannot be sent
are completed properly in all cases.

It is possible that a tweak to the protocol will be needed to eliminate
ShareSessionNotFoundException in all edge cases, but that would take a
KIP.

[DL1231]

Could you please rebase/merge trunk to resolve the conflicts?

[DL1231]

Thanks for the patch.

[DL1231]

Thanks fot the update, LGTM.

[apoorvmittal10]

Thanks for the patch, LGTM!