kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ij...@apache.org
Subject kafka git commit: KAFKA-6185; Remove channels from explictlyMutedChannels set when closed
Date Thu, 09 Nov 2017 16:08:49 GMT
Repository: kafka
Updated Branches:
  refs/heads/1.0 13f036dd2 -> 2283750e8


KAFKA-6185; Remove channels from explictlyMutedChannels set when closed

This memory leak could eventually lead to an OutOfMemoryError. This
was particularly likely in case of down conversions as the leaked
channels would hold on to the record batch (which is only loaded
into the heap in case of down conversions).

Author: Rajini Sivaram <rajinisivaram@googlemail.com>

Reviewers: Ismael Juma <ismael@juma.me.uk>

Closes #4193 from rajinisivaram/KAFKA-6185-oom

(cherry picked from commit 0653a895f51492a8687aaba0dc7a43b21fca2a56)
Signed-off-by: Ismael Juma <ismael@juma.me.uk>


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/2283750e
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/2283750e
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/2283750e

Branch: refs/heads/1.0
Commit: 2283750e8576666605b4f1ec84b666610fa4f5f4
Parents: 13f036d
Author: Rajini Sivaram <rajinisivaram@googlemail.com>
Authored: Thu Nov 9 16:02:04 2017 +0000
Committer: Ismael Juma <ismael@juma.me.uk>
Committed: Thu Nov 9 16:06:29 2017 +0000

----------------------------------------------------------------------
 .../apache/kafka/common/network/Selector.java   |  1 +
 .../kafka/common/network/SelectorTest.java      | 28 ++++++--
 .../unit/kafka/server/BaseRequestTest.scala     | 28 ++++++--
 .../unit/kafka/server/FetchRequestTest.scala    | 71 +++++++++++++++++++-
 .../server/SaslApiVersionsRequestTest.scala     |  4 +-
 5 files changed, 117 insertions(+), 15 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/2283750e/clients/src/main/java/org/apache/kafka/common/network/Selector.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/Selector.java b/clients/src/main/java/org/apache/kafka/common/network/Selector.java
index f70afe0..09df7b5 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/Selector.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/Selector.java
@@ -720,6 +720,7 @@ public class Selector implements Selectable, AutoCloseable {
         }
         this.sensors.connectionClosed.record();
         this.stagedReceives.remove(channel);
+        this.explicitlyMutedChannels.remove(channel);
         if (notifyDisconnect)
             this.disconnected.put(channel.id(), channel.state());
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/2283750e/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java b/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java
index e3d1831..36b847b 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java
@@ -34,6 +34,7 @@ import org.junit.Test;
 import java.io.ByteArrayOutputStream;
 import java.io.DataOutputStream;
 import java.io.IOException;
+import java.lang.reflect.Field;
 import java.net.InetSocketAddress;
 import java.net.ServerSocket;
 import java.nio.ByteBuffer;
