[improve][broker] If there is a deadlock in the service, the probe should return a failure because the service may be unavailable

pulsar-broker-common/src/main/java/org/apache/pulsar/common/configuration/VipStatus.java

@@ -47,13 +47,33 @@ public class VipStatus {
     // log a full thread dump when a deadlock is detected in status check once every 10 minutes
     // to prevent excessive logging
     private static final long LOG_THREADDUMP_INTERVAL_WHEN_DEADLOCK_DETECTED = 600000L;
-    private static volatile long threadDumpLoggedTimestamp;
+    private static volatile long lastCheckStatusTimestamp;
+
+    // Since the status endpoint doesn't have authentication, it will be necessary to have a solution to prevent
+    // introducing a new DoS vulnerability where calling the status endpoint in a tight loop could introduce
+    // significant load to the system. One way would be to check that the deadlock check is executed only
+    // when there's more than 1 seconds from the previous check.
+    // If it's less than that, the previous result of the deadlock check would be reused.
+    private static final long DEADLOCK_DETECTED_INTERVAL = 1000L;
+    private static volatile boolean brokerIsHealthy = true;
 
     @Context
     protected ServletContext servletContext;
 
     @GET
     public String checkStatus() {
+        synchronized (VipStatus.class) {
+            if (System.currentTimeMillis() - lastCheckStatusTimestamp < DEADLOCK_DETECTED_INTERVAL) {
+                lastCheckStatusTimestamp = System.currentTimeMillis();
+                if (brokerIsHealthy) {
+                    return "OK";
+                } else {
+                    throw new WebApplicationException(Status.SERVICE_UNAVAILABLE);
+                }
+            }
+            lastCheckStatusTimestamp = System.currentTimeMillis();
+        }
+
         String statusFilePath = (String) servletContext.getAttribute(ATTRIBUTE_STATUS_FILE_PATH);
         @SuppressWarnings("unchecked")
         Supplier<Boolean> isReadyProbe = (Supplier<Boolean>) servletContext.getAttribute(ATTRIBUTE_IS_READY_PROBE);
@@ -73,21 +93,23 @@ public String checkStatus() {
                             .map(threadInfo -> threadInfo.getThreadName()
                                     + "(tid=" + threadInfo.getThreadId() + ")")
                             .collect(Collectors.joining(", "));
-                    if (System.currentTimeMillis() - threadDumpLoggedTimestamp
+                    if (System.currentTimeMillis() - lastCheckStatusTimestamp
                             > LOG_THREADDUMP_INTERVAL_WHEN_DEADLOCK_DETECTED) {
                         String diagnosticResult = ThreadDumpUtil.buildThreadDiagnosticString();
                         log.error("Deadlock detected, service may be unavailable, "
                                 + "thread stack details are as follows: {}.", diagnosticResult);
-                        threadDumpLoggedTimestamp = System.currentTimeMillis();
                     } else {
                         log.error("Deadlocked threads detected. {}", threadNames);
                     }
+                    brokerIsHealthy = false;
                     throw new WebApplicationException(Status.SERVICE_UNAVAILABLE);
                 } else {
+                    brokerIsHealthy = true;
                     return "OK";
                 }
             }
         }
+        brokerIsHealthy = false;
         log.warn("Failed to access \"status.html\". The service is not ready");
         throw new WebApplicationException(Status.NOT_FOUND);
     }