track work before it gets queued/executed as it is awaiting a free th…

…read for processing

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java

@@ -21,6 +21,7 @@
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 import javax.annotation.concurrent.GuardedBy;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.Monitor;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.Monitor.Guard;
@@ -37,6 +38,7 @@ public class BoundedQueueExecutor {
   private final Monitor monitor = new Monitor();
   private int elementsOutstanding = 0;
   private long bytesOutstanding = 0;
+  private final AtomicLong numQueued;
 
   @GuardedBy("this")
   private int maximumElementsOutstanding;
@@ -61,6 +63,7 @@ public BoundedQueueExecutor(
       long maximumBytesOutstanding,
       ThreadFactory threadFactory) {
     this.maximumPoolSize = maximumPoolSize;
+    this.numQueued = new AtomicLong();
     executor =
         new ThreadPoolExecutor(
             maximumPoolSize,
@@ -99,6 +102,7 @@ protected void afterExecute(Runnable r, Throwable t) {
   // Before adding a Work to the queue, check that there are enough bytes of space or no other
   // outstanding elements of work.
   public void execute(Runnable work, long workBytes) {
+    numQueued.incrementAndGet();
     monitor.enterWhenUninterruptibly(
         new Guard(monitor) {
           @Override
@@ -108,6 +112,7 @@ public boolean isSatisfied() {
                     && elementsOutstanding < maximumElementsOutstanding());
           }
         });
+    numQueued.decrementAndGet();
     executeMonitorHeld(work, workBytes);
   }
 
@@ -195,6 +200,10 @@ public String summaryHtml() {
       builder.append(executor.getActiveCount());
       builder.append("<br>/n");
 
+      builder.append("Queued Work awaiting execution: ");
+      builder.append(numQueued.get());
+      builder.append("<br>/n");
+
       builder.append("Work Queue Size: ");
       builder.append(elementsOutstanding);
       builder.append("/");