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
