MINOR: remove FetchResponse.AbortedTransaction and redundant construc…

clients/src/main/java/org/apache/kafka/clients/FetchSessionHandler.java

@@ -318,7 +318,7 @@ static Set<TopicPartition> findMissing(Set<TopicPartition> toFind, Set<TopicPart
      * @param response  The response.
      * @return          True if the full fetch response partitions are valid.
      */
-    String verifyFullFetchResponsePartitions(FetchResponse<?> response) {
+    String verifyFullFetchResponsePartitions(FetchResponse response) {
         StringBuilder bld = new StringBuilder();
         Set<TopicPartition> extra =
             findMissing(response.responseData().keySet(), sessionPartitions.keySet());
@@ -343,7 +343,7 @@ String verifyFullFetchResponsePartitions(FetchResponse<?> response) {
      * @param response  The response.
      * @return          True if the incremental fetch response partitions are valid.
      */
-    String verifyIncrementalFetchResponsePartitions(FetchResponse<?> response) {
+    String verifyIncrementalFetchResponsePartitions(FetchResponse response) {
         Set<TopicPartition> extra =
             findMissing(response.responseData().keySet(), sessionPartitions.keySet());
         if (!extra.isEmpty()) {
@@ -362,7 +362,7 @@ String verifyIncrementalFetchResponsePartitions(FetchResponse<?> response) {
      * @param response  The FetchResponse.
      * @return          The string to log.
      */
-    private String responseDataToLogString(FetchResponse<?> response) {
+    private String responseDataToLogString(FetchResponse response) {
         if (!log.isTraceEnabled()) {
             int implied = sessionPartitions.size() - response.responseData().size();
             if (implied > 0) {
@@ -398,7 +398,7 @@ private String responseDataToLogString(FetchResponse<?> response) {
      * @return          True if the response is well-formed; false if it can't be processed
      *                  because of missing or unexpected partitions.
      */
-    public boolean handleResponse(FetchResponse<?> response) {
+    public boolean handleResponse(FetchResponse response) {
         if (response.error() != Errors.NONE) {
             log.info("Node {} was unable to process the fetch request with {}: {}.",
                 node, nextMetadata, response.error());

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

@@ -47,6 +47,7 @@
 import org.apache.kafka.common.errors.TopicAuthorizationException;
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.message.FetchResponseData;
 import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersion;
 import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition;
 import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartitionResponse;
@@ -66,7 +67,6 @@
 import org.apache.kafka.common.record.ControlRecordType;
 import org.apache.kafka.common.record.Record;
 import org.apache.kafka.common.record.RecordBatch;
-import org.apache.kafka.common.record.Records;
 import org.apache.kafka.common.record.TimestampType;
 import org.apache.kafka.common.requests.FetchRequest;
 import org.apache.kafka.common.requests.FetchResponse;
@@ -276,7 +276,7 @@ public void onSuccess(ClientResponse resp) {
                     synchronized (Fetcher.this) {
                         try {
                             @SuppressWarnings("unchecked")
-                            FetchResponse<Records> response = (FetchResponse<Records>) resp.responseBody();
+                            FetchResponse response = (FetchResponse) resp.responseBody();
                             FetchSessionHandler handler = sessionHandler(fetchTarget.id());
                             if (handler == null) {
                                 log.error("Unable to find FetchSessionHandler for node {}. Ignoring fetch response.",
@@ -290,7 +290,7 @@ public void onSuccess(ClientResponse resp) {
                             Set<TopicPartition> partitions = new HashSet<>(response.responseData().keySet());
                             FetchResponseMetricAggregator metricAggregator = new FetchResponseMetricAggregator(sensors, partitions);
 
-                            for (Map.Entry<TopicPartition, FetchResponse.PartitionData<Records>> entry : response.responseData().entrySet()) {
+                            for (Map.Entry<TopicPartition, FetchResponseData.PartitionData> entry : response.responseData().entrySet()) {
                                 TopicPartition partition = entry.getKey();
                                 FetchRequest.PartitionData requestData = data.sessionPartitions().get(partition);
                                 if (requestData == null) {
@@ -309,12 +309,12 @@ public void onSuccess(ClientResponse resp) {
                                     throw new IllegalStateException(message);
                                 } else {
                                     long fetchOffset = requestData.fetchOffset;
-                                    FetchResponse.PartitionData<Records> partitionData = entry.getValue();
+                                    FetchResponseData.PartitionData partitionData = entry.getValue();
 
                                     log.debug("Fetch {} at offset {} for partition {} returned fetch data {}",
                                             isolationLevel, fetchOffset, partition, partitionData);
 
-                                    Iterator<? extends RecordBatch> batches = partitionData.records().batches().iterator();
+                                    Iterator<? extends RecordBatch> batches = FetchResponse.records(partitionData).batches().iterator();
                                     short responseVersion = resp.requestHeader().apiVersion();
 
                                     completedFetches.add(new CompletedFetch(partition, partitionData,
@@ -617,7 +617,7 @@ public FetchedRecords<K, V> fetchedRecords() {
                             // The first condition ensures that the completedFetches is not stuck with the same completedFetch
                             // in cases such as the TopicAuthorizationException, and the second condition ensures that no
                             // potential data loss due to an exception in a following record.
-                            FetchResponse.PartitionData<Records> partition = records.partitionData;
+                            FetchResponseData.PartitionData partition = records.partitionData;
                             if (fetched.isEmpty() && (partition.records() == null || partition.records().sizeInBytes() == 0)) {
                                 completedFetches.poll();
                             }
@@ -1236,10 +1236,10 @@ private <T> Map<Node, Map<TopicPartition, T>> regroupPartitionMapByNode(Map<Topi
      */
     private CompletedFetch initializeCompletedFetch(CompletedFetch nextCompletedFetch) {
         TopicPartition tp = nextCompletedFetch.partition;
-        FetchResponse.PartitionData<Records> partition = nextCompletedFetch.partitionData;
+        FetchResponseData.PartitionData partition = nextCompletedFetch.partitionData;
         long fetchOffset = nextCompletedFetch.nextFetchOffset;
         CompletedFetch completedFetch = null;
-        Errors error = partition.error();
+        Errors error = Errors.forCode(partition.errorCode());
 
         try {
             if (!subscriptions.hasValidPosition(tp)) {
@@ -1257,7 +1257,7 @@ private CompletedFetch initializeCompletedFetch(CompletedFetch nextCompletedFetc
 
                 log.trace("Preparing to read {} bytes of data for partition {} with offset {}",
                         partition.records().sizeInBytes(), tp, position);
-                Iterator<? extends RecordBatch> batches = partition.records().batches().iterator();
+                Iterator<? extends RecordBatch> batches = FetchResponse.records(partition).batches().iterator();
                 completedFetch = nextCompletedFetch;
 
                 if (!batches.hasNext() && partition.records().sizeInBytes() > 0) {
@@ -1293,11 +1293,11 @@ private CompletedFetch initializeCompletedFetch(CompletedFetch nextCompletedFetc
                     subscriptions.updateLastStableOffset(tp, partition.lastStableOffset());
                 }
 
-                if (partition.preferredReadReplica().isPresent()) {
-                    subscriptions.updatePreferredReadReplica(completedFetch.partition, partition.preferredReadReplica().get(), () -> {
+                if (FetchResponse.isPreferredReplica(partition)) {
+                    subscriptions.updatePreferredReadReplica(completedFetch.partition, partition.preferredReadReplica(), () -> {
                         long expireTimeMs = time.milliseconds() + metadata.metadataExpireMs();
                         log.debug("Updating preferred read replica for partition {} to {}, set to expire at {}",
-                                tp, partition.preferredReadReplica().get(), expireTimeMs);
+                                tp, partition.preferredReadReplica(), expireTimeMs);
                         return expireTimeMs;
                     });
                 }
@@ -1462,8 +1462,8 @@ private class CompletedFetch {
         private final TopicPartition partition;
         private final Iterator<? extends RecordBatch> batches;
         private final Set<Long> abortedProducerIds;
-        private final PriorityQueue<FetchResponse.AbortedTransaction> abortedTransactions;
-        private final FetchResponse.PartitionData<Records> partitionData;
+        private final PriorityQueue<FetchResponseData.AbortedTransaction> abortedTransactions;
+        private final FetchResponseData.PartitionData partitionData;
         private final FetchResponseMetricAggregator metricAggregator;
         private final short responseVersion;
         private final long receivedTimestamp;
@@ -1481,7 +1481,7 @@ private class CompletedFetch {
         private boolean initialized = false;
 
         private CompletedFetch(TopicPartition partition,
-                               FetchResponse.PartitionData<Records> partitionData,
+                               FetchResponseData.PartitionData partitionData,
                                FetchResponseMetricAggregator metricAggregator,
                                Iterator<? extends RecordBatch> batches,
                                Long fetchOffset,
@@ -1651,22 +1651,22 @@ private void consumeAbortedTransactionsUpTo(long offset) {
             if (abortedTransactions == null)
                 return;
 
-            while (!abortedTransactions.isEmpty() && abortedTransactions.peek().firstOffset <= offset) {
-                FetchResponse.AbortedTransaction abortedTransaction = abortedTransactions.poll();
-                abortedProducerIds.add(abortedTransaction.producerId);
+            while (!abortedTransactions.isEmpty() && abortedTransactions.peek().firstOffset() <= offset) {
+                FetchResponseData.AbortedTransaction abortedTransaction = abortedTransactions.poll();
+                abortedProducerIds.add(abortedTransaction.producerId());
             }
         }
 
         private boolean isBatchAborted(RecordBatch batch) {
             return batch.isTransactional() && abortedProducerIds.contains(batch.producerId());
         }
 
-        private PriorityQueue<FetchResponse.AbortedTransaction> abortedTransactions(FetchResponse.PartitionData<?> partition) {
+        private PriorityQueue<FetchResponseData.AbortedTransaction> abortedTransactions(FetchResponseData.PartitionData partition) {
             if (partition.abortedTransactions() == null || partition.abortedTransactions().isEmpty())
                 return null;
 
-            PriorityQueue<FetchResponse.AbortedTransaction> abortedTransactions = new PriorityQueue<>(
-                    partition.abortedTransactions().size(), Comparator.comparingLong(o -> o.firstOffset)
+            PriorityQueue<FetchResponseData.AbortedTransaction> abortedTransactions = new PriorityQueue<>(
+                    partition.abortedTransactions().size(), Comparator.comparingLong(FetchResponseData.AbortedTransaction::firstOffset)
             );
             abortedTransactions.addAll(partition.abortedTransactions());
             return abortedTransactions;

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

@@ -19,10 +19,10 @@
 import org.apache.kafka.common.IsolationLevel;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.message.FetchRequestData;
+import org.apache.kafka.common.message.FetchResponseData;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.ByteBufferAccessor;
 import org.apache.kafka.common.protocol.Errors;
-import org.apache.kafka.common.record.MemoryRecords;
 import org.apache.kafka.common.record.RecordBatch;
 import org.apache.kafka.common.utils.Utils;
 
@@ -296,14 +296,11 @@ public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
         // may not be any partitions at all in the response.  For this reason, the top-level error code
         // is essential for them.
         Errors error = Errors.forException(e);
-        LinkedHashMap<TopicPartition, FetchResponse.PartitionData<MemoryRecords>> responseData = new LinkedHashMap<>();
+        LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseData = new LinkedHashMap<>();
         for (Map.Entry<TopicPartition, PartitionData> entry : fetchData.entrySet()) {
-            FetchResponse.PartitionData<MemoryRecords> partitionResponse = new FetchResponse.PartitionData<>(error,
-                FetchResponse.INVALID_HIGHWATERMARK, FetchResponse.INVALID_LAST_STABLE_OFFSET,
-                FetchResponse.INVALID_LOG_START_OFFSET, Optional.empty(), null, MemoryRecords.EMPTY);
-            responseData.put(entry.getKey(), partitionResponse);
+            responseData.put(entry.getKey(), FetchResponse.partitionResponse(entry.getKey().partition(), error));
         }
-        return new FetchResponse<>(error, responseData, throttleTimeMs, data.sessionId());
+        return FetchResponse.of(error, throttleTimeMs, data.sessionId(), responseData);
     }
 
     public int replicaId() {