@@ -85,9 +86,13 @@ public class SelectorTest {
 
     @After
     public void tearDown() throws Exception {
-        this.selector.close();
-        this.server.close();
-        this.metrics.close();
+        try {
+            verifySelectorEmpty();
+        } finally {
+            this.selector.close();
+            this.server.close();
+            this.metrics.close();
+        }
     }
 
     public SecurityProtocol securityProtocol() {
@@ -527,7 +532,6 @@ public class SelectorTest {
         control.verify();
     }
 
-
     private String blockingRequest(String node, String s) throws IOException {
         selector.send(createSend(node, s));
         selector.poll(1000L);
@@ -585,5 +589,21 @@ public class SelectorTest {
         }
     }
 
+    private void verifySelectorEmpty() throws Exception {
+        for (KafkaChannel channel : selector.channels())
+            selector.close(channel.id());
+        selector.poll(0);
+        selector.poll(0); // Poll a second time to clear everything
+        for (Field field : selector.getClass().getDeclaredFields()) {
+            field.setAccessible(true);
+            Object obj = field.get(selector);
+            if (obj instanceof Set)
+                assertTrue("Field not empty: " + field + " " + obj, ((Set<?>) obj).isEmpty());
+            else if (obj instanceof Map)
+                assertTrue("Field not empty: " + field + " " + obj, ((Map<?, ?>) obj).isEmpty());
+            else if (obj instanceof List)
+                assertTrue("Field not empty: " + field + " " + obj, ((List<?>) obj).isEmpty());
+        }
+    }
 
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/2283750e/core/src/test/scala/unit/kafka/server/BaseRequestTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/BaseRequestTest.scala b/core/src/test/scala/unit/kafka/server/BaseRequestTest.scala
index b34b5a0..f91afd4 100644
--- a/core/src/test/scala/unit/kafka/server/BaseRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/BaseRequestTest.scala
@@ -110,7 +110,7 @@ abstract class BaseRequestTest extends KafkaServerTestHarness {
                      apiVersion: Option[Short] = None,
                      protocol: SecurityProtocol = SecurityProtocol.PLAINTEXT): ByteBuffer
= {
     val socket = connect(destination, protocol)
-    try send(request, apiKey, socket, apiVersion)
+    try sendAndReceive(request, apiKey, socket, apiVersion)
     finally socket.close()
   }
 
@@ -123,18 +123,34 @@ abstract class BaseRequestTest extends KafkaServerTestHarness {
                            destination: SocketServer = anySocketServer,
                            protocol: SecurityProtocol = SecurityProtocol.PLAINTEXT): ByteBuffer
= {
     val socket = connect(destination, protocol)
-    try sendStruct(requestStruct, apiKey, socket, apiVersion)
+    try sendStructAndReceive(requestStruct, apiKey, socket, apiVersion)
     finally socket.close()
   }
 
   /**
     * Serializes and sends the request to the given api.
-    * A ByteBuffer containing the response is returned.
     */
-  def send(request: AbstractRequest, apiKey: ApiKeys, socket: Socket, apiVersion: Option[Short]
= None): ByteBuffer = {
+  def send(request: AbstractRequest, apiKey: ApiKeys, socket: Socket, apiVersion: Option[Short]
= None): Unit = {
     val header = nextRequestHeader(apiKey, apiVersion.getOrElse(request.version))
     val serializedBytes = request.serialize(header).array
-    val response = requestAndReceive(socket, serializedBytes)
+    sendRequest(socket, serializedBytes)
+  }
+
+  /**
+   * Receive response and return a ByteBuffer containing response without the header
+   */
+  def receive(socket: Socket): ByteBuffer = {
+    val response = receiveResponse(socket)
+    skipResponseHeader(response)
+  }
+
+  /**
+    * Serializes and sends the request to the given api.
+    * A ByteBuffer containing the response is returned.
+    */
+  def sendAndReceive(request: AbstractRequest, apiKey: ApiKeys, socket: Socket, apiVersion:
Option[Short] = None): ByteBuffer = {
+    send(request, apiKey, socket, apiVersion)
+    val response = receiveResponse(socket)
     skipResponseHeader(response)
   }
 
@@ -142,7 +158,7 @@ abstract class BaseRequestTest extends KafkaServerTestHarness {
     * Serializes and sends the requestStruct to the given api.
     * A ByteBuffer containing the response (without the response header) is returned.
     */
-  def sendStruct(requestStruct: Struct, apiKey: ApiKeys, socket: Socket, apiVersion: Short):
ByteBuffer = {
+  def sendStructAndReceive(requestStruct: Struct, apiKey: ApiKeys, socket: Socket, apiVersion:
Short): ByteBuffer = {
     val header = nextRequestHeader(apiKey, apiVersion)
     val serializedBytes = AbstractRequestResponse.serialize(header.toStruct, requestStruct).array
     val response = requestAndReceive(socket, serializedBytes)

http://git-wip-us.apache.org/repos/asf/kafka/blob/2283750e/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala b/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala
index 329772b..726757d 100644
--- a/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala
@@ -16,6 +16,7 @@
   */
 package kafka.server
 
+import java.io.DataInputStream
 import java.util
 import java.util.Properties
 
@@ -23,12 +24,12 @@ import kafka.api.KAFKA_0_11_0_IV2
 import kafka.log.LogConfig
 import kafka.utils.TestUtils
 import kafka.utils.TestUtils._
-import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
+import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
 import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.common.protocol.{ApiKeys, Errors}
 import org.apache.kafka.common.record.{Record, RecordBatch}
 import org.apache.kafka.common.requests.{FetchRequest, FetchResponse}
-import org.apache.kafka.common.serialization.StringSerializer
+import org.apache.kafka.common.serialization.{ByteArraySerializer, StringSerializer}
 import org.junit.Assert._
 import org.junit.Test
 
@@ -44,7 +45,8 @@ class FetchRequestTest extends BaseRequestTest {
   private var producer: KafkaProducer[String, String] = null
 
   override def tearDown() {
-    producer.close()
+    if (producer != null)
+      producer.close()
     super.tearDown()
   }
 
@@ -169,6 +171,69 @@ class FetchRequestTest extends BaseRequestTest {
   }
 
   /**
+   * Tests that down-conversions dont leak memory. Large down conversions are triggered
+   * in the server. The client closes its connection after reading partial data when the
+   * channel is muted in the server. If buffers are not released this will result in OOM.
+   */
+  @Test
+  def testDownConversionWithConnectionFailure(): Unit = {
+    val (topicPartition, leaderId) = createTopics(numTopics = 1, numPartitions = 1).head
+
+    val msgValueLen = 100 * 1000
+    val batchSize = 4 * msgValueLen
+    val propsOverride = new Properties
+    propsOverride.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize.toString)
+    val producer = TestUtils.createNewProducer(TestUtils.getBrokerListStrFromServers(servers),
+      retries = 5, lingerMs = Long.MaxValue,
+      keySerializer = new StringSerializer, valueSerializer = new ByteArraySerializer, props
= Some(propsOverride))
+    val bytes = new Array[Byte](msgValueLen)
+    val futures = try {
+      (0 to 1000).map { _ =>
+        producer.send(new ProducerRecord(topicPartition.topic, topicPartition.partition,
"key", bytes))
+      }
+    } finally {
+      producer.close()
+    }
+    // Check futures to ensure sends succeeded, but do this after close since the last
+    // batch is not complete, but sent when the producer is closed
+    futures.foreach(_.get)
+
+    def fetch(version: Short, maxPartitionBytes: Int, closeAfterPartialResponse: Boolean):
Option[FetchResponse] = {
+      val fetchRequest = FetchRequest.Builder.forConsumer(Int.MaxValue, 0, createPartitionMap(maxPartitionBytes,
+        Seq(topicPartition))).build(version)
+
+      val socket = connect(brokerSocketServer(leaderId))
+      try {
+        send(fetchRequest, ApiKeys.FETCH, socket)
+        if (closeAfterPartialResponse) {
+          // read some data to ensure broker has muted this channel and then close socket
+          val size = new DataInputStream(socket.getInputStream).readInt()
+          // Check that we have received almost `maxPartitionBytes` (minus a tolerance) since
in
+          // the case of OOM, the size will be significantly smaller. We can't check for
exactly
+          // maxPartitionBytes since we use approx message sizes that include only the message
value.
+          assertTrue(s"Fetch size too small $size, broker may have run out of memory",
+              size > maxPartitionBytes - batchSize)
+          None
+        } else {
+          Some(FetchResponse.parse(receive(socket), version))
+        }
+      } finally {
+        socket.close()
+      }
+    }
+
+    val version = 1.toShort
+    (0 to 15).foreach(_ => fetch(version, maxPartitionBytes = msgValueLen * 1000, closeAfterPartialResponse
= true))
+
+    val response = fetch(version, maxPartitionBytes = batchSize, closeAfterPartialResponse
= false)
+    val fetchResponse = response.getOrElse(throw new IllegalStateException("No fetch response"))
+    val partitionData = fetchResponse.responseData.get(topicPartition)
+    assertEquals(Errors.NONE, partitionData.error)
+    val batches = partitionData.records.batches.asScala.toBuffer
+    assertEquals(3, batches.size) // size is 3 (not 4) since maxPartitionBytes=msgValueSize*4,
excluding key and headers
+  }
+
+  /**
     * Ensure that we respect the fetch offset when returning records that were converted
from an uncompressed v2
     * record batch to multiple v0/v1 record batches with size 1. If the fetch offset points
to inside the record batch,
     * some records have to be dropped during the conversion.

http://git-wip-us.apache.org/repos/asf/kafka/blob/2283750e/core/src/test/scala/unit/kafka/server/SaslApiVersionsRequestTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/SaslApiVersionsRequestTest.scala b/core/src/test/scala/unit/kafka/server/SaslApiVersionsRequestTest.scala
index 7703993..00b9934 100644
--- a/core/src/test/scala/unit/kafka/server/SaslApiVersionsRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/SaslApiVersionsRequestTest.scala
@@ -90,13 +90,13 @@ class SaslApiVersionsRequestTest extends BaseRequestTest with SaslSetup
{
 
   private def sendApiVersionsRequest(socket: Socket, request: ApiVersionsRequest,
                                      apiVersion: Option[Short] = None): ApiVersionsResponse
= {
-    val response = send(request, ApiKeys.API_VERSIONS, socket, apiVersion)
+    val response = sendAndReceive(request, ApiKeys.API_VERSIONS, socket, apiVersion)
     ApiVersionsResponse.parse(response, request.version)
   }
 
   private def sendSaslHandshakeRequestValidateResponse(socket: Socket) {
     val request = new SaslHandshakeRequest("PLAIN")
-    val response = send(request, ApiKeys.SASL_HANDSHAKE, socket)
+    val response = sendAndReceive(request, ApiKeys.SASL_HANDSHAKE, socket)
     val handshakeResponse = SaslHandshakeResponse.parse(response, request.version)
     assertEquals(Errors.NONE, handshakeResponse.error)
     assertEquals(Collections.singletonList("PLAIN"), handshakeResponse.enabledMechanisms)


Mime
View raw message