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 2022/06/20 11:48:36 UTC

[GitHub] [kafka] soarez opened a new pull request, #12314: KAFKA-13903: Queue size metric in QuorumController

soarez opened a new pull request, #12314:
URL: https://github.com/apache/kafka/pull/12314

   In order to stay in-line with existing queue metrics in ControllerEventManager
   mbean, we need to include a queue size metric in KRaft mode.
   
   The current size of the underlying queue in QuorumController
   needs to be exposed as:
   
   ```
   kafka.controller:type=ControllerEventManager,name=EventQueueSize
   ```
   
   
   ### Committer Checklist (excluded from commit message)
   - [ ] Verify design and implementation 
   - [ ] Verify test coverage and CI build status
   - [ ] Verify documentation (including upgrade notes)
   
   
   [➡ Link to JIRA](https://issues.apache.org/jira/browse/KAFKA-13903)


-- 
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.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] soarez commented on a diff in pull request #12314: KAFKA-13903: Queue size metric in QuorumController

Posted by GitBox <gi...@apache.org>.
soarez commented on code in PR #12314:
URL: https://github.com/apache/kafka/pull/12314#discussion_r902056110


##########
server-common/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java:
##########
@@ -178,15 +190,18 @@ public void run() {
         }
 
         private void remove(EventContext eventContext) {
-            eventContext.remove();
+            boolean removed = eventContext.remove();
             if (eventContext.deadlineNs.isPresent()) {
-                deadlineMap.remove(eventContext.deadlineNs.getAsLong());
+                removed |= deadlineMap.remove(eventContext.deadlineNs.getAsLong()) != null;
                 eventContext.deadlineNs = OptionalLong.empty();
             }
             if (eventContext.tag != null) {
                 tagToEventContext.remove(eventContext.tag, eventContext);
                 eventContext.tag = null;
             }
+            if (removed) {
+                onQueueSizeChange.accept(size.decrementAndGet());

Review Comment:
   The goal isn't to have an arbitrary consumer, we just need to propagate the changes to `ControllerMetrics`. I'm not sure what you mean with "an async explicit metric publish event", but I agree it makes senses that we should be careful while holding the lock. I think we can invert the logic and instead of pushing the size change to `ControllerMetrics`, we can use a Supplier instead to pull it.



-- 
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.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] soarez commented on pull request #12314: KAFKA-13903: Queue size metric in QuorumController

Posted by GitBox <gi...@apache.org>.
soarez commented on PR #12314:
URL: https://github.com/apache/kafka/pull/12314#issuecomment-1195483345

   @mumrah could you take another look? Thanks


-- 
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.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] KAFKA-13903: Queue size metric in QuorumController [kafka]

Posted by "soarez (via GitHub)" <gi...@apache.org>.
soarez closed pull request #12314: KAFKA-13903: Queue size metric in QuorumController
URL: https://github.com/apache/kafka/pull/12314


-- 
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.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] soarez commented on pull request #12314: KAFKA-13903: Queue size metric in QuorumController

Posted by GitBox <gi...@apache.org>.
soarez commented on PR #12314:
URL: https://github.com/apache/kafka/pull/12314#issuecomment-1160920644

   Hi Divij, thanks for having a look at this. The metric [is documented](https://github.com/apache/kafka/blob/89c057397ec5ae320aee7e12e1a984b2c14a7fa5/docs/ops.html#L1424-L1428). The goal is to bring it to KRaft.


-- 
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.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] soarez commented on pull request #12314: KAFKA-13903: Queue size metric in QuorumController

Posted by GitBox <gi...@apache.org>.
soarez commented on PR #12314:
URL: https://github.com/apache/kafka/pull/12314#issuecomment-1181806166

   @mumrah could you review this PR? Thanks


-- 
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.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] soarez commented on pull request #12314: KAFKA-13903: Queue size metric in QuorumController

Posted by GitBox <gi...@apache.org>.
soarez commented on PR #12314:
URL: https://github.com/apache/kafka/pull/12314#issuecomment-1191462380

   @mumrah a couple of the builds succeeded but some failed with a flaky test — `testFencedLeaderRecovery`. There is already a JIRA for it https://issues.apache.org/jira/browse/KAFKA-14093


-- 
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.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] soarez commented on a diff in pull request #12314: KAFKA-13903: Queue size metric in QuorumController

Posted by GitBox <gi...@apache.org>.
soarez commented on code in PR #12314:
URL: https://github.com/apache/kafka/pull/12314#discussion_r902057670


##########
server-common/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java:
##########
@@ -166,6 +171,13 @@ private class EventHandler implements Runnable {
          * A condition variable for waking up the event handler thread.
          */
         private final Condition cond = lock.newCondition();
+        private final Consumer<Integer> onQueueSizeChange;
+
+        private AtomicInteger size = new AtomicInteger(0);

Review Comment:
   There should never be that many events in the queue that can't fit in an int. Each event is a runnable, held in a linked list. Mind that `java.util.LinkedList` for example, uses `int` for size.



-- 
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.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] soarez commented on pull request #12314: KAFKA-13903: Queue size metric in QuorumController

Posted by GitBox <gi...@apache.org>.
soarez commented on PR #12314:
URL: https://github.com/apache/kafka/pull/12314#issuecomment-1185293934

   Thanks for the review @mumrah.  I've applied your suggestion. It feels a bit strange to have the queue size supplier as an argument in `QuorumControllerMetrics`, but I like that the field is now `final`. Please have another look 🙏 


-- 
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.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] divijvaidya commented on a diff in pull request #12314: KAFKA-13903: Queue size metric in QuorumController

Posted by GitBox <gi...@apache.org>.
divijvaidya commented on code in PR #12314:
URL: https://github.com/apache/kafka/pull/12314#discussion_r901646698


##########
server-common/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java:
##########
@@ -166,6 +171,13 @@ private class EventHandler implements Runnable {
          * A condition variable for waking up the event handler thread.
          */
         private final Condition cond = lock.newCondition();
+        private final Consumer<Integer> onQueueSizeChange;
+
+        private AtomicInteger size = new AtomicInteger(0);

Review Comment:
   Long? Is there a reason we are limiting this to an integer?



##########
server-common/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java:
##########
@@ -178,15 +190,18 @@ public void run() {
         }
 
         private void remove(EventContext eventContext) {
-            eventContext.remove();
+            boolean removed = eventContext.remove();
             if (eventContext.deadlineNs.isPresent()) {
-                deadlineMap.remove(eventContext.deadlineNs.getAsLong());
+                removed |= deadlineMap.remove(eventContext.deadlineNs.getAsLong()) != null;
                 eventContext.deadlineNs = OptionalLong.empty();
             }
             if (eventContext.tag != null) {
                 tagToEventContext.remove(eventContext.tag, eventContext);
                 eventContext.tag = null;
             }
+            if (removed) {
+                onQueueSizeChange.accept(size.decrementAndGet());

Review Comment:
   I am a bit sceptical about running an arbitrary consumer in a synchronized block of code (this runs by acquiring a lock). Because the caller can potentially perform expensive things in callback which could cause block the other threads from accessing the lock, hence, block critical execution while performing a side-effect. Instead, could we make it an async explicit metric publish event?



-- 
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.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [kafka] soarez commented on pull request #12314: KAFKA-13903: Queue size metric in QuorumController

Posted by GitBox <gi...@apache.org>.
soarez commented on PR #12314:
URL: https://github.com/apache/kafka/pull/12314#issuecomment-1280064868

   Rebased and simplified the changes. PTAL @mumrah 


-- 
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.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org