Do some logging cleanup

Author: Quinn-With-Two-Ns

temporal-sdk/src/main/java/io/temporal/internal/worker/AsyncActivityPollTask.java

@@ -129,6 +129,11 @@ public void cancel(Throwable cause) {
     grpcContext.cancel(cause);
   }
 
+  @Override
+  public String getLabel() {
+    return "AsyncActivityPollTask";
+  }
+
   @Override
   public String toString() {
     return "AsyncActivityPollTask{}";

temporal-sdk/src/main/java/io/temporal/internal/worker/AsyncNexusPollTask.java

@@ -123,6 +123,11 @@ public void cancel(Throwable cause) {
     grpcContext.cancel(cause);
   }
 
+  @Override
+  public String getLabel() {
+    return "AsyncNexusPollTask";
+  }
+
   @Override
   public String toString() {
     return "AsyncNexusPollTask{}";

temporal-sdk/src/main/java/io/temporal/internal/worker/AsyncPoller.java

@@ -34,7 +34,7 @@ final class AsyncPoller<T extends ScalingTask> extends BasePoller<T> {
   private Throttler pollRateThrottler;
   private final Thread.UncaughtExceptionHandler uncaughtExceptionHandler =
       new PollerUncaughtExceptionHandler();
-  private final PollQueueBalancer balancer =
+  private final PollQueueBalancer pollerBalancer =
       new PollQueueBalancer(); // Used to balance the number of slots across pollers
 
   AsyncPoller(
@@ -111,15 +111,15 @@ public boolean start() {
               pollerBehavior.getMaxConcurrentTaskPollers(),
               pollerBehavior.getInitialMaxConcurrentTaskPollers(),
               (newTarget) -> {
-                log.info(
+                log.debug(
                     "Updating maximum number of pollers for {} to: {}",
                     asyncTaskPoller.getLabel(),
                     newTarget);
                 pollerSemaphore.setMaxPermits(newTarget);
               });
       PollQueueTask pollQueue =
           new PollQueueTask(asyncTaskPoller, pollerSemaphore, pollScaleReportHandle);
-      balancer.addPoller(asyncTaskPoller.getLabel());
+      pollerBalancer.addPoller(asyncTaskPoller.getLabel());
       exec.execute(pollQueue);
       exec.scheduleAtFixedRate(pollScaleReportHandle, 0, 100, TimeUnit.MILLISECONDS);
     }
@@ -133,6 +133,7 @@ public CompletableFuture<Void> shutdown(ShutdownManager shutdownManager, boolean
             (f) -> {
               for (PollTaskAsync<T> asyncTaskPoller : asyncTaskPollers) {
                 try {
+                  log.debug("Shutting down async poller: {}", asyncTaskPoller.getLabel());
                   asyncTaskPoller.cancel(new RuntimeException("Shutting down poller"));
                 } catch (Throwable e) {
                   log.error("Error while cancelling poll task", e);
@@ -186,7 +187,6 @@ class PollQueueTask implements Runnable {
 
     @Override
     public void run() {
-      int slotsReserved = 0;
       while (!abort) {
         // Permit to reserve a slot for the poll request.
         SlotPermit permit = null;
@@ -217,7 +217,7 @@ public void run() {
             return;
           }
 
-          balancer.balance(asyncTaskPoller.getLabel());
+          pollerBalancer.balance(asyncTaskPoller.getLabel());
           if (shouldTerminate()) {
             return;
           }
@@ -231,12 +231,8 @@ public void run() {
           }
           permit = BasePoller.getSlotPermitAndHandleInterrupts(future, slotSupplier);
           if (permit == null || shouldTerminate()) {
-            if (permit != null) {
-              slotsReserved++;
-            }
             return;
           }
-          slotsReserved++;
 
           pollerSemaphore.acquire();
           pollerSemaphoreAcquired = true;
@@ -249,12 +245,12 @@ public void run() {
           CompletableFuture<T> pollRequest = asyncTaskPoller.poll(permit);
           // Mark that we have made a poll request
           pollRequestMade = true;
-          balancer.startPoll(asyncTaskPoller.getLabel());
+          pollerBalancer.startPoll(asyncTaskPoller.getLabel());
 
           pollRequest
               .handle(
                   (task, e) -> {
-                    balancer.endPoll(asyncTaskPoller.getLabel());
+                    pollerBalancer.endPoll(asyncTaskPoller.getLabel());
                     if (e instanceof CompletionException) {
                       e = e.getCause();
                     }
@@ -302,18 +298,22 @@ public void run() {
           }
 
           if (shouldTerminate()) {
-            balancer.removePoller(asyncTaskPoller.getLabel());
+            pollerBalancer.removePoller(asyncTaskPoller.getLabel());
             abort = true;
           }
         }
       }
       log.info(
-          "poll loop is terminated: {} - {}",
+          "Poll loop is terminated: {} - {}",
           AsyncPoller.this.getClass().getSimpleName(),
           asyncTaskPoller.getLabel());
     }
   }
 
+  /**
+   * PollQueueBalancer is used to ensure that at least one poller is running for each task. This is
+   * necessary to avoid one poller from consuming all the slots and starving other pollers.
+   */
   @ThreadSafe
   class PollQueueBalancer {
     Map<String, AtomicInteger> taskCounts = new HashMap<>();
@@ -374,8 +374,6 @@ void balance(String p) throws InterruptedException {
             }
           }
           if (!allOtherTasksHavePolls) {
-            log.info(
-                "Waiting for other tasks to have at least one poll request before proceeding with task");
             balancerCondition.await();
           } else {
             return;

temporal-sdk/src/main/java/io/temporal/internal/worker/PollScaleReportHandle.java

@@ -36,7 +36,6 @@ public PollScaleReportHandle(
 
   public synchronized void report(T task, Throwable e) {
     if (e != null) {
-      logger.info("Poller scaling report error: {}.", e.getMessage());
       if ((e instanceof StatusRuntimeException)) {
         StatusRuntimeException statusRuntimeException = (StatusRuntimeException) e;
         if (statusRuntimeException.getStatus().getCode() == Status.Code.RESOURCE_EXHAUSTED) {
@@ -56,7 +55,6 @@ public synchronized void report(T task, Throwable e) {
       ScalingTask.ScalingDecision scalingDecision = task.getScalingDecision();
       everSawScalingDecision = true;
       int deltaSuggestion = scalingDecision.getPollRequestDeltaSuggestion();
-      logger.info("Poller scaling report.  Delta suggestion: {}.", deltaSuggestion);
       if (deltaSuggestion > 0) {
         if (scaleUpAllowed) {
           updateTarget((t -> t + deltaSuggestion));
@@ -68,7 +66,6 @@ public synchronized void report(T task, Throwable e) {
     } else if (task == null && everSawScalingDecision) {
       // We want to avoid scaling down on empty polls if the server has never made any
       // scaling decisions - otherwise we might never scale up again.
-      logger.info("Poller scaling report.  No task.  Scaling down.");
       updateTarget((t) -> t - 1);
     }
   }