[ https://issues.apache.org/jira/browse/KAFKA-13829?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17522199#comment-17522199 ]
Luke Chen commented on KAFKA-13829: ----------------------------------- > In the InFlightRequests.canSendMore() method, {*}if the > `queue.peekFirst().send.completed()` condition is not met{*}, the effect of > setting max.in.flight.requests.per.connection greater than 1 is equivalent to > setting 1. This sounds like a bug. Anyway, I can't just say this is correct or wrong from what you've described. Maybe other experts can do that. My suggestion is the same, to have a PR with tests to prove your understanding is correct, which should be better and clearer. > The function of max.in.flight.requests.per.connection parameter does not > work, it conflicts with the underlying NIO sending data > -------------------------------------------------------------------------------------------------------------------------------- > > Key: KAFKA-13829 > URL: https://issues.apache.org/jira/browse/KAFKA-13829 > Project: Kafka > Issue Type: Improvement > Components: clients > Reporter: RivenSun > Priority: Major > > Due to the implementation mechanism of the `OP_WRITE` event of Kafka's > underlying NIO, the function of the `max.in.flight.requests.per.connection` > parameter does not work. This will greatly affect Kafka's network sending > performance. > The process of Kafka's Selector sending ClientRequest can be simply divided > into the following two major steps: > h2. 1) Prepare the request data to be sent > 1. NetworkClient.ready(Node node, long now) -> > NetworkClient.canSendRequest(...) method is called: determine whether the > request is eligible to be sent. > 2.NetworkClient.doSend(...): Construct `Send send` and cache the request by > inFlightRequests.add(inFlightRequest). > 3. Execute the `KafkaChannel.setSend()` method: > Judging that `this.send` *must be null* at present, otherwise {*}an > IllegalStateException is thrown{*}; > Set `this.send` value; > transportLayer adds `OP_WRITE` event. > > h2. 2) Selector.poll(long timeout) should then be called to consume the send > and send data to the network. > > {code:java} > Selector.poll() -> this.nioSelector.select(timeout) > Selector.pollSelectionKeys() -> > Selector.attemptWrite() -> > Selector.write(channel) -> > KafkaChannel.write() & KafkaChannel.maybeCompleteSend(){code} > > 1.Selector.poll -> this.nioSelector.select(timeout) : Get the previously > registered `OP_WRITE` event > 2. Execute the `Selector.attemptWrite()` method > 3. In the `KafkaChannel.write()` method, after the data is successfully sent, > update `this.send` to the completed state. > 4. In the `KafkaChannel.maybeCompleteSend()` method, check whether send is in > the completed state, otherwise do nothing. > 5. If send is completed, transportLayer removes the `OP_WRITE` event and > resets the KafkaChannel.send object to null. > 6. Wait for the next request data that is ready to be sent. > > It seems that there is no problem with the whole process above, but carefully > read the method of NetworkClient.canSendRequest(...), there is such a > condition in inFlightRequests.canSendMore(node): > {code:java} > queue.peekFirst().send.completed(){code} > Secondly, the `inFlightRequests.add(inFlightRequest)` method also calls > *addFirst(request).* > > Currently, *only one send object* is stored in KafkaChannel, not a sendObject > {*}collection{*}; > During OP_WRITE event registration and removal, *only* *one* *send* *object* > *will be sent* in the KafkaChannel.write() method and *only one send* > *object* *will be completed* in the KafkaChannel.maybeCompleteSend() method. > So whether the clientRequest is eligible to be sent will {color:#ff0000}*only > be limited by queue.peekFirst().send.completed()*{color}, the > {color:#ff0000}*max.in.flight.requests.per.connection*{color} parameter will > lose its effect, and {*}{color:#ff0000}the effect of setting greater than 1 > is equivalent in 1{color}{*}. > h2. > Suggest > Due to the KafkaClient architecture, we do not need to consider the > concurrent execution of multiple threads of the `NetworkClient.poll` method. > 1.NetworkClient.canSendRequest(...) removes the condition,: > {code:java} > queue.peekFirst().send.completed(){code} > `max.in.flight.requests.per.connection` parameter will work again. > 2. Before, in KafkaChannel.setSend(), register the `OP_WRITE` event, and in > KafkaChannel.maybeCompleteSend(), remove the `OP_WRITE` event. > It may no longer be appropriate now. Because we want to send more than one > send object data during the registration and removal of `OP_WRITE` events, it > is recommended {*}not to repeatedly register and remove `OP_WRITE` events{*}, > but choose to {*}register the `OP_WRITE` event in the > `transportLayer.finishConnect()` method{*}: > {code:java} > key.interestOps(key.interestOps() & ~SelectionKey.OP_CONNECT | > SelectionKey.OP_READ | SelectionKey.OP_WRITE);{code} > 3. The KafkaChannel.setSend() method does not only cache *an* incoming send > object, but the incoming send object is stored in the *sendCollection* > structure. > 4. When KafkaChannel.write() is executed, copy sendCollection to > {*}midWriteSendCollection{*}, then clear sendCollection, and finally send all > the data in {*}midWriteSendCollection{*}. > 5. In the KafkaChannel.maybeCompleteSend() method, determine whether there is > a completed state send in the midWriteSendCollection, and then remove all > completedSends from the midWriteSendCollection, and return completedSends for > adding into Selector.completedSends. > 6. Selector.attemptRead(channel) method execution has preconditions: > `{*}!hasCompletedReceive(channel){*}`, so the KafkaChannel.receive object > does not need to be changed. > 7. A little extra: > ConsumerNetworkClient.poll(Timer timer, PollCondition pollCondition, boolean > disableWakeup) method. > There are three places involved in sending data in this method: > {code:java} > long pollDelayMs = trySend(timer.currentTimeMs()); > -> > client.poll(...) > -> > trySend(timer.currentTimeMs());{code} > There is a problem with this process: > when calling the trySend(...) method for the second time, we should > immediately call client.poll(0, timer.currentTimeMs()); , to ensure that the > send generated each time can be consumed by the next Selector.poll() method. -- This message was sent by Atlassian Jira (v8.20.1#820001)