junrao commented on a change in pull request #8657: URL: https://github.com/apache/kafka/pull/8657#discussion_r479706303
########## File path: core/src/main/scala/kafka/server/ActionQueue.scala ########## @@ -37,7 +37,10 @@ class ActionQueue { * picks up an action to complete. */ def tryCompleteAction(): Unit = { - val action = queue.poll() - if (action != null) action() + var action = queue.poll() + while (action != null) { + action() Review comment: Perhaps we could do the try/catch of each action here instead of KafkaApis. This way, we are guaranteed that all pending actions are processed in time. ########## File path: core/src/main/scala/kafka/server/ActionQueue.scala ########## @@ -0,0 +1,46 @@ +/** + * 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.server + +import java.util.concurrent.LinkedBlockingDeque + +/** + * This queue is used to collect actions which need to be executed later. One use case is that ReplicaManager#appendRecords + * produces record changes so we need to check and complete delayed requests. In order to avoid conflicting locking, + * we add those actions to this queue and then complete them at the end of KafkaApis.handle() or DelayedJoin.onExpiration. + */ +class ActionQueue { + private val queue = new LinkedBlockingDeque[() => Unit]() + + /** + * add action to this queue. + * @param action action + */ + def add(action: () => Unit): Unit = queue.put(action) + + /** + * try to complete all delayed actions + */ + def tryCompleteActions(): Unit = { + var action = queue.poll() + while (action != null) { Review comment: If we are unlucky, a single thread could be held up in this loop for a long time. Perhaps we could let each thread only complete the number of actions that it sees when entering tryCompleteActions(). ########## File path: core/src/main/scala/kafka/server/ReplicaManager.scala ########## @@ -562,6 +564,10 @@ class ReplicaManager(val config: KafkaConfig, * Append messages to leader replicas of the partition, and wait for them to be replicated to other replicas; * the callback function will be triggered either when timeout or the required acks are satisfied; * if the callback function itself is already synchronized on some object then pass this object to avoid deadlock. + * + * Noted that all pending delayed check operations are stored in a queue. All callers to ReplicaManager.appendRecords() + * are expected to take up to 1 item from that queue and check the completeness for all affected partitions, without Review comment: Since we are draining more than 1 item now, this comment is no longer accurate. ---------------------------------------------------------------- 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: us...@infra.apache.org