frankvicky commented on code in PR #19167:
URL: https://github.com/apache/kafka/pull/19167#discussion_r2020097506
##########
clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java:
##########
@@ -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);
+ }
Review Comment:
Given that leveraging the existing `of` methods can be tricky for some use
cases, I added a new `of` method.
For example:
https://github.com/apache/kafka/blob/4a5ae144eaaff1607c04dcb35ffa1ebbfeb019c7/core/src/main/scala/kafka/server/ControllerApis.scala#L187-L190
If we use the existing `of`, we must do much extra work to match the
arguments, which will introduce overheads.
```scala
handleRaftRequest(request, response => {
val data = response.asInstanceOf[FetchResponseData]
val tpToPartition = new util.LinkedHashMap[TopicIdPartition,
FetchResponseData.PartitionData]()
data.responses().forEach(res => res.partitions().forEach(part =>
tpToPartition.put(
new TopicIdPartition(res.topicId(), new TopicPartition(res.topic(),
part.partitionIndex())),
part
)))
FetchResponse.of(Errors.forCode(data.errorCode()),
data.throttleTimeMs(), data.sessionId(), tpToPartition)
})
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
