Merge remote-tracking branch 'origin/kip-932' into AKCORE-253

Author: ShivsundarR

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java

@@ -575,7 +575,6 @@ private void handleShareAcknowledgeFailure(Node fetchTarget,
                         metadata.topicNames().get(topic.topicId()));
                     metricsManager.recordFailedAcknowledgements(acknowledgeRequestState.getInFlightAcknowledgementsCount(tip));
                     acknowledgeRequestState.handleAcknowledgeErrorCode(tip, Errors.forException(error));
-                    acknowledgeRequestState.inFlightAcknowledgements.remove(tip);
             }));
         } finally {
             log.debug("Removing pending request for node {} - failed", fetchTarget);
@@ -599,7 +598,6 @@ private void handleShareAcknowledgeCloseSuccess(Node fetchTarget,
                     metricsManager.recordFailedAcknowledgements(acknowledgeRequestState.getInFlightAcknowledgementsCount(tip));
                 }
                 acknowledgeRequestState.handleAcknowledgeErrorCode(tip, Errors.forCode(partition.errorCode()));
-                acknowledgeRequestState.inFlightAcknowledgements.remove(tip);
             }));
 
             metricsManager.recordLatency(resp.requestLatencyMs());
@@ -625,7 +623,6 @@ private void handleShareAcknowledgeCloseFailure(Node fetchTarget,
                         metadata.topicNames().get(topic.topicId()));
                 metricsManager.recordFailedAcknowledgements(acknowledgeRequestState.getInFlightAcknowledgementsCount(tip));
                 acknowledgeRequestState.handleAcknowledgeErrorCode(tip, Errors.forException(error));
-                acknowledgeRequestState.inFlightAcknowledgements.remove(tip);
             }));
         } finally {
             log.debug("Removing pending request for node {} - failed", fetchTarget);
@@ -839,10 +836,6 @@ void retryRequest() {
             inFlightAcknowledgements.clear();
         }
 
-        boolean isIncompleteMapEmpty() {
-            return incompleteAcknowledgements.isEmpty();
-        }
-
         boolean maybeExpire() {
             return numAttempts > 0 && isExpired();
         }

tools/src/main/java/org/apache/kafka/tools/ShareConsumerPerformance.java

@@ -73,6 +73,7 @@ public static void main(String[] args) {
             Map<MetricName, ? extends Metric> metrics = null;
             if (options.printMetrics())
                 metrics = shareConsumer.metrics();
+            shareConsumer.commitAsync();
             shareConsumer.close();
 
             // print final stats