KAFKA-20736: Improve share session handling on leader change (apache#22766)

In some situations, the share consumer does not tidy up share sessions
sufficiently on leadership change. In particular, if a broker loses
leadership of a partition, the share consumer should send a further
ShareFetch/ShareAcknowledge to remove the partition from its share
session. In most cases, this removal happens naturally, but it depends
upon whether the share consumer has other requests to send to that
broker. This PR tidies up the maintenance of share sessions when
leadership changes occur.

Reviewers: Lianet Magrans <lmagrans@confluent.io>, Shivsundar R
 <shr@confluent.io>, Apoorv Mittal <apoorvmittal10@gmail.com>

Author: AndrewJSchofield

clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java

@@ -33,10 +33,9 @@
 import org.apache.kafka.common.errors.GroupMaxSizeReachedException;
 import org.apache.kafka.common.errors.InterruptException;
 import org.apache.kafka.common.errors.InvalidTopicException;
-import org.apache.kafka.common.errors.NotLeaderOrFollowerException;
+import org.apache.kafka.common.errors.NetworkException;
 import org.apache.kafka.common.errors.RecordDeserializationException;
 import org.apache.kafka.common.errors.SerializationException;
-import org.apache.kafka.common.errors.ShareSessionNotFoundException;
 import org.apache.kafka.common.errors.UnknownTopicIdException;
 import org.apache.kafka.common.errors.WakeupException;
 import org.apache.kafka.common.header.Header;
@@ -1054,7 +1053,7 @@ public void testLeaderRestartWithoutLeadershipChangeExplicitAcknowledgementSync(
 
             AtomicBoolean callbackCalled = new AtomicBoolean(false);
             shareConsumer.setAcknowledgementCommitCallback((offsetsByTopicPartition, exception) -> {
-                assertInstanceOf(NotLeaderOrFollowerException.class, exception);
+                assertInstanceOf(NetworkException.class, exception);
                 callbackCalled.set(true);
             });
 
@@ -1083,7 +1082,7 @@ public void testLeaderRestartWithoutLeadershipChangeExplicitAcknowledgementSync(
             assertEquals(1, commitResult.size());
             TopicIdPartition tidp = commitResult.keySet().iterator().next();
             assertTrue(commitResult.get(tidp).isPresent());
-            assertInstanceOf(NotLeaderOrFollowerException.class, commitResult.get(tidp).get());
+            assertInstanceOf(NetworkException.class, commitResult.get(tidp).get());
 
             assertTrue(callbackCalled.get());
         }
@@ -1098,7 +1097,7 @@ public void testLeaderRestartWithoutLeadershipChangeExplicitAcknowledgementAsync
 
             AtomicBoolean callbackCalled = new AtomicBoolean(false);
             shareConsumer.setAcknowledgementCommitCallback((offsetsByTopicPartition, exception) -> {
-                assertInstanceOf(NotLeaderOrFollowerException.class, exception);
+                assertInstanceOf(NetworkException.class, exception);
                 callbackCalled.set(true);
             });
 
@@ -1147,7 +1146,7 @@ public void testLeaderRestartWithoutLeadershipChangeImplicitAcknowledgement() {
 
             AtomicBoolean callbackCalled = new AtomicBoolean(false);
             shareConsumer.setAcknowledgementCommitCallback((offsetsByTopicPartition, exception) -> {
-                assertInstanceOf(ShareSessionNotFoundException.class, exception);
+                assertInstanceOf(NetworkException.class, exception);
                 callbackCalled.set(true);
             });