KAFKA-18935: Ensure brokers do not return null records in FetchResponse (#19167)

JIRA: KAFKA-18935  This patch ensures the broker will not return null
records in FetchResponse.   For more details, please refer to the
ticket.

Reviewers: Ismael Juma <[email protected]>, Chia-Ping Tsai
 <[email protected]>, Jun Rao <[email protected]>

Author: frankvicky

clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java

@@ -360,7 +360,7 @@ public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
                         .setPartitions(partitionResponses));
             });
         }
-        return new FetchResponse(new FetchResponseData()
+        return FetchResponse.of(new FetchResponseData()
                 .setThrottleTimeMs(throttleTimeMs)
                 .setErrorCode(error.code())
                 .setSessionId(data.sessionId())

clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java

@@ -87,7 +87,7 @@ public FetchResponseData data() {
      * We may also return INCONSISTENT_TOPIC_ID error as a partition-level error when a partition in the session has a topic ID
      * inconsistent with the log.
      */
-    public FetchResponse(FetchResponseData fetchResponseData) {
+    private FetchResponse(FetchResponseData fetchResponseData) {
         super(ApiKeys.FETCH);
         this.data = fetchResponseData;
     }
@@ -138,6 +138,13 @@ public Map<Errors, Integer> errorCounts() {
         return errorCounts;
     }
 
+    /**
+     * Creates a {@link org.apache.kafka.common.requests.FetchResponse} from the given byte buffer.
+     * Unlike {@link org.apache.kafka.common.requests.FetchResponse#of(FetchResponseData)}, this method doesn't convert
+     * null records to {@link org.apache.kafka.common.record.MemoryRecords#EMPTY}.
+     *
+     * <p><strong>This method should only be used in client-side.</strong></p>
+     */
     public static FetchResponse parse(ByteBuffer buffer, short version) {
         return new FetchResponse(new FetchResponseData(new ByteBufferAccessor(buffer), version));
     }
@@ -220,6 +227,23 @@ public static int recordsSize(FetchResponseData.PartitionData partition) {
         return partition.records() == null ? 0 : partition.records().sizeInBytes();
     }
 
+    /**
+     * Creates a {@link org.apache.kafka.common.requests.FetchResponse} from the given data.
+     * This method converts null records to {@link org.apache.kafka.common.record.MemoryRecords#EMPTY}
+     * to ensure consistent record representation in the response.
+     *
+     * <p><strong>This method should only be used in server-side.</strong></p>
+     */
+    public static FetchResponse of(FetchResponseData data) {
+        for (FetchResponseData.FetchableTopicResponse response : data.responses()) {
+            for (FetchResponseData.PartitionData partition : response.partitions()) {
+                if (partition.records() == null)
+                    partition.setRecords(MemoryRecords.EMPTY);
+            }
+        }
+        return new FetchResponse(data);
+    }
+
     // TODO: remove as a part of KAFKA-12410
     public static FetchResponse of(Errors error,
                                    int throttleTimeMs,
@@ -258,6 +282,11 @@ private static FetchResponseData toMessage(Errors error,
             FetchResponseData.PartitionData partitionData = entry.getValue();
             // Since PartitionData alone doesn't know the partition ID, we set it here
             partitionData.setPartitionIndex(entry.getKey().topicPartition().partition());
+            // To protect the clients from failing due to null records,
+            // we always convert null records to MemoryRecords.EMPTY
+            // We will propose a KIP to change the schema definitions in the future
+            if (partitionData.records() == null)
+                partitionData.setRecords(MemoryRecords.EMPTY);
             // We have to keep the order of input topic-partition. Hence, we batch the partitions only if the last
             // batch is in the same topic group.
             FetchResponseData.FetchableTopicResponse previousTopic = topicResponseList.isEmpty() ? null

clients/src/main/java/org/apache/kafka/common/requests/ShareFetchRequest.java

@@ -20,13 +20,13 @@
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.message.ShareFetchRequestData;
-import org.apache.kafka.common.message.ShareFetchResponseData;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.protocol.Readable;
 
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -161,9 +161,7 @@ public ShareFetchRequestData data() {
     @Override
     public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
         Errors error = Errors.forException(e);
-        return new ShareFetchResponse(new ShareFetchResponseData()
-                .setThrottleTimeMs(throttleTimeMs)
-                .setErrorCode(error.code()));
+        return ShareFetchResponse.of(error, throttleTimeMs, new LinkedHashMap<>(), List.of(), 0);
     }
 
     public static ShareFetchRequest parse(Readable readable, short version) {

clients/src/main/java/org/apache/kafka/common/requests/ShareFetchResponse.java

@@ -55,7 +55,7 @@ public class ShareFetchResponse extends AbstractResponse {
 
     private final ShareFetchResponseData data;
 
-    public ShareFetchResponse(ShareFetchResponseData data) {
+    private ShareFetchResponse(ShareFetchResponseData data) {
         super(ApiKeys.SHARE_FETCH);
         this.data = data;
     }
@@ -103,6 +103,13 @@ public void maybeSetThrottleTimeMs(int throttleTimeMs) {
         data.setThrottleTimeMs(throttleTimeMs);
     }
 
+    /**
+     * Creates a {@link org.apache.kafka.common.requests.ShareFetchResponse} from the given byte buffer.
+     * Unlike {@link org.apache.kafka.common.requests.ShareFetchResponse#of(Errors, int, LinkedHashMap, List, int)},
+     * this method doesn't convert null records to {@link org.apache.kafka.common.record.MemoryRecords#EMPTY}.
+     *
+     * <p><strong>This method should only be used in client-side.</strong></p>
+     */
     public static ShareFetchResponse parse(ByteBuffer buffer, short version) {
         return new ShareFetchResponse(
                 new ShareFetchResponseData(new ByteBufferAccessor(buffer), version)
@@ -145,14 +152,21 @@ public static int recordsSize(ShareFetchResponseData.PartitionData partition) {
         return partition.records() == null ? 0 : partition.records().sizeInBytes();
     }
 
+    /**
+     * Creates a {@link org.apache.kafka.common.requests.ShareFetchResponse} from the given data.
+     * This method converts null records to {@link org.apache.kafka.common.record.MemoryRecords#EMPTY}
+     * to ensure consistent record representation in the response.
+     *
+     * <p><strong>This method should only be used in server-side.</strong></p>
+     */
     public static ShareFetchResponse of(Errors error,
                                         int throttleTimeMs,
                                         LinkedHashMap<TopicIdPartition, ShareFetchResponseData.PartitionData> responseData,
                                         List<Node> nodeEndpoints, int acquisitionLockTimeout) {
         return new ShareFetchResponse(toMessage(error, throttleTimeMs, responseData.entrySet().iterator(), nodeEndpoints, acquisitionLockTimeout));
     }
 
-    public static ShareFetchResponseData toMessage(Errors error, int throttleTimeMs,
+    private static ShareFetchResponseData toMessage(Errors error, int throttleTimeMs,
                                                    Iterator<Map.Entry<TopicIdPartition, ShareFetchResponseData.PartitionData>> partIterator,
                                                    List<Node> nodeEndpoints, int acquisitionLockTimeout) {
         Map<Uuid, ShareFetchResponseData.ShareFetchableTopicResponse> topicResponseList = new LinkedHashMap<>();
@@ -161,6 +175,11 @@ public static ShareFetchResponseData toMessage(Errors error, int throttleTimeMs,
             ShareFetchResponseData.PartitionData partitionData = entry.getValue();
             // Since PartitionData alone doesn't know the partition ID, we set it here
             partitionData.setPartitionIndex(entry.getKey().topicPartition().partition());
+            // To protect the clients from failing due to null records,
+            // we always convert null records to MemoryRecords.EMPTY
+            // We will propose a KIP to change the schema definitions in the future
+            if (partitionData.records() == null)
+                partitionData.setRecords(MemoryRecords.EMPTY);
             // Checking if the topic is already present in the map
             if (topicResponseList.containsKey(entry.getKey().topicId())) {
                 topicResponseList.get(entry.getKey().topicId()).partitions().add(partitionData);
@@ -193,6 +212,7 @@ public static ShareFetchResponseData.PartitionData partitionResponse(TopicIdPart
     public static ShareFetchResponseData.PartitionData partitionResponse(int partition, Errors error) {
         return new ShareFetchResponseData.PartitionData()
                 .setPartitionIndex(partition)
-                .setErrorCode(error.code());
+                .setErrorCode(error.code())
+                .setRecords(MemoryRecords.EMPTY);
     }
 }

clients/src/test/java/org/apache/kafka/clients/consumer/KafkaShareConsumerTest.java

@@ -56,6 +56,7 @@
 import java.nio.ByteBuffer;
 import java.time.Duration;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -392,13 +393,7 @@ private ShareFetchResponse shareFetchResponse(TopicIdPartition tip, int count) {
             .setPartitionIndex(tip.partition())
             .setRecords(records)
             .setAcquiredRecords(List.of(new ShareFetchResponseData.AcquiredRecords().setFirstOffset(0).setLastOffset(count - 1).setDeliveryCount((short) 1)));
-        ShareFetchResponseData.ShareFetchableTopicResponse topicResponse = new ShareFetchResponseData.ShareFetchableTopicResponse()
-            .setTopicId(tip.topicId())
-            .setPartitions(List.of(partData));
-        return new ShareFetchResponse(
-            new ShareFetchResponseData()
-                .setResponses(List.of(topicResponse))
-        );
+        return ShareFetchResponse.of(Errors.NONE, 0, new LinkedHashMap<>(Map.of(tip, partData)), List.of(), 0);
     }
 
     private ShareAcknowledgeResponse shareAcknowledgeResponse() {

clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareSessionHandlerTest.java

@@ -139,15 +139,14 @@ private static final class RespEntry {
         }
     }
 
-    private static List<ShareFetchResponseData.ShareFetchableTopicResponse> respList(RespEntry... entries) {
-        HashMap<TopicIdPartition, ShareFetchResponseData.ShareFetchableTopicResponse> map = new HashMap<>();
+    private static LinkedHashMap<TopicIdPartition, ShareFetchResponseData.PartitionData> buildResponseData(RespEntry... entries) {
+        LinkedHashMap<TopicIdPartition, ShareFetchResponseData.PartitionData> topicIdPartitionToPartition = new LinkedHashMap<>();
         for (RespEntry entry : entries) {
-            ShareFetchResponseData.ShareFetchableTopicResponse response = map.computeIfAbsent(entry.part, topicIdPartition ->
-                    new ShareFetchResponseData.ShareFetchableTopicResponse().setTopicId(topicIdPartition.topicId()));
-            response.partitions().add(new ShareFetchResponseData.PartitionData()
-                    .setPartitionIndex(entry.part.partition()));
+            ShareFetchResponseData.PartitionData partitionData = new ShareFetchResponseData.PartitionData()
+                .setPartitionIndex(entry.part.partition());
+            topicIdPartitionToPartition.put(entry.part, partitionData);
         }
-        return new ArrayList<>(map.values());
+        return topicIdPartitionToPartition;
     }
 
     @Test
@@ -170,13 +169,11 @@ public void testShareSession() {
         assertListEquals(expectedToSend1, reqFetchList(requestData1, topicNames));
         assertEquals(memberId.toString(), requestData1.memberId());
 
-        ShareFetchResponse resp = new ShareFetchResponse(
-                new ShareFetchResponseData()
-                        .setErrorCode(Errors.NONE.code())
-                        .setThrottleTimeMs(0)
-                        .setResponses(respList(
-                                new RespEntry("foo", 0, fooId),
-                                new RespEntry("foo", 1, fooId))));
+        ShareFetchResponse resp = ShareFetchResponse.of(Errors.NONE,
+            0,
+            buildResponseData(new RespEntry("foo", 0, fooId), new RespEntry("foo", 1, fooId)),
+            List.of(),
+            0);
         handler.handleResponse(resp, ApiKeys.SHARE_FETCH.latestVersion(true));
 
         // Test a fetch request which adds one partition
@@ -194,18 +191,15 @@ public void testShareSession() {
         expectedToSend2.add(new TopicIdPartition(barId, 0, "bar"));
         assertListEquals(expectedToSend2, reqFetchList(requestData2, topicNames));
 
-        ShareFetchResponse resp2 = new ShareFetchResponse(
-                new ShareFetchResponseData()
-                        .setErrorCode(Errors.NONE.code())
-                        .setThrottleTimeMs(0)
-                        .setResponses(respList(
-                                new RespEntry("foo", 1, fooId))));
+        ShareFetchResponse resp2 = ShareFetchResponse.of(Errors.NONE,
+            0,
+            buildResponseData(new RespEntry("foo", 1, fooId)),
+            List.of(),
+            0);
         handler.handleResponse(resp2, ApiKeys.SHARE_FETCH.latestVersion(true));
 
         // A top-level error code will reset the session epoch
-        ShareFetchResponse resp3 = new ShareFetchResponse(
-                new ShareFetchResponseData()
-                        .setErrorCode(Errors.INVALID_SHARE_SESSION_EPOCH.code()));
+        ShareFetchResponse resp3 = ShareFetchResponse.of(Errors.INVALID_SHARE_SESSION_EPOCH, 0, new LinkedHashMap<>(), List.of(), 0);
         handler.handleResponse(resp3, ApiKeys.SHARE_FETCH.latestVersion(true));
 
         ShareFetchRequestData requestData4 = handler.newShareFetchBuilder(groupId, fetchConfig).build().data();
@@ -251,14 +245,14 @@ public void testPartitionRemoval() {
         assertListEquals(expectedToSend1, reqFetchList(requestData1, topicNames));
         assertEquals(memberId.toString(), requestData1.memberId());
 
-        ShareFetchResponse resp = new ShareFetchResponse(
-                new ShareFetchResponseData()
-                        .setErrorCode(Errors.NONE.code())
-                        .setThrottleTimeMs(0)
-                        .setResponses(respList(
-                                new RespEntry("foo", 0, fooId),
-                                new RespEntry("foo", 1, fooId),
-                                new RespEntry("bar", 0, barId))));
+        ShareFetchResponse resp = ShareFetchResponse.of(Errors.NONE,
+            0,
+            buildResponseData(
+                new RespEntry("foo", 0, fooId),
+                new RespEntry("foo", 1, fooId),
+                new RespEntry("bar", 0, barId)),
+            List.of(),
+            0);
         handler.handleResponse(resp, ApiKeys.SHARE_FETCH.latestVersion(true));
 
         // Test a fetch request which removes two partitions
@@ -275,9 +269,7 @@ public void testPartitionRemoval() {
         assertListEquals(expectedToForget2, reqForgetList(requestData2, topicNames));
 
         // A top-level error code will reset the session epoch
-        ShareFetchResponse resp2 = new ShareFetchResponse(
-                new ShareFetchResponseData()
-                        .setErrorCode(Errors.INVALID_SHARE_SESSION_EPOCH.code()));
+        ShareFetchResponse resp2 = ShareFetchResponse.of(Errors.INVALID_SHARE_SESSION_EPOCH, 0, new LinkedHashMap<>(), List.of(), 0);
         handler.handleResponse(resp2, ApiKeys.SHARE_FETCH.latestVersion(true));
 
         handler.addPartitionToFetch(foo1, null);
@@ -309,12 +301,11 @@ public void testTopicIdReplaced() {
         expectedToSend1.add(new TopicIdPartition(topicId1, 0, "foo"));
         assertListEquals(expectedToSend1, reqFetchList(requestData1, topicNames));
 
-        ShareFetchResponse resp = new ShareFetchResponse(
-                new ShareFetchResponseData()
-                        .setErrorCode(Errors.NONE.code())
-                        .setThrottleTimeMs(0)
-                        .setResponses(respList(
-                                new RespEntry("foo", 0, topicId1))));
+        ShareFetchResponse resp = ShareFetchResponse.of(Errors.NONE,
+            0,
+            buildResponseData(new RespEntry("foo", 0, topicId1)),
+            List.of(),
+            0);
         handler.handleResponse(resp, ApiKeys.SHARE_FETCH.latestVersion(true));
 
         // Try to add a new topic ID
@@ -354,12 +345,11 @@ public void testPartitionForgottenOnAcknowledgeOnly() {
         expectedToSend1.add(new TopicIdPartition(topicId, 0, "foo"));
         assertListEquals(expectedToSend1, reqFetchList(requestData1, topicNames));
 
-        ShareFetchResponse resp = new ShareFetchResponse(
-                new ShareFetchResponseData()
-                        .setErrorCode(Errors.NONE.code())
-                        .setThrottleTimeMs(0)
-                        .setResponses(respList(
-                                new RespEntry("foo", 0, topicId))));
+        ShareFetchResponse resp = ShareFetchResponse.of(Errors.NONE,
+            0,
+            buildResponseData(new RespEntry("foo", 0, topicId)),
+            List.of(),
+            0);
         handler.handleResponse(resp, ApiKeys.SHARE_FETCH.latestVersion(true));
 
         // Remove the topic from the session by setting acknowledgements only - this is not asking to fetch records
@@ -390,12 +380,11 @@ public void testForgottenPartitions() {
         expectedToSend1.add(new TopicIdPartition(topicId, 0, "foo"));
         assertListEquals(expectedToSend1, reqFetchList(requestData1, topicNames));
 
-        ShareFetchResponse resp = new ShareFetchResponse(
-            new ShareFetchResponseData()
-                .setErrorCode(Errors.NONE.code())
-                .setThrottleTimeMs(0)
-                .setResponses(respList(
-                    new RespEntry("foo", 0, topicId))));
+        ShareFetchResponse resp = ShareFetchResponse.of(Errors.NONE,
+            0,
+            buildResponseData(new RespEntry("foo", 0, topicId)),
+            List.of(),
+            0);
         handler.handleResponse(resp, ApiKeys.SHARE_FETCH.latestVersion(true));
 
         // Remove the topic from the session
@@ -424,23 +413,18 @@ public void testAddNewIdAfterTopicRemovedFromSession() {
         expectedToSend1.add(new TopicIdPartition(topicId, 0, "foo"));
         assertListEquals(expectedToSend1, reqFetchList(requestData1, topicNames));
 
-        ShareFetchResponse resp = new ShareFetchResponse(
-                new ShareFetchResponseData()
-                        .setErrorCode(Errors.NONE.code())
-                        .setThrottleTimeMs(0)
-                        .setResponses(respList(
-                                new RespEntry("foo", 0, topicId))));
+        ShareFetchResponse resp = ShareFetchResponse.of(Errors.NONE,
+            0,
+            buildResponseData(new RespEntry("foo", 0, topicId)),
+            List.of(),
+            0);
         handler.handleResponse(resp, ApiKeys.SHARE_FETCH.latestVersion(true));
 
         // Remove the partition from the session
         ShareFetchRequestData requestData2 = handler.newShareFetchBuilder(groupId, fetchConfig).build().data();
         assertTrue(handler.sessionPartitionMap().isEmpty());
         assertTrue(requestData2.topics().isEmpty());
-        ShareFetchResponse resp2 = new ShareFetchResponse(
-                new ShareFetchResponseData()
-                        .setErrorCode(Errors.NONE.code())
-                        .setThrottleTimeMs(0)
-                        .setResponses(respList()));
+        ShareFetchResponse resp2 = ShareFetchResponse.of(Errors.NONE, 0, new LinkedHashMap<>(), List.of(), 0);
         handler.handleResponse(resp2, ApiKeys.SHARE_FETCH.latestVersion(true));
 
         // After the topic is removed, add a recreated topic with a new ID