kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject kafka git commit: MINOR: Null out all buffer references in RequestChannel.Request
Date Thu, 03 Aug 2017 21:59:43 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 6bee1e9e5 -> 13deb84dc


MINOR: Null out all buffer references in RequestChannel.Request

The previous code kept two references to `Buffer` and only nulled
out one of them.

As part of this, I removed the `case` modifier from
`RequestChannel.{Request, Response}`. They don't seem to be good
matches given the types of fields they contain (mutable buffers and
opaque `Send` instances).

Also removed a couple of unused files in `kafka.network`.

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

Reviewers: Radai Rosenblatt <radai.rosenblatt@gmail.com>, Jason Gustafson <jason@confluent.io>

Closes #3596 from ijuma/release-buffer-in-request


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

Branch: refs/heads/trunk
Commit: 13deb84dcf4b3e583997dbd05a0e25840f024375
Parents: 6bee1e9
Author: Ismael Juma <ismael@juma.me.uk>
Authored: Thu Aug 3 14:56:20 2017 -0700
Committer: Jason Gustafson <jason@confluent.io>
Committed: Thu Aug 3 14:56:20 2017 -0700

----------------------------------------------------------------------
 .../scala/kafka/network/ConnectionConfig.scala  | 27 ---------------
 core/src/main/scala/kafka/network/Handler.scala | 35 --------------------
 .../scala/kafka/network/RequestChannel.scala    | 31 +++++++++++------
 .../main/scala/kafka/network/SocketServer.scala |  7 ++--
 .../unit/kafka/network/SocketServerTest.scala   |  2 +-
 .../scala/unit/kafka/server/KafkaApisTest.scala |  3 +-
 6 files changed, 26 insertions(+), 79 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/13deb84d/core/src/main/scala/kafka/network/ConnectionConfig.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/network/ConnectionConfig.scala b/core/src/main/scala/kafka/network/ConnectionConfig.scala
deleted file mode 100644
index cde7c09..0000000
--- a/core/src/main/scala/kafka/network/ConnectionConfig.scala
+++ /dev/null
@@ -1,27 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * 
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kafka.network
-
-trait ConnectionConfig {
-  val host: String
-  val port: Int
-  val sendBufferSize: Int = -1
-  val receiveBufferSize: Int = -1
-  val tcpNoDelay = true
-  val keepAlive = false  
-}

http://git-wip-us.apache.org/repos/asf/kafka/blob/13deb84d/core/src/main/scala/kafka/network/Handler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/network/Handler.scala b/core/src/main/scala/kafka/network/Handler.scala
deleted file mode 100644
index 1a7d56e..0000000
--- a/core/src/main/scala/kafka/network/Handler.scala
+++ /dev/null
@@ -1,35 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * 
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package kafka.network
-
-import org.apache.kafka.common.network.{NetworkReceive, Send}
-
-private[kafka] object Handler {
-  
-  /**
-   * A request handler is a function that turns an incoming 
-   * transmission into an outgoing transmission
-   */
-  type Handler = NetworkReceive => Option[Send]
-  
-  /**
-   * A handler mapping finds the right Handler function for a given request
-   */
-  type HandlerMapping = (Short, NetworkReceive) => Handler
-
-}

