From 1fffc681d28ada66facef2b4766b7729ccbe2416 Mon Sep 17 00:00:00 2001
From: Clint Wylie <cjwylie@gmail.com>
Date: Thu, 1 Feb 2018 08:57:01 -0800
Subject: [PATCH] fix RemoteTaskRunner terminating lazy workers below
 autoscaler minNumWorkers value (#5310)

* fix RemoteTaskRunner terminating lazy workers below autoscaler minNumWorkers value

* add comment
---
 .../indexing/overlord/RemoteTaskRunner.java    |  5 +++++
 .../overlord/RemoteTaskRunnerTest.java         | 18 ++++++++++++++++++
 2 files changed, 23 insertions(+)

diff --git a/indexing-service/src/main/java/io/druid/indexing/overlord/RemoteTaskRunner.java b/indexing-service/src/main/java/io/druid/indexing/overlord/RemoteTaskRunner.java
index 6524a2ac800..e9e623cc801 100644
--- a/indexing-service/src/main/java/io/druid/indexing/overlord/RemoteTaskRunner.java
+++ b/indexing-service/src/main/java/io/druid/indexing/overlord/RemoteTaskRunner.java
@@ -1235,6 +1235,11 @@ public class RemoteTaskRunner implements WorkerTaskRunner, TaskLogStreamer
   @Override
   public Collection<Worker> markWorkersLazy(Predicate<ImmutableWorkerInfo> isLazyWorker, int maxWorkers)
   {
+    // skip the lock and bail early if we should not mark any workers lazy (e.g. number
+    // of current workers is at or below the minNumWorkers of autoscaler config)
+    if (maxWorkers < 1) {
+      return Collections.emptyList();
+    }
     // status lock is used to prevent any tasks being assigned to the worker while we mark it lazy
     synchronized (statusLock) {
       Iterator<String> iterator = zkWorkers.keySet().iterator();
diff --git a/indexing-service/src/test/java/io/druid/indexing/overlord/RemoteTaskRunnerTest.java b/indexing-service/src/test/java/io/druid/indexing/overlord/RemoteTaskRunnerTest.java
index ae57a300f4f..a681c2b8898 100644
--- a/indexing-service/src/test/java/io/druid/indexing/overlord/RemoteTaskRunnerTest.java
+++ b/indexing-service/src/test/java/io/druid/indexing/overlord/RemoteTaskRunnerTest.java
@@ -550,6 +550,24 @@ public class RemoteTaskRunnerTest
     Assert.assertEquals(1, remoteTaskRunner.getLazyWorkers().size());
   }
 
+  @Test
+  public void testFindLazyWorkerNotRunningAnyTaskButWithZeroMaxWorkers() throws Exception
+  {
+    doSetup();
+    Collection<Worker> lazyworkers = remoteTaskRunner.markWorkersLazy(
+        new Predicate<ImmutableWorkerInfo>()
+        {
+          @Override
+          public boolean apply(ImmutableWorkerInfo input)
+          {
+            return true;
+          }
+        }, 0
+    );
+    Assert.assertEquals(0, lazyworkers.size());
+    Assert.assertEquals(0, remoteTaskRunner.getLazyWorkers().size());
+  }
+
   @Test
   public void testWorkerZKReconnect() throws Exception
   {