KAFKA-19946: Simplify skipping of empty ShareFetch requests (#21033)

ShareFetch requests can be used to fetch records for share consumers,
but in some cases, no records are required and the request is just being
used to update the share session or send acknowledgements. As a result,
there are situations in which a ShareFetch request would be built, only
for it to be entirely empty (no fetch, no asks, no share session
update). This PR simplifies the logic for detecting when the request is
empty and then to avoid building it entirely. It also adds some tests
for this case.

Reviewers: Apoorv Mittal <apoorvmittal10@gmail.com>

Author: AndrewJSchofield

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

@@ -253,24 +253,20 @@ public PollResult poll(long currentTimeMs) {
             Node target = entry.getKey();
             ShareSessionHandler handler = entry.getValue();
 
-            log.trace("Building ShareFetch request to send to node {}", target.id());
-            ShareFetchRequest.Builder requestBuilder = handler.newShareFetchBuilder(groupId, shareFetchConfig);
-
             // For record_limit mode, we only send a full ShareFetch to a single node at a time.
             // We prepare to build ShareFetch requests for all nodes with session handlers to permit
             // piggy-backing of acknowledgements, and also to adjust the topic-partitions
-            // in the share session.
-            if (isShareAcquireModeRecordLimit() && target.id() != fetchRecordsNodeId.get()) {
-                ShareFetchRequestData data = requestBuilder.data();
-                // If there's nothing to send, just skip building the record.
-                if (data.topics().isEmpty() && data.forgottenTopicsData().isEmpty()) {
-                    return null;
-                } else {
-                    // There is something to send, but we don't want to fetch any records.
-                    requestBuilder.data().setMaxRecords(0);
-                }
+            // in the share session, but if the request would contain neither of those, it can be skipped.
+            boolean canSkipIfRequestEmpty = isShareAcquireModeRecordLimit() && target.id() != fetchRecordsNodeId.get();
+
+            ShareFetchRequest.Builder requestBuilder = handler.newShareFetchBuilder(groupId, shareFetchConfig, canSkipIfRequestEmpty);
+            if (requestBuilder == null) {
+                log.trace("Skipping ShareFetch request to send to node {}", target.id());
+                return null;
             }
 
+            log.trace("Building ShareFetch request to send to node {}", target.id());
+
             nodesWithPendingRequests.add(target.id());
 
             BiConsumer<ClientResponse, Throwable> responseHandler = (clientResponse, error) -> {

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

@@ -54,6 +54,7 @@
  * <p>ShareSessionHandler tracks the partitions which are in the session. It also determines
  * which partitions need to be included in each ShareFetch/ShareAcknowledge request.
  */
+@SuppressWarnings({"NPathComplexity", "CyclomaticComplexity"})
 public class ShareSessionHandler {
     private final Logger log;
     private final int node;
@@ -112,7 +113,7 @@ public boolean isNewSession() {
         return nextMetadata.isNewSession();
     }
 
-    public ShareFetchRequest.Builder newShareFetchBuilder(String groupId, ShareFetchConfig shareFetchConfig) {
+    public ShareFetchRequest.Builder newShareFetchBuilder(String groupId, ShareFetchConfig shareFetchConfig, boolean canSkipIfRequestEmpty) {
         List<TopicIdPartition> added = new ArrayList<>();
         List<TopicIdPartition> removed = new ArrayList<>();
         List<TopicIdPartition> replaced = new ArrayList<>();
@@ -158,15 +159,6 @@ public ShareFetchRequest.Builder newShareFetchBuilder(String groupId, ShareFetch
             }
         }
 
-        if (log.isDebugEnabled()) {
-            log.debug("Build ShareFetch {} for node {}. Added {}, removed {}, replaced {} out of {}",
-                    nextMetadata, node,
-                    topicIdPartitionsToLogString(added),
-                    topicIdPartitionsToLogString(removed),
-                    topicIdPartitionsToLogString(replaced),
-                    topicIdPartitionsToLogString(sessionPartitions.values()));
-        }
-
         // The replaced topic-partitions need to be removed, and their replacements are already added
         removed.addAll(replaced);
 
@@ -187,6 +179,19 @@ public ShareFetchRequest.Builder newShareFetchBuilder(String groupId, ShareFetch
         nextPartitions = new LinkedHashMap<>();
         nextAcknowledgements = new LinkedHashMap<>();
 
+        if (canSkipIfRequestEmpty && added.isEmpty() && removed.isEmpty() && acknowledgementBatches.isEmpty()) {
+            return null;
+        }
+
+        if (log.isDebugEnabled()) {
+            log.debug("Build ShareFetch {} for node {}. Added {}, removed {}, replaced {} out of {}",
+                nextMetadata, node,
+                topicIdPartitionsToLogString(added),
+                topicIdPartitionsToLogString(removed),
+                topicIdPartitionsToLogString(replaced),
+                topicIdPartitionsToLogString(sessionPartitions.values()));
+        }
+
         if (hasRenewAcknowledgements) {
             // If the request has renew acknowledgements, the ShareFetch is only used to send the acknowledgements
             // and potentially update the share session. The parameters for wait time, number of bytes and number of
@@ -196,6 +201,14 @@ public ShareFetchRequest.Builder newShareFetchBuilder(String groupId, ShareFetch
                 0, 0, 0,
                 0, shareFetchConfig.shareAcquireMode.id, true,
                 added, removed, acknowledgementBatches);
+        } else if (canSkipIfRequestEmpty) {
+            // The request contains changes to the share session or acknowledgements only. The parameters for wait time,
+            // number of bytes and number of records are all zero.
+            return ShareFetchRequest.Builder.forConsumer(
+                groupId, nextMetadata, 0,
+                0, 0, 0,
+                0, shareFetchConfig.shareAcquireMode.id, false,
+                added, removed, acknowledgementBatches);
         } else {
             return ShareFetchRequest.Builder.forConsumer(
                 groupId, nextMetadata, shareFetchConfig.maxWaitMs,