KAFKA-18889: Make records in ShareFetchResponse non-nullable

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

@@ -209,7 +209,7 @@ public static ShareFetchResponseData.PartitionData partitionResponse(TopicIdPart
         return partitionResponse(topicIdPartition.topicPartition().partition(), error);
     }
 
-    public static ShareFetchResponseData.PartitionData partitionResponse(int partition, Errors error) {
+    private static ShareFetchResponseData.PartitionData partitionResponse(int partition, Errors error) {
         return new ShareFetchResponseData.PartitionData()
                 .setPartitionIndex(partition)
                 .setErrorCode(error.code())

clients/src/main/resources/common/message/ShareFetchResponse.json

@@ -67,7 +67,7 @@
           { "name": "LeaderEpoch", "type": "int32", "versions": "0+",
             "about": "The latest known leader epoch." }
         ]},
-        { "name": "Records", "type": "records", "versions": "0+", "nullableVersions": "0+", "about": "The record data." },
+        { "name": "Records", "type": "records", "versions": "0+", "nullableVersions": "0", "about": "The record data." },
         { "name": "AcquiredRecords", "type": "[]AcquiredRecords", "versions": "0+", "about": "The acquired records.", "fields":  [
           { "name": "FirstOffset", "type":  "int64", "versions": "0+", "about": "The earliest offset in this batch of acquired records." },
           { "name": "LastOffset", "type": "int64", "versions": "0+", "about": "The last offset of this batch of acquired records." },

core/src/main/java/kafka/server/share/ShareFetchUtils.java

@@ -29,6 +29,7 @@
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.record.FileLogInputStream.FileChannelRecordBatch;
 import org.apache.kafka.common.record.FileRecords;
+import org.apache.kafka.common.record.MemoryRecords;
 import org.apache.kafka.common.record.Records;
 import org.apache.kafka.common.requests.ListOffsetsRequest;
 import org.apache.kafka.coordinator.group.GroupConfigManager;
@@ -83,7 +84,7 @@ static Map<TopicIdPartition, ShareFetchResponseData.PartitionData> processFetchR
 
             if (fetchPartitionData.error.code() != Errors.NONE.code()) {
                 partitionData
-                    .setRecords(null)
+                    .setRecords(MemoryRecords.EMPTY)
                     .setErrorCode(fetchPartitionData.error.code())
                     .setErrorMessage(fetchPartitionData.error.message())
                     .setAcquiredRecords(List.of());
@@ -123,7 +124,7 @@ static Map<TopicIdPartition, ShareFetchResponseData.PartitionData> processFetchR
                 // if we want parallel requests for the same share partition or not.
                 if (shareAcquiredRecords.acquiredRecords().isEmpty()) {
                     partitionData
-                        .setRecords(null)
+                        .setRecords(MemoryRecords.EMPTY)
                         .setAcquiredRecords(List.of());
                 } else {
                     partitionData

core/src/main/scala/kafka/server/KafkaApis.scala

@@ -3116,6 +3116,7 @@ class KafkaApis(val requestChannel: RequestChannel,
                     .setPartitionIndex(partitionIndex)
                     .setErrorCode(Errors.NONE.code)
                     .setAcknowledgeErrorCode(value)
+                    .setRecords(MemoryRecords.EMPTY)
                   topic.partitions.add(fetchPartitionData)
                 }
                 topicPartitionAcknowledgements.remove(topicId)
@@ -3131,6 +3132,7 @@ class KafkaApis(val requestChannel: RequestChannel,
                 .setPartitionIndex(partitionIndex)
                 .setErrorCode(Errors.NONE.code)
                 .setAcknowledgeErrorCode(value)
+                .setRecords(MemoryRecords.EMPTY)
               topicData.partitions.add(fetchPartitionData)
             }
             shareFetchResponse.data.responses.add(topicData)

core/src/test/java/kafka/server/share/ShareFetchUtilsTest.java

@@ -61,7 +61,6 @@
 import static org.apache.kafka.server.share.fetch.ShareFetchTestUtils.createShareAcquiredRecords;
 import static org.apache.kafka.server.share.fetch.ShareFetchTestUtils.memoryRecordsBuilder;
 import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyBoolean;
@@ -329,7 +328,7 @@ public void testProcessFetchResponseWhenNoRecordsAreAcquired() {
         assertEquals(1, resultData.size());
         assertTrue(resultData.containsKey(tp0));
         assertEquals(0, resultData.get(tp0).partitionIndex());
-        assertNull(resultData.get(tp0).records());
+        assertEquals(MemoryRecords.EMPTY, resultData.get(tp0).records());
         assertTrue(resultData.get(tp0).acquiredRecords().isEmpty());
         assertEquals(Errors.NONE.code(), resultData.get(tp0).errorCode());
 
@@ -344,7 +343,7 @@ public void testProcessFetchResponseWhenNoRecordsAreAcquired() {
         assertEquals(1, resultData.size());
         assertTrue(resultData.containsKey(tp0));
         assertEquals(0, resultData.get(tp0).partitionIndex());
-        assertNull(resultData.get(tp0).records());
+        assertEquals(MemoryRecords.EMPTY, resultData.get(tp0).records());
         assertTrue(resultData.get(tp0).acquiredRecords().isEmpty());
         assertEquals(Errors.NONE.code(), resultData.get(tp0).errorCode());
 

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala

@@ -6199,7 +6199,7 @@ class KafkaApisTest extends Logging {
     assertEquals(partitionIndex, topicResponses.get(0).partitions.get(0).partitionIndex)
     assertEquals(Errors.NONE.code, topicResponses.get(0).partitions.get(0).errorCode)
     assertEquals(Errors.NONE.code, topicResponses.get(0).partitions.get(0).acknowledgeErrorCode)
-    assertNull(topicResponses.get(0).partitions.get(0).records)
+    assertEquals(MemoryRecords.EMPTY, topicResponses.get(0).partitions.get(0).records)
     assertEquals(0, topicResponses.get(0).partitions.get(0).acquiredRecords.toArray().length)
   }
 

server/src/main/java/org/apache/kafka/server/share/fetch/ShareFetch.java

@@ -20,6 +20,7 @@
 import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.message.ShareFetchResponseData.PartitionData;
 import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.record.MemoryRecords;
 import org.apache.kafka.server.storage.log.FetchParams;
 import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
 
@@ -217,7 +218,8 @@ private synchronized void addErroneousToResponse(Map<TopicIdPartition, Partition
                 response.put(topicIdPartition, new PartitionData()
                     .setPartitionIndex(topicIdPartition.partition())
                     .setErrorCode(Errors.forException(throwable).code())
-                    .setErrorMessage(throwable.getMessage()));
+                    .setErrorMessage(throwable.getMessage())
+                    .setRecords(MemoryRecords.EMPTY));
             });
             erroneousTopics.forEach(topic -> {
                 brokerTopicStats.allTopicsStats().failedShareFetchRequestRate().mark();