You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/01/15 21:06:21 UTC
[GitHub] [kafka] hachikuji opened a new pull request #9912: MINOR: Let TestRaftRequestHandler use RequestHandlerHelper
hachikuji opened a new pull request #9912:
URL: https://github.com/apache/kafka/pull/9912
Since we factored out `RequestHandlerHelper`, we may as well make use of it in `TestRaftRequestHandler`.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9912: MINOR: Let TestRaftRequestHandler use RequestHandlerHelper
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9912:
URL: https://github.com/apache/kafka/pull/9912#discussion_r580606013
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -2127,21 +2118,15 @@ class KafkaApisTest {
error = Errors.NONE
))
- val response = readResponse(joinGroupRequest, capturedResponse)
- .asInstanceOf[JoinGroupResponse]
+ val response = capturedResponse.getValue.asInstanceOf[JoinGroupResponse]
assertEquals(Errors.NONE, response.error)
assertEquals(0, response.data.members.size)
assertEquals(memberId, response.data.memberId)
assertEquals(0, response.data.generationId)
assertEquals(memberId, response.data.leader)
assertEquals(protocolName, response.data.protocolName)
-
- if (version >= 7) {
Review comment:
Now these test cases are working with the raw response object returned by `KafkaApis`. This means we don't have to deal with fields which are dropped during serialization.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on pull request #9912: MINOR: Let TestRaftRequestHandler use RequestHandlerHelper
Posted by GitBox <gi...@apache.org>.
hachikuji commented on pull request #9912:
URL: https://github.com/apache/kafka/pull/9912#issuecomment-763219863
So what I ended up doing is moving the helpers that `TestRaftRequestHandler` needed into `RequestChannel`. Basically this encapsulates the construction of the `RequestChannel.Response`, which seems a bit nicer since it was confusing to see `sendResponse` used for throttling events. It is also nice because it ensures that we do not need to rely on the caller to ensure that `updateErrorMetrics` gets called.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9912: MINOR: Let TestRaftRequestHandler use RequestHandlerHelper
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9912:
URL: https://github.com/apache/kafka/pull/9912#discussion_r561189636
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -227,37 +225,33 @@ class KafkaApisTest {
authorizeResource(authorizer, operation, ResourceType.TOPIC, resourceName, AuthorizationResult.ALLOWED)
- val capturedResponse = expectNoThrottling()
-
val configResource = new ConfigResource(ConfigResource.Type.TOPIC, resourceName)
EasyMock.expect(adminManager.alterConfigs(anyObject(), EasyMock.eq(false)))
.andAnswer(() => {
Map(configResource -> alterConfigHandler.apply())
})
- EasyMock.replay(replicaManager, clientRequestQuotaManager, requestChannel, authorizer,
- adminManager, controller)
-
val configs = Map(
configResource -> new AlterConfigsRequest.Config(
Seq(new AlterConfigsRequest.ConfigEntry("foo", "bar")).asJava))
val alterConfigsRequest = new AlterConfigsRequest.Builder(configs.asJava, false).build(requestHeader.apiVersion)
val request = buildRequestWithEnvelope(alterConfigsRequest, fromPrivilegedListener = true)
+ val capturedResponse = EasyMock.newCapture[AbstractResponse]()
+ val capturedRequest = EasyMock.newCapture[RequestChannel.Request]()
- createKafkaApis(authorizer = Some(authorizer), enableForwarding = true).handle(request)
-
- val envelopeRequest = request.body[EnvelopeRequest]
- val response = readResponse(envelopeRequest, capturedResponse)
- .asInstanceOf[EnvelopeResponse]
-
- assertEquals(Errors.NONE, response.error)
Review comment:
There remains some awkwardness in the handling of envelope requests. Basically the flow is like this:
1. KafkaApis.handle(envelope(alterConfigRequest))
2. KafkaApis.handle(alterConfigRequest)
3. RequestChannel.sendResponse(alterConfigResponse)
4. Request.buildResponseSend() -> envelope(alterConfigResponse)
Previously `KafkaApisTest` had to work by parsing the response send, so we had to unwrap the envelope. But now we get the direct call to `sendResponse` with the embedded `AbstractResponse` instance. So basically we do not hit step 4 anymore.
I think a nicer way to structure this which we can consider separately is to change `KafkaApis.handle` so that it takes a callback rather than assuming that responses are sent directly to the request channel. Then when we make the recursive call to `handle` after unwrapping the envelope, we can provide a callback which wraps the underlying response with the response envelope. Alternatively, we can have `KafkaApis.handle` return a `CompletableFuture`. The main point is that we allow for some custom behavior when the response is ready.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9912: MINOR: Let TestRaftRequestHandler use RequestHandlerHelper
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9912:
URL: https://github.com/apache/kafka/pull/9912#discussion_r560342977
##########
File path: core/src/main/scala/kafka/server/RequestHandlerHelper.scala
##########
@@ -58,16 +57,68 @@ object RequestHandlerHelper {
}
}
+class UnthrottledRequestHandlerHelper(
+ requestChannel: RequestChannel,
+ logPrefix: String
+) extends Logging {
+ this.logIdent = logPrefix
+ def handleError(request: RequestChannel.Request, e: Throwable): Unit = {
+ error("Error when handling request: " +
+ s"clientId=${request.header.clientId}, " +
+ s"correlationId=${request.header.correlationId}, " +
+ s"api=${request.header.apiKey}, " +
+ s"version=${request.header.apiVersion}, " +
+ s"body=${request.body[AbstractRequest]}", e)
+ sendErrorOrCloseConnection(request, e, 0)
Review comment:
The approach I took here was intended as a lighter touch. Many of the APIs exposed by `RequestHandlerHelper` set throttling expectations explicitly (e.g. `sendResponseExemptThrottle`). That makes it a little awkward to push down the throttling logic. If we only depended on the request quota, then I would be tempted to try, but we also depend on the controller mutation quota. I did consider changing the dependence from `QuotaManagers` to `ClientRequestQuotaManager` and `ControllerMutationQuotaManager`. Then we could pull out a trait for both of these. But all of this seemed like more work with little benefit.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9912: MINOR: Let TestRaftRequestHandler use RequestHandlerHelper
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9912:
URL: https://github.com/apache/kafka/pull/9912#discussion_r558921130
##########
File path: core/src/main/scala/kafka/server/RequestHandlerHelper.scala
##########
@@ -58,16 +57,68 @@ object RequestHandlerHelper {
}
}
+class UnthrottledRequestHandlerHelper(
+ requestChannel: RequestChannel,
+ logPrefix: String
+) extends Logging {
+ this.logIdent = logPrefix
+ def handleError(request: RequestChannel.Request, e: Throwable): Unit = {
+ error("Error when handling request: " +
+ s"clientId=${request.header.clientId}, " +
+ s"correlationId=${request.header.correlationId}, " +
+ s"api=${request.header.apiKey}, " +
+ s"version=${request.header.apiVersion}, " +
+ s"body=${request.body[AbstractRequest]}", e)
+ sendErrorOrCloseConnection(request, e, 0)
Review comment:
Did we consider having the throttle calls (`maybeRecordExempt`, `throttle`, `maybeRecordAndGetThrottleTimeMs`, etc.) split into a separate interface and then have an unthrottled implementation of that?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9912: MINOR: Let TestRaftRequestHandler use RequestHandlerHelper
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9912:
URL: https://github.com/apache/kafka/pull/9912#discussion_r561175037
##########
File path: core/src/main/scala/kafka/network/RequestChannel.scala
##########
@@ -371,9 +372,44 @@ class RequestChannel(val queueSize: Int,
requestQueue.put(request)
}
- /** Send a response back to the socket server to be sent over the network */
- def sendResponse(response: RequestChannel.Response): Unit = {
+ def closeConnection(
+ request: RequestChannel.Request,
+ errorCounts: java.util.Map[Errors, Integer]
+ ): Unit = {
+ // This case is used when the request handler has encountered an error, but the client
+ // does not expect a response (e.g. when produce request has acks set to 0)
+ updateErrorMetrics(request.header.apiKey, errorCounts.asScala)
+ sendResponse(new RequestChannel.CloseConnectionResponse(request))
+ }
+
+ def sendResponse(
+ request: RequestChannel.Request,
+ response: AbstractResponse,
+ onComplete: Option[Send => Unit]
Review comment:
That's fair. I will check the scope of this change.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9912: MINOR: Let TestRaftRequestHandler use RequestHandlerHelper
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9912:
URL: https://github.com/apache/kafka/pull/9912#discussion_r563455980
##########
File path: core/src/main/scala/kafka/network/RequestChannel.scala
##########
@@ -371,9 +372,44 @@ class RequestChannel(val queueSize: Int,
requestQueue.put(request)
}
- /** Send a response back to the socket server to be sent over the network */
- def sendResponse(response: RequestChannel.Response): Unit = {
+ def closeConnection(
+ request: RequestChannel.Request,
+ errorCounts: java.util.Map[Errors, Integer]
+ ): Unit = {
+ // This case is used when the request handler has encountered an error, but the client
+ // does not expect a response (e.g. when produce request has acks set to 0)
+ updateErrorMetrics(request.header.apiKey, errorCounts.asScala)
+ sendResponse(new RequestChannel.CloseConnectionResponse(request))
+ }
+
+ def sendResponse(
+ request: RequestChannel.Request,
+ response: AbstractResponse,
+ onComplete: Option[Send => Unit]
Review comment:
Sounds good.
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -227,37 +225,33 @@ class KafkaApisTest {
authorizeResource(authorizer, operation, ResourceType.TOPIC, resourceName, AuthorizationResult.ALLOWED)
- val capturedResponse = expectNoThrottling()
-
val configResource = new ConfigResource(ConfigResource.Type.TOPIC, resourceName)
EasyMock.expect(adminManager.alterConfigs(anyObject(), EasyMock.eq(false)))
.andAnswer(() => {
Map(configResource -> alterConfigHandler.apply())
})
- EasyMock.replay(replicaManager, clientRequestQuotaManager, requestChannel, authorizer,
- adminManager, controller)
-
val configs = Map(
configResource -> new AlterConfigsRequest.Config(
Seq(new AlterConfigsRequest.ConfigEntry("foo", "bar")).asJava))
val alterConfigsRequest = new AlterConfigsRequest.Builder(configs.asJava, false).build(requestHeader.apiVersion)
val request = buildRequestWithEnvelope(alterConfigsRequest, fromPrivilegedListener = true)
+ val capturedResponse = EasyMock.newCapture[AbstractResponse]()
+ val capturedRequest = EasyMock.newCapture[RequestChannel.Request]()
- createKafkaApis(authorizer = Some(authorizer), enableForwarding = true).handle(request)
-
- val envelopeRequest = request.body[EnvelopeRequest]
- val response = readResponse(envelopeRequest, capturedResponse)
- .asInstanceOf[EnvelopeResponse]
-
- assertEquals(Errors.NONE, response.error)
Review comment:
Thanks for the explanation.
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -529,15 +518,14 @@ class KafkaApisTest {
new IncrementalAlterConfigsRequest.Builder(resourceMap, false)
}
+ // FIXME: Unused arg
Review comment:
Does this still have to be addressed? Either way, the FIXME needs to be removed.
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -2127,21 +2118,15 @@ class KafkaApisTest {
error = Errors.NONE
))
- val response = readResponse(joinGroupRequest, capturedResponse)
- .asInstanceOf[JoinGroupResponse]
+ val response = capturedResponse.getValue.asInstanceOf[JoinGroupResponse]
assertEquals(Errors.NONE, response.error)
assertEquals(0, response.data.members.size)
assertEquals(memberId, response.data.memberId)
assertEquals(0, response.data.generationId)
assertEquals(memberId, response.data.leader)
assertEquals(protocolName, response.data.protocolName)
-
- if (version >= 7) {
Review comment:
Why is this not needed?
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -579,11 +566,11 @@ class KafkaApisTest {
testForwardableAPI(ApiKeys.ALTER_CLIENT_QUOTAS, requestBuilder)
}
+ // FIXME: Unused arg
Review comment:
Similar to the other comment.
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -2200,17 +2183,11 @@ class KafkaApisTest {
error = Errors.NONE
))
- val response = readResponse(syncGroupRequest, capturedResponse)
- .asInstanceOf[SyncGroupResponse]
+ val response = capturedResponse.getValue.asInstanceOf[SyncGroupResponse]
assertEquals(Errors.NONE, response.error)
assertArrayEquals(Array.empty[Byte], response.data.assignment)
-
- if (version >= 5) {
Review comment:
Similar question, why is this not needed?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9912: MINOR: Let TestRaftRequestHandler use RequestHandlerHelper
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9912:
URL: https://github.com/apache/kafka/pull/9912#discussion_r560345597
##########
File path: core/src/main/scala/kafka/server/RequestHandlerHelper.scala
##########
@@ -58,16 +57,68 @@ object RequestHandlerHelper {
}
}
+class UnthrottledRequestHandlerHelper(
+ requestChannel: RequestChannel,
+ logPrefix: String
+) extends Logging {
+ this.logIdent = logPrefix
+ def handleError(request: RequestChannel.Request, e: Throwable): Unit = {
+ error("Error when handling request: " +
+ s"clientId=${request.header.clientId}, " +
+ s"correlationId=${request.header.correlationId}, " +
+ s"api=${request.header.apiKey}, " +
+ s"version=${request.header.apiVersion}, " +
+ s"body=${request.body[AbstractRequest]}", e)
+ sendErrorOrCloseConnection(request, e, 0)
Review comment:
Thanks for the explanation. The `handleError` duplication seems a bit ugly. Also, inheriting from concrete classes is generally not great. So, to me, the current approach is lighter touch, but doesn't feel right.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji merged pull request #9912: MINOR: Move `RequestChannel.Response` creation logic into `RequestChannel`
Posted by GitBox <gi...@apache.org>.
hachikuji merged pull request #9912:
URL: https://github.com/apache/kafka/pull/9912
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9912: MINOR: Let TestRaftRequestHandler use RequestHandlerHelper
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9912:
URL: https://github.com/apache/kafka/pull/9912#discussion_r561415958
##########
File path: core/src/main/scala/kafka/network/RequestChannel.scala
##########
@@ -371,9 +372,44 @@ class RequestChannel(val queueSize: Int,
requestQueue.put(request)
}
- /** Send a response back to the socket server to be sent over the network */
- def sendResponse(response: RequestChannel.Response): Unit = {
+ def closeConnection(
+ request: RequestChannel.Request,
+ errorCounts: java.util.Map[Errors, Integer]
+ ): Unit = {
+ // This case is used when the request handler has encountered an error, but the client
+ // does not expect a response (e.g. when produce request has acks set to 0)
+ updateErrorMetrics(request.header.apiKey, errorCounts.asScala)
+ sendResponse(new RequestChannel.CloseConnectionResponse(request))
+ }
+
+ def sendResponse(
+ request: RequestChannel.Request,
+ response: AbstractResponse,
+ onComplete: Option[Send => Unit]
Review comment:
I decided not to do this here. I didn't like replacing `None` with `_ => {}` in uses, and neither did I like making the argument optional. The alternative is to introduce a constant "no-op" function, but I found this also a little awkward. If you think of a nice way to do it, I can review. I do think it is better having a simpler type.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9912: MINOR: Let TestRaftRequestHandler use RequestHandlerHelper
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9912:
URL: https://github.com/apache/kafka/pull/9912#discussion_r561415958
##########
File path: core/src/main/scala/kafka/network/RequestChannel.scala
##########
@@ -371,9 +372,44 @@ class RequestChannel(val queueSize: Int,
requestQueue.put(request)
}
- /** Send a response back to the socket server to be sent over the network */
- def sendResponse(response: RequestChannel.Response): Unit = {
+ def closeConnection(
+ request: RequestChannel.Request,
+ errorCounts: java.util.Map[Errors, Integer]
+ ): Unit = {
+ // This case is used when the request handler has encountered an error, but the client
+ // does not expect a response (e.g. when produce request has acks set to 0)
+ updateErrorMetrics(request.header.apiKey, errorCounts.asScala)
+ sendResponse(new RequestChannel.CloseConnectionResponse(request))
+ }
+
+ def sendResponse(
+ request: RequestChannel.Request,
+ response: AbstractResponse,
+ onComplete: Option[Send => Unit]
Review comment:
I decided not to do this here. I didn't like replacing `None` with `_ => {}` in uses, and neither did I like making the argument optional. The alternative is to introduce a constant "no-op" function, but I found this also a little awkward. If you think of a nice way to do it, I can review. I do think it is better having a simpler type.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9912: MINOR: Let TestRaftRequestHandler use RequestHandlerHelper
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9912:
URL: https://github.com/apache/kafka/pull/9912#discussion_r560389436
##########
File path: core/src/main/scala/kafka/server/RequestHandlerHelper.scala
##########
@@ -58,16 +57,68 @@ object RequestHandlerHelper {
}
}
+class UnthrottledRequestHandlerHelper(
+ requestChannel: RequestChannel,
+ logPrefix: String
+) extends Logging {
+ this.logIdent = logPrefix
+ def handleError(request: RequestChannel.Request, e: Throwable): Unit = {
+ error("Error when handling request: " +
+ s"clientId=${request.header.clientId}, " +
+ s"correlationId=${request.header.correlationId}, " +
+ s"api=${request.header.apiKey}, " +
+ s"version=${request.header.apiVersion}, " +
+ s"body=${request.body[AbstractRequest]}", e)
+ sendErrorOrCloseConnection(request, e, 0)
Review comment:
There is no state inside `UnthrottledRequestHandlerHelper`, so we could pull these methods into the companion I guess. Not sure that's really better, but at least it removes the inheritance. Let me see how this looks.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9912: MINOR: Let TestRaftRequestHandler use RequestHandlerHelper
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9912:
URL: https://github.com/apache/kafka/pull/9912#discussion_r561002429
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/RequestContext.java
##########
@@ -175,4 +175,19 @@ public String clientId() {
public int correlationId() {
return header.correlationId();
}
+
+ @Override
+ public String toString() {
+ return "RequestContext(" +
+ "header=" + header +
+ ", connectionId='" + connectionId + '\'' +
+ ", clientAddress=" + clientAddress +
+ ", principal=" + principal +
+ ", listenerName=" + listenerName +
+ ", securityProtocol=" + securityProtocol +
+ ", clientInformation=" + clientInformation +
+ ", fromPrivilegedListener=" + fromPrivilegedListener +
+ ", principalSerde=" + principalSerde +
Review comment:
Have we checked that no privileged information (i.e. credentials) are logged by any of these fields? Seems OK to me, but worth double checking.
##########
File path: core/src/test/scala/unit/kafka/server/ThrottledChannelExpirationTest.scala
##########
@@ -44,28 +33,13 @@ class ThrottledChannelExpirationTest {
private val metrics = new org.apache.kafka.common.metrics.Metrics(new MetricConfig(),
Collections.emptyList(),
time)
- private val request = buildRequest(FetchRequest.Builder.forConsumer(0, 1000, new util.HashMap[TopicPartition, PartitionData]))._2
-
- private def buildRequest[T <: AbstractRequest](builder: AbstractRequest.Builder[T],
- listenerName: ListenerName = ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT)): (T, RequestChannel.Request) = {
-
- val request = builder.build()
- val buffer = RequestTestUtils.serializeRequestWithHeader(
- new RequestHeader(builder.apiKey, request.version, "", 0), request)
- val requestChannelMetrics: RequestChannel.Metrics = EasyMock.createNiceMock(classOf[RequestChannel.Metrics])
-
- // read the header from the buffer first so that the body can be read next from the Request constructor
- val header = RequestHeader.parse(buffer)
- val context = new RequestContext(header, "1", InetAddress.getLocalHost, KafkaPrincipal.ANONYMOUS,
- listenerName, SecurityProtocol.PLAINTEXT, ClientInformation.EMPTY, false)
- (request, new RequestChannel.Request(processor = 1, context = context, startTimeNanos = 0, MemoryPool.NONE, buffer,
- requestChannelMetrics))
- }
Review comment:
Nice that we don't need this anymore.
##########
File path: core/src/main/scala/kafka/network/RequestChannel.scala
##########
@@ -371,9 +372,44 @@ class RequestChannel(val queueSize: Int,
requestQueue.put(request)
}
- /** Send a response back to the socket server to be sent over the network */
- def sendResponse(response: RequestChannel.Response): Unit = {
+ def closeConnection(
+ request: RequestChannel.Request,
+ errorCounts: java.util.Map[Errors, Integer]
+ ): Unit = {
+ // This case is used when the request handler has encountered an error, but the client
+ // does not expect a response (e.g. when produce request has acks set to 0)
+ updateErrorMetrics(request.header.apiKey, errorCounts.asScala)
+ sendResponse(new RequestChannel.CloseConnectionResponse(request))
+ }
+
+ def sendResponse(
+ request: RequestChannel.Request,
+ response: AbstractResponse,
+ onComplete: Option[Send => Unit]
Review comment:
I wonder if this should be `Send => Unit` instead. It doesn't seem like there is much value in the `Option` here. A function that does nothing is equivalent to `None`. I know this is an existing issue, so ok if you prefer not to address now.
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -227,37 +225,33 @@ class KafkaApisTest {
authorizeResource(authorizer, operation, ResourceType.TOPIC, resourceName, AuthorizationResult.ALLOWED)
- val capturedResponse = expectNoThrottling()
-
val configResource = new ConfigResource(ConfigResource.Type.TOPIC, resourceName)
EasyMock.expect(adminManager.alterConfigs(anyObject(), EasyMock.eq(false)))
.andAnswer(() => {
Map(configResource -> alterConfigHandler.apply())
})
- EasyMock.replay(replicaManager, clientRequestQuotaManager, requestChannel, authorizer,
- adminManager, controller)
-
val configs = Map(
configResource -> new AlterConfigsRequest.Config(
Seq(new AlterConfigsRequest.ConfigEntry("foo", "bar")).asJava))
val alterConfigsRequest = new AlterConfigsRequest.Builder(configs.asJava, false).build(requestHeader.apiVersion)
val request = buildRequestWithEnvelope(alterConfigsRequest, fromPrivilegedListener = true)
+ val capturedResponse = EasyMock.newCapture[AbstractResponse]()
+ val capturedRequest = EasyMock.newCapture[RequestChannel.Request]()
- createKafkaApis(authorizer = Some(authorizer), enableForwarding = true).handle(request)
-
- val envelopeRequest = request.body[EnvelopeRequest]
- val response = readResponse(envelopeRequest, capturedResponse)
- .asInstanceOf[EnvelopeResponse]
-
- assertEquals(Errors.NONE, response.error)
Review comment:
We don't need this assert anymore? Can you explain a bit more what we're trying to do with the changes in this method? That will probably help with some other envelope related test changes in this class too.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9912: MINOR: Let TestRaftRequestHandler use RequestHandlerHelper
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9912:
URL: https://github.com/apache/kafka/pull/9912#discussion_r560342977
##########
File path: core/src/main/scala/kafka/server/RequestHandlerHelper.scala
##########
@@ -58,16 +57,68 @@ object RequestHandlerHelper {
}
}
+class UnthrottledRequestHandlerHelper(
+ requestChannel: RequestChannel,
+ logPrefix: String
+) extends Logging {
+ this.logIdent = logPrefix
+ def handleError(request: RequestChannel.Request, e: Throwable): Unit = {
+ error("Error when handling request: " +
+ s"clientId=${request.header.clientId}, " +
+ s"correlationId=${request.header.correlationId}, " +
+ s"api=${request.header.apiKey}, " +
+ s"version=${request.header.apiVersion}, " +
+ s"body=${request.body[AbstractRequest]}", e)
+ sendErrorOrCloseConnection(request, e, 0)
Review comment:
The approach I took here was intended as a lighter touch. Many of the APIs exposed by `RequestHandlerHelper` set throttling expectations explicitly (e.g. `sendResponseExemptThrottle`). That makes it a little awkward to push down the throttling logic. If we only depended on the request quota, then I would be tempted to try, but we also depend on the controller mutation quota. I did consider changing the dependence from `QuotaManagers` to `ClientRequestQuotaManager` and `ControllerMutationQuotaManager`. Then we could pull out a trait for both of this. But all of this seemed like more work with little benefit.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9912: MINOR: Let TestRaftRequestHandler use RequestHandlerHelper
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9912:
URL: https://github.com/apache/kafka/pull/9912#discussion_r561174514
##########
File path: clients/src/main/java/org/apache/kafka/common/requests/RequestContext.java
##########
@@ -175,4 +175,19 @@ public String clientId() {
public int correlationId() {
return header.correlationId();
}
+
+ @Override
+ public String toString() {
+ return "RequestContext(" +
+ "header=" + header +
+ ", connectionId='" + connectionId + '\'' +
+ ", clientAddress=" + clientAddress +
+ ", principal=" + principal +
+ ", listenerName=" + listenerName +
+ ", securityProtocol=" + securityProtocol +
+ ", clientInformation=" + clientInformation +
+ ", fromPrivilegedListener=" + fromPrivilegedListener +
+ ", principalSerde=" + principalSerde +
Review comment:
Yeah, I considered it, but didn't see anything in here that looked sensitive. Most of this information is already included in the request logging. The client address is not, but that seems ok. Including the principal serde is annoying, but doesn't seem like a problem.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on pull request #9912: MINOR: Let TestRaftRequestHandler use RequestHandlerHelper
Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #9912:
URL: https://github.com/apache/kafka/pull/9912#issuecomment-763684101
Also, looks like there is a compiler error:
> 02:22:44 [Error] /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-9912/core/src/main/scala/kafka/server/KafkaApis.scala:148: value closeConnection is not a member of kafka.server.RequestHandlerHelper
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9912: MINOR: Let TestRaftRequestHandler use RequestHandlerHelper
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9912:
URL: https://github.com/apache/kafka/pull/9912#discussion_r563460396
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -529,15 +518,14 @@ class KafkaApisTest {
new IncrementalAlterConfigsRequest.Builder(resourceMap, false)
}
+ // FIXME: Unused arg
Review comment:
Does this still have to be addressed? Either way, the FIXME needs to be removed.
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -2127,21 +2118,15 @@ class KafkaApisTest {
error = Errors.NONE
))
- val response = readResponse(joinGroupRequest, capturedResponse)
- .asInstanceOf[JoinGroupResponse]
+ val response = capturedResponse.getValue.asInstanceOf[JoinGroupResponse]
assertEquals(Errors.NONE, response.error)
assertEquals(0, response.data.members.size)
assertEquals(memberId, response.data.memberId)
assertEquals(0, response.data.generationId)
assertEquals(memberId, response.data.leader)
assertEquals(protocolName, response.data.protocolName)
-
- if (version >= 7) {
Review comment:
Why is this not needed?
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -579,11 +566,11 @@ class KafkaApisTest {
testForwardableAPI(ApiKeys.ALTER_CLIENT_QUOTAS, requestBuilder)
}
+ // FIXME: Unused arg
Review comment:
Similar to the other comment.
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -2200,17 +2183,11 @@ class KafkaApisTest {
error = Errors.NONE
))
- val response = readResponse(syncGroupRequest, capturedResponse)
- .asInstanceOf[SyncGroupResponse]
+ val response = capturedResponse.getValue.asInstanceOf[SyncGroupResponse]
assertEquals(Errors.NONE, response.error)
assertArrayEquals(Array.empty[Byte], response.data.assignment)
-
- if (version >= 5) {
Review comment:
Similar question, why is this not needed?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9912: MINOR: Let TestRaftRequestHandler use RequestHandlerHelper
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9912:
URL: https://github.com/apache/kafka/pull/9912#discussion_r563459295
##########
File path: core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
##########
@@ -227,37 +225,33 @@ class KafkaApisTest {
authorizeResource(authorizer, operation, ResourceType.TOPIC, resourceName, AuthorizationResult.ALLOWED)
- val capturedResponse = expectNoThrottling()
-
val configResource = new ConfigResource(ConfigResource.Type.TOPIC, resourceName)
EasyMock.expect(adminManager.alterConfigs(anyObject(), EasyMock.eq(false)))
.andAnswer(() => {
Map(configResource -> alterConfigHandler.apply())
})
- EasyMock.replay(replicaManager, clientRequestQuotaManager, requestChannel, authorizer,
- adminManager, controller)
-
val configs = Map(
configResource -> new AlterConfigsRequest.Config(
Seq(new AlterConfigsRequest.ConfigEntry("foo", "bar")).asJava))
val alterConfigsRequest = new AlterConfigsRequest.Builder(configs.asJava, false).build(requestHeader.apiVersion)
val request = buildRequestWithEnvelope(alterConfigsRequest, fromPrivilegedListener = true)
+ val capturedResponse = EasyMock.newCapture[AbstractResponse]()
+ val capturedRequest = EasyMock.newCapture[RequestChannel.Request]()
- createKafkaApis(authorizer = Some(authorizer), enableForwarding = true).handle(request)
-
- val envelopeRequest = request.body[EnvelopeRequest]
- val response = readResponse(envelopeRequest, capturedResponse)
- .asInstanceOf[EnvelopeResponse]
-
- assertEquals(Errors.NONE, response.error)
Review comment:
Thanks for the explanation.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #9912: MINOR: Let TestRaftRequestHandler use RequestHandlerHelper
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #9912:
URL: https://github.com/apache/kafka/pull/9912#discussion_r563455980
##########
File path: core/src/main/scala/kafka/network/RequestChannel.scala
##########
@@ -371,9 +372,44 @@ class RequestChannel(val queueSize: Int,
requestQueue.put(request)
}
- /** Send a response back to the socket server to be sent over the network */
- def sendResponse(response: RequestChannel.Response): Unit = {
+ def closeConnection(
+ request: RequestChannel.Request,
+ errorCounts: java.util.Map[Errors, Integer]
+ ): Unit = {
+ // This case is used when the request handler has encountered an error, but the client
+ // does not expect a response (e.g. when produce request has acks set to 0)
+ updateErrorMetrics(request.header.apiKey, errorCounts.asScala)
+ sendResponse(new RequestChannel.CloseConnectionResponse(request))
+ }
+
+ def sendResponse(
+ request: RequestChannel.Request,
+ response: AbstractResponse,
+ onComplete: Option[Send => Unit]
Review comment:
Sounds good.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org