implemented review suggstions

Author: edoardocomar

clients/clients-integration-tests/src/test/java/org/apache/kafka/common/SocketServerMemoryPoolTest.java

@@ -16,18 +16,14 @@
  */
 package org.apache.kafka.common;
 
-import kafka.network.SocketServer;
-
 import org.apache.kafka.common.message.ProduceRequestData;
-import org.apache.kafka.common.network.ListenerName;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.requests.RequestHeader;
 import org.apache.kafka.common.requests.RequestUtils;
 import org.apache.kafka.common.test.ClusterInstance;
 import org.apache.kafka.common.test.api.ClusterConfigProperty;
 import org.apache.kafka.common.test.api.ClusterTest;
 import org.apache.kafka.common.test.api.ClusterTestDefaults;
-import org.apache.kafka.common.test.api.TestKitDefaults;
 import org.apache.kafka.common.test.api.Type;
 import org.apache.kafka.network.SocketServerConfigs;
 import org.apache.kafka.server.IntegrationTestUtils;
@@ -38,6 +34,7 @@
 import java.net.SocketTimeoutException;
 import java.nio.ByteBuffer;
 
+import static org.apache.kafka.common.test.api.TestKitDefaults.BROKER_ID_OFFSET;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -66,6 +63,7 @@ public void testRequestWithCorruptBody(ClusterInstance clusterInstance) throws E
         byte[] rawRequestBytes = buffer.array();
 
         // corrupt body but leave header valid
+        assertTrue(rawRequestBytes.length > header.size());
         for (int i = header.size(); i < rawRequestBytes.length; i++) {
             rawRequestBytes[i] = (byte) 0xFF;
         }
@@ -75,7 +73,7 @@ public void testRequestWithCorruptBody(ClusterInstance clusterInstance) throws E
     private void sendAndAssert(ClusterInstance clusterInstance, byte[] rawRequestBytes) throws Exception {
         long initialMemoryPoolAvailable = getMemoryPoolAvailable(clusterInstance);
 
-        try (Socket socket = IntegrationTestUtils.connect(getBrokerBoundPort(clusterInstance))) {
+        try (Socket socket = IntegrationTestUtils.connect(clusterInstance.brokerBoundPorts().get(BROKER_ID_OFFSET))) {
             socket.setSoTimeout(/* readTimeoutMs */ 5_000);
             IntegrationTestUtils.sendRequest(socket, rawRequestBytes);
             assertTrue(drainUntilClosed(socket.getInputStream()), "expected connection closed");
@@ -86,16 +84,8 @@ private void sendAndAssert(ClusterInstance clusterInstance, byte[] rawRequestByt
         assertEquals(initialMemoryPoolAvailable, finalMemoryPoolAvailable);
     }
 
-    private SocketServer getSocketServer(ClusterInstance clusterInstance) {
-        return clusterInstance.brokers().get(TestKitDefaults.BROKER_ID_OFFSET).socketServer();
-    }
-
-    private int getBrokerBoundPort(ClusterInstance clusterInstance) {
-        return getSocketServer(clusterInstance).boundPort(ListenerName.normalised(TestKitDefaults.DEFAULT_BROKER_LISTENER_NAME));
-    }
-
     private long getMemoryPoolAvailable(ClusterInstance clusterInstance) {
-        return getSocketServer(clusterInstance).memoryPool().availableMemory();
+        return clusterInstance.brokers().get(BROKER_ID_OFFSET).socketServer().memoryPool().availableMemory();
     }
 
     /*