You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jo...@apache.org on 2023/08/21 15:44:08 UTC
[kafka] branch trunk updated: KAFKA-15380: Execute action queue after callback request (#14197)
This is an automated email from the ASF dual-hosted git repository.
jolshan pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 4b383378a0f KAFKA-15380: Execute action queue after callback request (#14197)
4b383378a0f is described below
commit 4b383378a0fd19d6d3c9ae7c2175fa3459661a04
Author: Justine Olshan <jo...@confluent.io>
AuthorDate: Mon Aug 21 08:44:00 2023 -0700
KAFKA-15380: Execute action queue after callback request (#14197)
KIP-890 part 1 introduced the callback request type. It is used to execute a callback after KafkaApis.handle has returned. We did not account for tryCompleteActions at the end of handle when making this change.
In tests, we saw produce p99 increase dramatically (likely because we have to wait for another request before we can complete DelayedProduce). As a result, we should add the tryCompleteActions after the callback as well. In testing, this improved the produce performance.
Reviewers: Artem Livshits <al...@confluent.io>, Jason Gustafson <ja...@confluent.io>
---
core/src/main/scala/kafka/server/KafkaApis.scala | 4 ++
.../scala/kafka/server/KafkaRequestHandler.scala | 11 ++--
.../kafka/server/KafkaRequestHandlerTest.scala | 75 ++++++++++++++++------
3 files changed, 68 insertions(+), 22 deletions(-)
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
index 9010806bb7d..9af56815817 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -259,6 +259,10 @@ class KafkaApis(val requestChannel: RequestChannel,
}
}
+ override def tryCompleteActions(): Unit = {
+ replicaManager.tryCompleteActions()
+ }
+
def handleLeaderAndIsrRequest(request: RequestChannel.Request): Unit = {
val zkSupport = metadataSupport.requireZkOrThrow(KafkaApis.shouldNeverReceive(request))
// ensureTopicExists is only for client facing requests
diff --git a/core/src/main/scala/kafka/server/KafkaRequestHandler.scala b/core/src/main/scala/kafka/server/KafkaRequestHandler.scala
index 6191bf88120..316cf92ca5a 100755
--- a/core/src/main/scala/kafka/server/KafkaRequestHandler.scala
+++ b/core/src/main/scala/kafka/server/KafkaRequestHandler.scala
@@ -35,6 +35,7 @@ import scala.jdk.CollectionConverters._
trait ApiRequestHandler {
def handle(request: RequestChannel.Request, requestLocal: RequestLocal): Unit
+ def tryCompleteActions(): Unit = {}
}
object KafkaRequestHandler {
@@ -111,9 +112,9 @@ class KafkaRequestHandler(id: Int,
return
case callback: RequestChannel.CallbackRequest =>
+ val originalRequest = callback.originalRequest
try {
- val originalRequest = callback.originalRequest
-
+
// If we've already executed a callback for this request, reset the times and subtract the callback time from the
// new dequeue time. This will allow calculation of multiple callback times.
// Otherwise, set dequeue time to now.
@@ -127,14 +128,16 @@ class KafkaRequestHandler(id: Int,
threadCurrentRequest.set(originalRequest)
callback.fun()
- if (originalRequest.callbackRequestCompleteTimeNanos.isEmpty)
- originalRequest.callbackRequestCompleteTimeNanos = Some(time.nanoseconds())
} catch {
case e: FatalExitError =>
completeShutdown()
Exit.exit(e.statusCode)
case e: Throwable => error("Exception when handling request", e)
} finally {
+ // When handling requests, we try to complete actions after, so we should try to do so here as well.
+ apis.tryCompleteActions()
+ if (originalRequest.callbackRequestCompleteTimeNanos.isEmpty)
+ originalRequest.callbackRequestCompleteTimeNanos = Some(time.nanoseconds())
threadCurrentRequest.remove()
}
diff --git a/core/src/test/scala/kafka/server/KafkaRequestHandlerTest.scala b/core/src/test/scala/kafka/server/KafkaRequestHandlerTest.scala
index bcdafd4bd4d..822dd5263d6 100644
--- a/core/src/test/scala/kafka/server/KafkaRequestHandlerTest.scala
+++ b/core/src/test/scala/kafka/server/KafkaRequestHandlerTest.scala
@@ -24,7 +24,7 @@ import org.apache.kafka.common.network.{ClientInformation, ListenerName}
import org.apache.kafka.common.protocol.ApiKeys
import org.apache.kafka.common.requests.{RequestContext, RequestHeader}
import org.apache.kafka.common.security.auth.{KafkaPrincipal, SecurityProtocol}
-import org.apache.kafka.common.utils.MockTime
+import org.apache.kafka.common.utils.{MockTime, Time}
import org.apache.kafka.server.log.remote.storage.{RemoteLogManagerConfig, RemoteStorageMetrics}
import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
import org.junit.jupiter.api.Test
@@ -47,38 +47,65 @@ class KafkaRequestHandlerTest {
val metrics = new RequestChannel.Metrics(None)
val requestChannel = new RequestChannel(10, "", time, metrics)
val apiHandler = mock(classOf[ApiRequestHandler])
+ try {
+ val handler = new KafkaRequestHandler(0, 0, mock(classOf[Meter]), new AtomicInteger(1), requestChannel, apiHandler, time)
- // Make unsupported API versions request to avoid having to parse a real request
- val requestHeader = mock(classOf[RequestHeader])
- when(requestHeader.apiKey()).thenReturn(ApiKeys.API_VERSIONS)
- when(requestHeader.apiVersion()).thenReturn(0.toShort)
+ val request = makeRequest(time, metrics)
+ requestChannel.sendRequest(request)
- val context = new RequestContext(requestHeader, "0", mock(classOf[InetAddress]), new KafkaPrincipal("", ""),
- new ListenerName(""), SecurityProtocol.PLAINTEXT, mock(classOf[ClientInformation]), false)
- val request = new RequestChannel.Request(0, context, time.nanoseconds(),
- mock(classOf[MemoryPool]), ByteBuffer.allocate(0), metrics)
+ def callback(ms: Int): Unit = {
+ time.sleep(ms)
+ handler.stop()
+ }
+ when(apiHandler.handle(ArgumentMatchers.eq(request), any())).thenAnswer { _ =>
+ time.sleep(2)
+ KafkaRequestHandler.wrap(callback(_: Int))(1)
+ request.apiLocalCompleteTimeNanos = time.nanoseconds
+ }
+
+ handler.run()
+
+ assertEquals(startTime, request.requestDequeueTimeNanos)
+ assertEquals(startTime + 2000000, request.apiLocalCompleteTimeNanos)
+ assertEquals(Some(startTime + 2000000), request.callbackRequestDequeueTimeNanos)
+ assertEquals(Some(startTime + 3000000), request.callbackRequestCompleteTimeNanos)
+ } finally {
+ metrics.close()
+ }
+ }
+
+ @Test
+ def testCallbackTryCompleteActions(): Unit = {
+ val time = new MockTime()
+ val metrics = mock(classOf[RequestChannel.Metrics])
+ val apiHandler = mock(classOf[ApiRequestHandler])
+ val requestChannel = new RequestChannel(10, "", time, metrics)
val handler = new KafkaRequestHandler(0, 0, mock(classOf[Meter]), new AtomicInteger(1), requestChannel, apiHandler, time)
- requestChannel.sendRequest(request)
+ var handledCount = 0
+ var tryCompleteActionCount = 0
- def callback(ms: Int): Unit = {
- time.sleep(ms)
+ def callback(x: Int): Unit = {
handler.stop()
}
+ val request = makeRequest(time, metrics)
+ requestChannel.sendRequest(request)
+
when(apiHandler.handle(ArgumentMatchers.eq(request), any())).thenAnswer { _ =>
- time.sleep(2)
+ handledCount = handledCount + 1
KafkaRequestHandler.wrap(callback(_: Int))(1)
- request.apiLocalCompleteTimeNanos = time.nanoseconds
+ }
+
+ when(apiHandler.tryCompleteActions()).thenAnswer { _ =>
+ tryCompleteActionCount = tryCompleteActionCount + 1
}
handler.run()
- assertEquals(startTime, request.requestDequeueTimeNanos)
- assertEquals(startTime + 2000000, request.apiLocalCompleteTimeNanos)
- assertEquals(Some(startTime + 2000000), request.callbackRequestDequeueTimeNanos)
- assertEquals(Some(startTime + 3000000), request.callbackRequestCompleteTimeNanos)
+ assertEquals(1, handledCount)
+ assertEquals(1, tryCompleteActionCount)
}
@@ -98,4 +125,16 @@ class KafkaRequestHandlerTest {
}
})
}
+
+ def makeRequest(time: Time, metrics: RequestChannel.Metrics): RequestChannel.Request = {
+ // Make unsupported API versions request to avoid having to parse a real request
+ val requestHeader = mock(classOf[RequestHeader])
+ when(requestHeader.apiKey()).thenReturn(ApiKeys.API_VERSIONS)
+ when(requestHeader.apiVersion()).thenReturn(0.toShort)
+
+ val context = new RequestContext(requestHeader, "0", mock(classOf[InetAddress]), new KafkaPrincipal("", ""),
+ new ListenerName(""), SecurityProtocol.PLAINTEXT, mock(classOf[ClientInformation]), false)
+ new RequestChannel.Request(0, context, time.nanoseconds(),
+ mock(classOf[MemoryPool]), ByteBuffer.allocate(0), metrics)
+ }
}