KAFKA-10697: Remove ProduceResponse.responses

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

@@ -24,7 +24,6 @@
 import org.apache.kafka.common.record.RecordBatch;
 
 import java.nio.ByteBuffer;
-import java.util.AbstractMap;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -112,31 +111,6 @@ public ProduceResponseData data() {
         return this.data;
     }
 
-    /**
-     * this method is used by testing only.
-     * refactor the tests which are using this method and then remove this method from production code.
-     * https://issues.apache.org/jira/browse/KAFKA-10697
-     */
-    @Deprecated
-    public Map<TopicPartition, PartitionResponse> responses() {
-        return data.responses()
-                .stream()
-                .flatMap(t -> t.partitionResponses()
-                        .stream()
-                        .map(p -> new AbstractMap.SimpleEntry<>(new TopicPartition(t.name(), p.index()),
-                                new PartitionResponse(
-                                        Errors.forCode(p.errorCode()),
-                                        p.baseOffset(),
-                                        p.logAppendTimeMs(),
-                                        p.logStartOffset(),
-                                        p.recordErrors()
-                                                .stream()
-                                                .map(e -> new RecordError(e.batchIndex(), e.batchIndexErrorMessage()))
-                                                .collect(Collectors.toList()),
-                                        p.errorMessage()))))
-                .collect(Collectors.toMap(AbstractMap.SimpleEntry::getKey, AbstractMap.SimpleEntry::getValue));
-    }
-
     @Override
     public int throttleTimeMs() {
         return this.data.throttleTimeMs();

clients/src/test/java/org/apache/kafka/common/requests/ProduceResponseTest.java

@@ -17,15 +17,19 @@
 
 package org.apache.kafka.common.requests;
 
+import org.apache.kafka.common.message.ProduceResponseData;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.record.RecordBatch;
+
 import org.junit.jupiter.api.Test;
 
 import java.nio.ByteBuffer;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 import static org.apache.kafka.common.protocol.ApiKeys.PRODUCE;
@@ -40,42 +44,56 @@ public class ProduceResponseTest {
     public void produceResponseV5Test() {
         Map<TopicPartition, ProduceResponse.PartitionResponse> responseData = new HashMap<>();
         TopicPartition tp0 = new TopicPartition("test", 0);
-        responseData.put(tp0, new ProduceResponse.PartitionResponse(Errors.NONE,
-                10000, RecordBatch.NO_TIMESTAMP, 100));
+        responseData.put(tp0, new ProduceResponse.PartitionResponse(Errors.NONE, 10000, RecordBatch.NO_TIMESTAMP, 100));
 
         ProduceResponse v5Response = new ProduceResponse(responseData, 10);
         short version = 5;
 
         ByteBuffer buffer = RequestTestUtils.serializeResponseWithHeader(v5Response, version, 0);
 
         ResponseHeader.parse(buffer, ApiKeys.PRODUCE.responseHeaderVersion(version)); // throw away.
-        ProduceResponse v5FromBytes = (ProduceResponse) AbstractResponse.parseResponse(ApiKeys.PRODUCE,
-                buffer, version);
-
-        assertEquals(1, v5FromBytes.responses().size());
-        assertTrue(v5FromBytes.responses().containsKey(tp0));
-        ProduceResponse.PartitionResponse partitionResponse = v5FromBytes.responses().get(tp0);
-        assertEquals(100, partitionResponse.logStartOffset);
-        assertEquals(10000, partitionResponse.baseOffset);
-        assertEquals(10, v5FromBytes.throttleTimeMs());
-        assertEquals(responseData, v5Response.responses());
+        ProduceResponse v5FromBytes = (ProduceResponse) AbstractResponse.parseResponse(ApiKeys.PRODUCE, buffer, version);
+
+        assertEquals(1, v5FromBytes.data().responses().size());
+        ProduceResponseData.TopicProduceResponse topicProduceResponse = v5FromBytes.data().responses().iterator().next();
+        assertEquals(1, topicProduceResponse.partitionResponses().size());  
+        ProduceResponseData.PartitionProduceResponse partitionProduceResponse = topicProduceResponse.partitionResponses().iterator().next();
+        TopicPartition tp = new TopicPartition(topicProduceResponse.name(), partitionProduceResponse.index());
+        assertEquals(tp0, tp);
+
+        assertEquals(100, partitionProduceResponse.logStartOffset());
+        assertEquals(10000, partitionProduceResponse.baseOffset());
+        assertEquals(RecordBatch.NO_TIMESTAMP, partitionProduceResponse.logAppendTimeMs());
+        assertEquals(Errors.NONE, Errors.forCode(partitionProduceResponse.errorCode()));
+        assertNull(partitionProduceResponse.errorMessage());
+        assertTrue(partitionProduceResponse.recordErrors().isEmpty());
     }
 
     @SuppressWarnings("deprecation")
     @Test
     public void produceResponseVersionTest() {
         Map<TopicPartition, ProduceResponse.PartitionResponse> responseData = new HashMap<>();
-        responseData.put(new TopicPartition("test", 0), new ProduceResponse.PartitionResponse(Errors.NONE,
-                10000, RecordBatch.NO_TIMESTAMP, 100));
+        responseData.put(new TopicPartition("test", 0), new ProduceResponse.PartitionResponse(Errors.NONE, 10000, RecordBatch.NO_TIMESTAMP, 100));
         ProduceResponse v0Response = new ProduceResponse(responseData);
         ProduceResponse v1Response = new ProduceResponse(responseData, 10);
         ProduceResponse v2Response = new ProduceResponse(responseData, 10);
         assertEquals(0, v0Response.throttleTimeMs(), "Throttle time must be zero");
         assertEquals(10, v1Response.throttleTimeMs(), "Throttle time must be 10");
         assertEquals(10, v2Response.throttleTimeMs(), "Throttle time must be 10");
-        assertEquals(responseData, v0Response.responses(), "Response data does not match");
-        assertEquals(responseData, v1Response.responses(), "Response data does not match");
-        assertEquals(responseData, v2Response.responses(), "Response data does not match");
+
+        List<ProduceResponse> arrResponse = Arrays.asList(v0Response, v1Response, v2Response);
+        for(ProduceResponse produceResponse:arrResponse) {
+            assertEquals(1, produceResponse.data().responses().size());
+            ProduceResponseData.TopicProduceResponse topicProduceResponse = produceResponse.data().responses().iterator().next();
+            assertEquals(1, topicProduceResponse.partitionResponses().size());  
+            ProduceResponseData.PartitionProduceResponse partitionProduceResponse = topicProduceResponse.partitionResponses().iterator().next();
+            assertEquals(100, partitionProduceResponse.logStartOffset());
+            assertEquals(10000, partitionProduceResponse.baseOffset());
+            assertEquals(RecordBatch.NO_TIMESTAMP, partitionProduceResponse.logAppendTimeMs());
+            assertEquals(Errors.NONE, Errors.forCode(partitionProduceResponse.errorCode()));
+            assertNull(partitionProduceResponse.errorMessage());
+            assertTrue(partitionProduceResponse.recordErrors().isEmpty());
+        }
     }
 
     @SuppressWarnings("deprecation")
@@ -91,15 +109,18 @@ public void produceResponseRecordErrorsTest() {
 
         for (short version : PRODUCE.allVersions()) {
             ProduceResponse response = new ProduceResponse(responseData);
-            ProduceResponse.PartitionResponse deserialized = ProduceResponse.parse(response.serialize(version), version).responses().get(tp);
+
+            ProduceResponse produceResponse = ProduceResponse.parse(response.serialize(version), version);
+            ProduceResponseData.TopicProduceResponse topicProduceResponse = produceResponse.data().responses().iterator().next();
+            ProduceResponseData.PartitionProduceResponse deserialized = topicProduceResponse.partitionResponses().iterator().next();
             if (version >= 8) {
-                assertEquals(1, deserialized.recordErrors.size());
-                assertEquals(3, deserialized.recordErrors.get(0).batchIndex);
-                assertEquals("Record error", deserialized.recordErrors.get(0).message);
-                assertEquals("Produce failed", deserialized.errorMessage);
+                assertEquals(1, deserialized.recordErrors().size());
+                assertEquals(3, deserialized.recordErrors().get(0).batchIndex());
+                assertEquals("Record error", deserialized.recordErrors().get(0).batchIndexErrorMessage());
+                assertEquals("Produce failed", deserialized.errorMessage());
             } else {
-                assertEquals(0, deserialized.recordErrors.size());
-                assertNull(deserialized.errorMessage);
+                assertEquals(0, deserialized.recordErrors().size());
+                assertNull(deserialized.errorMessage());
             }
         }
     }

clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java

@@ -158,6 +158,7 @@
 import org.apache.kafka.common.message.OffsetForLeaderEpochResponseData.EpochEndOffset;
 import org.apache.kafka.common.message.OffsetForLeaderEpochResponseData.OffsetForLeaderTopicResult;
 import org.apache.kafka.common.message.ProduceRequestData;
+import org.apache.kafka.common.message.ProduceResponseData;
 import org.apache.kafka.common.message.RenewDelegationTokenRequestData;
 import org.apache.kafka.common.message.RenewDelegationTokenResponseData;
 import org.apache.kafka.common.message.SaslAuthenticateRequestData;
@@ -210,7 +211,6 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.LinkedList;
 import java.util.List;
@@ -784,24 +784,25 @@ public void produceRequestToStringTest() {
     @Test
     public void produceRequestGetErrorResponseTest() {
         ProduceRequest request = createProduceRequest(ApiKeys.PRODUCE.latestVersion());
-        Set<TopicPartition> partitions = new HashSet<>(request.partitionSizes().keySet());
 
         ProduceResponse errorResponse = (ProduceResponse) request.getErrorResponse(new NotEnoughReplicasException());
-        assertEquals(partitions, errorResponse.responses().keySet());
-        ProduceResponse.PartitionResponse partitionResponse = errorResponse.responses().values().iterator().next();
-        assertEquals(Errors.NOT_ENOUGH_REPLICAS, partitionResponse.error);
-        assertEquals(ProduceResponse.INVALID_OFFSET, partitionResponse.baseOffset);
-        assertEquals(RecordBatch.NO_TIMESTAMP, partitionResponse.logAppendTime);
+        ProduceResponseData.TopicProduceResponse topicProduceResponse = errorResponse.data().responses().iterator().next();
+        ProduceResponseData.PartitionProduceResponse partitionProduceResponse = topicProduceResponse.partitionResponses().iterator().next();
+
+        assertEquals(Errors.NOT_ENOUGH_REPLICAS, Errors.forCode(partitionProduceResponse.errorCode()));
+        assertEquals(ProduceResponse.INVALID_OFFSET, partitionProduceResponse.baseOffset());
+        assertEquals(RecordBatch.NO_TIMESTAMP, partitionProduceResponse.logAppendTimeMs());
 
         request.clearPartitionRecords();
 
         // `getErrorResponse` should behave the same after `clearPartitionRecords`
         errorResponse = (ProduceResponse) request.getErrorResponse(new NotEnoughReplicasException());
-        assertEquals(partitions, errorResponse.responses().keySet());
-        partitionResponse = errorResponse.responses().values().iterator().next();
-        assertEquals(Errors.NOT_ENOUGH_REPLICAS, partitionResponse.error);
-        assertEquals(ProduceResponse.INVALID_OFFSET, partitionResponse.baseOffset);
-        assertEquals(RecordBatch.NO_TIMESTAMP, partitionResponse.logAppendTime);
+        topicProduceResponse = errorResponse.data().responses().iterator().next();
+        partitionProduceResponse = topicProduceResponse.partitionResponses().iterator().next();
+
+        assertEquals(Errors.NOT_ENOUGH_REPLICAS, Errors.forCode(partitionProduceResponse.errorCode()));
+        assertEquals(ProduceResponse.INVALID_OFFSET, partitionProduceResponse.baseOffset());
+        assertEquals(RecordBatch.NO_TIMESTAMP, partitionProduceResponse.logAppendTimeMs());
     }
 
     @Test

core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala

@@ -155,10 +155,16 @@ class AuthorizerIntegrationTest extends BaseRequestTest {
       classOf[PrincipalBuilder].getName)
   }
 
-  @nowarn("cat=deprecation")
   val requestKeyToError = (topicNames: Map[Uuid, String]) => Map[ApiKeys, Nothing => Errors](
     ApiKeys.METADATA -> ((resp: requests.MetadataResponse) => resp.errors.asScala.find(_._1 == topic).getOrElse(("test", Errors.NONE))._2),
-    ApiKeys.PRODUCE -> ((resp: requests.ProduceResponse) => resp.responses.asScala.find(_._1 == tp).get._2.error),
+    ApiKeys.PRODUCE -> ((resp: requests.ProduceResponse) => {
+      Errors.forCode(
+        resp.data
+          .responses.find(topic)
+          .partitionResponses.asScala.find(_.index == part).get
+          .errorCode
+      )      
+    }),
     ApiKeys.FETCH -> ((resp: requests.FetchResponse) => Errors.forCode(resp.responseData.asScala.find(_._1 == tp).get._2.errorCode)),
     ApiKeys.LIST_OFFSETS -> ((resp: ListOffsetsResponse) => {
       Errors.forCode(

core/src/test/scala/integration/kafka/network/DynamicConnectionQuotaTest.scala

@@ -37,7 +37,6 @@ import org.apache.kafka.common.{KafkaException, requests}
 import org.junit.jupiter.api.Assertions._
 import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
 
-import scala.annotation.nowarn
 import scala.jdk.CollectionConverters._
 
 class DynamicConnectionQuotaTest extends BaseRequestTest {
@@ -337,12 +336,13 @@ class DynamicConnectionQuotaTest extends BaseRequestTest {
     }
   }
 
-  @nowarn("cat=deprecation")
   private def verifyConnection(socket: Socket): Unit = {
     val produceResponse = sendAndReceive[ProduceResponse](produceRequest, socket)
-    assertEquals(1, produceResponse.responses.size)
-    val (_, partitionResponse) = produceResponse.responses.asScala.head
-    assertEquals(Errors.NONE, partitionResponse.error)
+    assertEquals(1, produceResponse.data.responses.size)
+    val topicProduceResponse = produceResponse.data.responses.asScala.head
+    assertEquals(1, topicProduceResponse.partitionResponses.size)    
+    val partitionProduceResponse = topicProduceResponse.partitionResponses.asScala.head
+    assertEquals(Errors.NONE, Errors.forCode(partitionProduceResponse.errorCode))
   }
 
   private def verifyMaxConnections(maxConnections: Int, connectWithFailure: () => Unit): Unit = {

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

@@ -37,7 +37,7 @@ import org.apache.kafka.common.{TopicPartition, requests}
 import org.junit.jupiter.api.Assertions._
 import org.junit.jupiter.api.Test
 
-import scala.annotation.nowarn
+import scala.jdk.CollectionConverters._
 
 class EdgeCaseRequestTest extends KafkaServerTestHarness {
 
@@ -116,7 +116,6 @@ class EdgeCaseRequestTest extends KafkaServerTestHarness {
     }
   }
 
-  @nowarn("cat=deprecation")
   @Test
   def testProduceRequestWithNullClientId(): Unit = {
     val topic = "topic"
@@ -154,11 +153,12 @@ class EdgeCaseRequestTest extends KafkaServerTestHarness {
 
     assertEquals(0, responseBuffer.remaining, "The response should parse completely")
     assertEquals(correlationId, responseHeader.correlationId, "The correlationId should match request")
-    assertEquals(1, produceResponse.responses.size, "One partition response should be returned")
-
-    val partitionResponse = produceResponse.responses.get(topicPartition)
-    assertNotNull(partitionResponse)
-    assertEquals(Errors.NONE, partitionResponse.error, "There should be no error")
+    assertEquals(1, produceResponse.data.responses.size, "One topic response should be returned")
+    val topicProduceResponse = produceResponse.data.responses.asScala.head
+    assertEquals(1, topicProduceResponse.partitionResponses.size, "One partition response should be returned")    
+    val partitionProduceResponse = topicProduceResponse.partitionResponses.asScala.head
+    assertNotNull(partitionProduceResponse)
+    assertEquals(Errors.NONE, Errors.forCode(partitionProduceResponse.errorCode), "There should be no error")
   }
 
   @Test

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

@@ -79,7 +79,6 @@ import org.junit.jupiter.params.ParameterizedTest
 import org.junit.jupiter.params.provider.ValueSource
 import org.mockito.{ArgumentMatchers, Mockito}
 
-import scala.annotation.nowarn
 import scala.collection.{Map, Seq, mutable}
 import scala.jdk.CollectionConverters._
 
@@ -1408,7 +1407,6 @@ class KafkaApisTest {
     }
   }
 
-  @nowarn("cat=deprecation")
   @Test
   def shouldReplaceProducerFencedWithInvalidProducerEpochInProduceResponse(): Unit = {
     val topic = "topic"
@@ -1455,10 +1453,11 @@ class KafkaApisTest {
 
       val response = capturedResponse.getValue.asInstanceOf[ProduceResponse]
 
-      assertEquals(1, response.responses().size())
-      for (partitionResponse <- response.responses().asScala) {
-        assertEquals(Errors.INVALID_PRODUCER_EPOCH, partitionResponse._2.error)
-      }
+      assertEquals(1, response.data.responses.size)
+      val topicProduceResponse = response.data.responses.asScala.head
+      assertEquals(1, topicProduceResponse.partitionResponses.size)   
+      val partitionProduceResponse = topicProduceResponse.partitionResponses.asScala.head
+      assertEquals(Errors.INVALID_PRODUCER_EPOCH, Errors.forCode(partitionProduceResponse.errorCode))
     }
   }