http://git-wip-us.apache.org/repos/asf/kafka/blob/13deb84d/core/src/main/scala/kafka/network/RequestChannel.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/network/RequestChannel.scala b/core/src/main/scala/kafka/network/RequestChannel.scala
index 7553c10..4ebef7c 100644
--- a/core/src/main/scala/kafka/network/RequestChannel.scala
+++ b/core/src/main/scala/kafka/network/RequestChannel.scala
@@ -41,9 +41,9 @@ import org.apache.log4j.Logger
 import scala.reflect.ClassTag
 
 object RequestChannel extends Logging {
-  val AllDone = Request(processor = 1, connectionId = "2", Session(KafkaPrincipal.ANONYMOUS,
InetAddress.getLocalHost),
-    buffer = shutdownReceive, memoryPool = MemoryPool.NONE, startTimeNanos = 0, listenerName
= new ListenerName(""),
-    securityProtocol = SecurityProtocol.PLAINTEXT)
+  val AllDone = new Request(processor = 1, connectionId = "2", Session(KafkaPrincipal.ANONYMOUS,
InetAddress.getLocalHost),
+    startTimeNanos = 0, listenerName = new ListenerName(""), securityProtocol = SecurityProtocol.PLAINTEXT,
+    MemoryPool.NONE, shutdownReceive)
   private val requestLogger = Logger.getLogger("kafka.request.logger")
 
   private def shutdownReceive: ByteBuffer = {
@@ -57,12 +57,11 @@ object RequestChannel extends Logging {
     val sanitizedUser = QuotaId.sanitize(principal.getName)
   }
 
-  case class Request(processor: Int, connectionId: String, session: Session, buffer: ByteBuffer,
-                     private val memoryPool: MemoryPool, startTimeNanos: Long, listenerName:
ListenerName, 
-                     securityProtocol: SecurityProtocol) {
+  class Request(val processor: Int, val connectionId: String, val session: Session, startTimeNanos:
Long,
+                val listenerName: ListenerName, val securityProtocol: SecurityProtocol, memoryPool:
MemoryPool,
+                @volatile private var buffer: ByteBuffer) {
     // These need to be volatile because the readers are in the network thread and the writers
are in the request
     // handler threads or the purgatory threads
-    @volatile var bufferReference = buffer
     @volatile var requestDequeueTimeNanos = -1L
     @volatile var apiLocalCompleteTimeNanos = -1L
     @volatile var responseCompleteTimeNanos = -1L
@@ -204,11 +203,19 @@ object RequestChannel extends Logging {
     }
 
     def dispose(): Unit = {
-      if (bufferReference != null) {
-        memoryPool.release(bufferReference)
-        bufferReference = null
+      if (buffer != null) {
+        memoryPool.release(buffer)
+        buffer = null
       }
     }
+
+    override def toString = s"Request(processor=$processor, " +
+      s"connectionId=$connectionId, " +
+      s"session=$session, " +
+      s"listenerName=$listenerName, " +
+      s"securityProtocol=$securityProtocol, " +
+      s"buffer=$buffer)"
+
   }
 
   object Response {
@@ -227,11 +234,13 @@ object RequestChannel extends Logging {
 
   }
 
-  case class Response(request: Request, responseSend: Option[Send], responseAction: ResponseAction)
{
+  class Response(val request: Request, val responseSend: Option[Send], val responseAction:
ResponseAction) {
     request.responseCompleteTimeNanos = Time.SYSTEM.nanoseconds
     if (request.apiLocalCompleteTimeNanos == -1L) request.apiLocalCompleteTimeNanos = Time.SYSTEM.nanoseconds
 
     def processor: Int = request.processor
+
+    override def toString = s"Response(request=$request, responseSend=$responseSend, responseAction=$responseAction)"
   }
 
   trait ResponseAction

http://git-wip-us.apache.org/repos/asf/kafka/blob/13deb84d/core/src/main/scala/kafka/network/SocketServer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/network/SocketServer.scala b/core/src/main/scala/kafka/network/SocketServer.scala
index e541015..f418fdd 100644
--- a/core/src/main/scala/kafka/network/SocketServer.scala
+++ b/core/src/main/scala/kafka/network/SocketServer.scala
@@ -533,10 +533,9 @@ private[kafka] class Processor(val id: Int,
         val openOrClosingChannel = if (openChannel != null) openChannel else selector.closingChannel(receive.source)
         val session = RequestChannel.Session(new KafkaPrincipal(KafkaPrincipal.USER_TYPE,
openOrClosingChannel.principal.getName), openOrClosingChannel.socketAddress)
 
-        val req = RequestChannel.Request(processor = id, connectionId = receive.source, session
= session,
-          buffer = receive.payload, startTimeNanos = time.nanoseconds,
-          listenerName = listenerName, securityProtocol = securityProtocol,
-          memoryPool = memoryPool)
+        val req = new RequestChannel.Request(processor = id, connectionId = receive.source,
session = session,
+          startTimeNanos = time.nanoseconds, listenerName = listenerName, securityProtocol
= securityProtocol,
+          memoryPool, receive.payload)
         requestChannel.sendRequest(req)
         selector.mute(receive.source)
       } catch {

http://git-wip-us.apache.org/repos/asf/kafka/blob/13deb84d/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
index ed35269..4d969cf 100644
--- a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
+++ b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
@@ -180,7 +180,7 @@ class SocketServerTest extends JUnitSuite {
     for (_ <- 0 until 10) {
       val request = server.requestChannel.receiveRequest(2000)
       assertNotNull("receiveRequest timed out", request)
-      server.requestChannel.sendResponse(RequestChannel.Response(request, None, RequestChannel.NoOpAction))
+      server.requestChannel.sendResponse(new RequestChannel.Response(request, None, RequestChannel.NoOpAction))
     }
   }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/13deb84d/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
index 38d4bb3..6cc3ede 100644
--- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
@@ -396,7 +396,8 @@ class KafkaApisTest {
     val header = new RequestHeader(builder.apiKey.id, request.version, "", 0)
     val buffer = request.serialize(header)
     val session = Session(KafkaPrincipal.ANONYMOUS, InetAddress.getLocalHost)
-    (request, RequestChannel.Request(1, "1", session, buffer, MemoryPool.NONE, 0, new ListenerName(""),
SecurityProtocol.PLAINTEXT))
+    (request, new RequestChannel.Request(1, "1", session, 0, new ListenerName(""), SecurityProtocol.PLAINTEXT,
+      MemoryPool.NONE, buffer))
   }
 
   private def readResponse(api: ApiKeys, request: AbstractRequest, capturedResponse: Capture[RequestChannel.Response]):
AbstractResponse = {


Mime
View raw message