diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/WorkerTaskRunner.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/WorkerTaskRunner.java
index c0620f85940..6520de2111f 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/WorkerTaskRunner.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/WorkerTaskRunner.java
@@ -30,6 +30,12 @@ import java.util.Collection;
 @PublicApi
 public interface WorkerTaskRunner extends TaskRunner
 {
+  enum ActionType
+  {
+    ENABLE,
+    DISABLE
+  }
+
   /**
    * List of known workers who can accept tasks for running
    */
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/WorkerTaskRunnerQueryAdapter.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/WorkerTaskRunnerQueryAdapter.java
new file mode 100644
index 00000000000..76567613b77
--- /dev/null
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/WorkerTaskRunnerQueryAdapter.java
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.overlord;
+
+import com.google.common.base.Optional;
+import com.google.common.base.Throwables;
+import com.google.common.collect.Iterables;
+import io.netty.handler.timeout.TimeoutException;
+import org.apache.druid.guice.annotations.EscalatedGlobal;
+import org.apache.druid.indexing.overlord.hrtr.HttpRemoteTaskRunner;
+import org.apache.druid.java.util.common.RE;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.Request;
+import org.apache.druid.java.util.http.client.response.StatusResponseHandler;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.inject.Inject;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.ExecutionException;
+
+public class WorkerTaskRunnerQueryAdapter
+{
+  private static final EmittingLogger log = new EmittingLogger(HttpRemoteTaskRunner.class);
+
+  private final TaskMaster taskMaster;
+  private final HttpClient httpClient;
+
+  @Inject
+  public WorkerTaskRunnerQueryAdapter(TaskMaster taskMaster, @EscalatedGlobal final HttpClient httpClient)
+  {
+    this.taskMaster = taskMaster;
+    this.httpClient = httpClient;
+  }
+
+  public void enableWorker(String host)
+  {
+    sendRequestToWorker(host, WorkerTaskRunner.ActionType.ENABLE);
+  }
+
+  public void disableWorker(String host)
+  {
+    sendRequestToWorker(host, WorkerTaskRunner.ActionType.DISABLE);
+  }
+
+  private void sendRequestToWorker(String workerHost, WorkerTaskRunner.ActionType action)
+  {
+    WorkerTaskRunner workerTaskRunner = getWorkerTaskRunner();
+
+    if (workerTaskRunner == null) {
+      throw new RE("Task Runner does not support enable/disable worker actions");
+    }
+
+    Optional<ImmutableWorkerInfo> workerInfo = Iterables.tryFind(
+        workerTaskRunner.getWorkers(),
+        entry -> entry.getWorker()
+                      .getHost()
+                      .equals(workerHost)
+    );
+
+    if (!workerInfo.isPresent()) {
+      throw new RE(
+          "Worker on host %s does not exists",
+          workerHost
+      );
+    }
+
+    String actionName = WorkerTaskRunner.ActionType.ENABLE.equals(action) ? "enable" : "disable";
+    final URL workerUrl = TaskRunnerUtils.makeWorkerURL(
+        workerInfo.get().getWorker(),
+        "/druid/worker/v1/%s",
+        actionName
+    );
+
+    try {
+      final StatusResponseHolder response = httpClient.go(
+          new Request(HttpMethod.POST, workerUrl),
+          new StatusResponseHandler(StandardCharsets.UTF_8)
+      ).get();
+
+      log.info(
+          "Sent %s action request to worker: %s, status: %s, response: %s",
+          action,
+          workerHost,
+          response.getStatus(),
+          response.getContent()
+      );
+
+      if (!HttpResponseStatus.OK.equals(response.getStatus())) {
+        throw new RE(
+            "Action [%s] failed for worker [%s] with status %s(%s)",
+            action,
+            workerHost,
+            response.getStatus().getCode(),
+            response.getStatus().getReasonPhrase()
+        );
+      }
+    }
+    catch (ExecutionException | InterruptedException | TimeoutException e) {
+      Throwables.propagate(e);
+    }
+  }
+
+  private WorkerTaskRunner getWorkerTaskRunner()
+  {
+    Optional<TaskRunner> taskRunnerOpt = taskMaster.getTaskRunner();
+    if (taskRunnerOpt.isPresent() && taskRunnerOpt.get() instanceof WorkerTaskRunner) {
+      return (WorkerTaskRunner) taskRunnerOpt.get();
+    } else {
+      return null;
+    }
+  }
+}
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordResource.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordResource.java
index e5abac4070e..0efb5a5f22a 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordResource.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordResource.java
@@ -50,6 +50,7 @@ import org.apache.druid.indexing.overlord.TaskRunner;
 import org.apache.druid.indexing.overlord.TaskRunnerWorkItem;
 import org.apache.druid.indexing.overlord.TaskStorageQueryAdapter;
 import org.apache.druid.indexing.overlord.WorkerTaskRunner;
+import org.apache.druid.indexing.overlord.WorkerTaskRunnerQueryAdapter;
 import org.apache.druid.indexing.overlord.autoscaling.ScalingStats;
 import org.apache.druid.indexing.overlord.http.security.TaskResourceFilter;
 import org.apache.druid.indexing.overlord.setup.WorkerBehaviorConfig;
@@ -104,6 +105,7 @@ import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 
 /**
+ *
  */
 @Path("/druid/indexer/v1")
 public class OverlordResource
@@ -117,6 +119,7 @@ public class OverlordResource
   private final JacksonConfigManager configManager;
   private final AuditManager auditManager;
   private final AuthorizerMapper authorizerMapper;
+  private final WorkerTaskRunnerQueryAdapter workerTaskRunnerQueryAdapter;
 
   private AtomicReference<WorkerBehaviorConfig> workerConfigRef = null;
   private static final List API_TASK_STATES = ImmutableList.of("pending", "waiting", "running", "complete");
@@ -129,7 +132,8 @@ public class OverlordResource
       TaskLogStreamer taskLogStreamer,
       JacksonConfigManager configManager,
       AuditManager auditManager,
-      AuthorizerMapper authorizerMapper
+      AuthorizerMapper authorizerMapper,
+      WorkerTaskRunnerQueryAdapter workerTaskRunnerQueryAdapter
   )
   {
     this.taskMaster = taskMaster;
@@ -139,6 +143,7 @@ public class OverlordResource
     this.configManager = configManager;
     this.auditManager = auditManager;
     this.authorizerMapper = authorizerMapper;
+    this.workerTaskRunnerQueryAdapter = workerTaskRunnerQueryAdapter;
   }
 
   @POST
@@ -726,6 +731,47 @@ public class OverlordResource
     );
   }
 
+  @POST
+  @Path("/worker/{host}/enable")
+  @Produces(MediaType.APPLICATION_JSON)
+  @ResourceFilters(StateResourceFilter.class)
+  public Response enableWorker(@PathParam("host") final String host)
+  {
+    return changeWorkerStatus(host, WorkerTaskRunner.ActionType.ENABLE);
+  }
+
+  @POST
+  @Path("/worker/{host}/disable")
+  @Produces(MediaType.APPLICATION_JSON)
+  @ResourceFilters(StateResourceFilter.class)
+  public Response disableWorker(@PathParam("host") final String host)
+  {
+    return changeWorkerStatus(host, WorkerTaskRunner.ActionType.DISABLE);
+  }
+
+  private Response changeWorkerStatus(String host, WorkerTaskRunner.ActionType action)
+  {
+    try {
+      if (WorkerTaskRunner.ActionType.DISABLE.equals(action)) {
+        workerTaskRunnerQueryAdapter.disableWorker(host);
+        return Response.ok(ImmutableMap.of(host, "disabled")).build();
+      } else if (WorkerTaskRunner.ActionType.ENABLE.equals(action)) {
+        workerTaskRunnerQueryAdapter.enableWorker(host);
+        return Response.ok(ImmutableMap.of(host, "enabled")).build();
+      } else {
+        return Response.serverError()
+                       .entity(ImmutableMap.of("error", "Worker does not support " + action + " action!"))
+                       .build();
+      }
+    }
+    catch (Exception e) {
+      log.error(e, "Error in posting [%s] action to [%s]", action, host);
+      return Response.serverError()
+                     .entity(ImmutableMap.of("error", e.getMessage()))
+                     .build();
+    }
+  }
+
   @GET
   @Path("/scaling")
   @Produces(MediaType.APPLICATION_JSON)
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/WorkerTaskRunnerQueryAdpaterTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/WorkerTaskRunnerQueryAdpaterTest.java
new file mode 100644
index 00000000000..330ff2c8b7e
--- /dev/null
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/WorkerTaskRunnerQueryAdpaterTest.java
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.overlord;
+
+import com.google.common.base.Optional;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.util.concurrent.SettableFuture;
+import org.apache.druid.indexing.worker.Worker;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.RE;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.Request;
+import org.apache.druid.java.util.http.client.response.HttpResponseHandler;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.easymock.Capture;
+import org.easymock.EasyMock;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.net.URL;
+
+
+public class WorkerTaskRunnerQueryAdpaterTest
+{
+  private WorkerTaskRunnerQueryAdapter workerTaskRunnerQueryAdapter;
+  private HttpClient httpClient;
+  private WorkerTaskRunner workerTaskRunner;
+  private TaskMaster taskMaster;
+
+  @Before
+  public void setup()
+  {
+    httpClient = EasyMock.createNiceMock(HttpClient.class);
+    workerTaskRunner = EasyMock.createMock(WorkerTaskRunner.class);
+    taskMaster = EasyMock.createStrictMock(TaskMaster.class);
+
+    workerTaskRunnerQueryAdapter = new WorkerTaskRunnerQueryAdapter(taskMaster, httpClient);
+
+    EasyMock.expect(taskMaster.getTaskRunner()).andReturn(
+        Optional.of(workerTaskRunner)
+    ).once();
+
+    EasyMock.expect(workerTaskRunner.getWorkers()).andReturn(
+        ImmutableList.of(
+            new ImmutableWorkerInfo(
+                new Worker(
+                    "http", "worker-host1", "192.0.0.1", 10, "v1"
+                ),
+                2,
+                ImmutableSet.of("grp1", "grp2"),
+                ImmutableSet.of("task1", "task2"),
+                DateTimes.of("2015-01-01T01:01:01Z")
+            ),
+            new ImmutableWorkerInfo(
+                new Worker(
+                    "https", "worker-host2", "192.0.0.2", 4, "v1"
+                ),
+                1,
+                ImmutableSet.of("grp1"),
+                ImmutableSet.of("task1"),
+                DateTimes.of("2015-01-01T01:01:01Z")
+            )
+        )
+    ).once();
+  }
+
+  @After
+  public void tearDown()
+  {
+    EasyMock.verify(workerTaskRunner, taskMaster, httpClient);
+  }
+
+  @Test
+  public void testDisableWorker() throws Exception
+  {
+    final URL workerUrl = new URL("http://worker-host1/druid/worker/v1/disable");
+    final String workerResponse = "{\"worker-host1\":\"disabled\"}";
+    Capture<Request> capturedRequest = getHttpClientRequestCapture(HttpResponseStatus.OK, workerResponse);
+
+    EasyMock.replay(workerTaskRunner, taskMaster, httpClient);
+
+    workerTaskRunnerQueryAdapter.disableWorker("worker-host1");
+
+    Assert.assertEquals(HttpMethod.POST, capturedRequest.getValue().getMethod());
+    Assert.assertEquals(workerUrl, capturedRequest.getValue().getUrl());
+  }
+
+  @Test
+  public void testDisableWorkerWhenWorkerRaisesError() throws Exception
+  {
+    final URL workerUrl = new URL("http://worker-host1/druid/worker/v1/disable");
+    Capture<Request> capturedRequest = getHttpClientRequestCapture(HttpResponseStatus.INTERNAL_SERVER_ERROR, "");
+
+    EasyMock.replay(workerTaskRunner, taskMaster, httpClient);
+
+    try {
+      workerTaskRunnerQueryAdapter.disableWorker("worker-host1");
+      Assert.fail("Should raise RE exception!");
+    }
+    catch (RE re) {
+    }
+
+    Assert.assertEquals(HttpMethod.POST, capturedRequest.getValue().getMethod());
+    Assert.assertEquals(workerUrl, capturedRequest.getValue().getUrl());
+  }
+
+  @Test(expected = RE.class)
+  public void testDisableWorkerWhenWorkerNotExists()
+  {
+    EasyMock.replay(workerTaskRunner, taskMaster, httpClient);
+
+    workerTaskRunnerQueryAdapter.disableWorker("not-existing-worker");
+  }
+
+  @Test
+  public void testEnableWorker() throws Exception
+  {
+    final URL workerUrl = new URL("https://worker-host2/druid/worker/v1/enable");
+    final String workerResponse = "{\"worker-host2\":\"enabled\"}";
+    Capture<Request> capturedRequest = getHttpClientRequestCapture(HttpResponseStatus.OK, workerResponse);
+
+    EasyMock.replay(workerTaskRunner, taskMaster, httpClient);
+
+    workerTaskRunnerQueryAdapter.enableWorker("worker-host2");
+
+    Assert.assertEquals(HttpMethod.POST, capturedRequest.getValue().getMethod());
+    Assert.assertEquals(workerUrl, capturedRequest.getValue().getUrl());
+  }
+
+  @Test
+  public void testEnableWorkerWhenWorkerRaisesError() throws Exception
+  {
+    final URL workerUrl = new URL("https://worker-host2/druid/worker/v1/enable");
+    Capture<Request> capturedRequest = getHttpClientRequestCapture(HttpResponseStatus.INTERNAL_SERVER_ERROR, "");
+
+    EasyMock.replay(workerTaskRunner, taskMaster, httpClient);
+
+    try {
+      workerTaskRunnerQueryAdapter.enableWorker("worker-host2");
+      Assert.fail("Should raise RE exception!");
+    }
+    catch (RE re) {
+    }
+
+    Assert.assertEquals(HttpMethod.POST, capturedRequest.getValue().getMethod());
+    Assert.assertEquals(workerUrl, capturedRequest.getValue().getUrl());
+  }
+
+  @Test(expected = RE.class)
+  public void testEnableWorkerWhenWorkerNotExists()
+  {
+    EasyMock.replay(workerTaskRunner, taskMaster, httpClient);
+
+    workerTaskRunnerQueryAdapter.enableWorker("not-existing-worker");
+  }
+
+  private Capture<Request> getHttpClientRequestCapture(HttpResponseStatus httpStatus, String responseContent)
+  {
+    SettableFuture<StatusResponseHolder> futureResult = SettableFuture.create();
+    futureResult.set(
+        new StatusResponseHolder(httpStatus, new StringBuilder(responseContent))
+    );
+    Capture<Request> capturedRequest = EasyMock.newCapture();
+    EasyMock.expect(
+        httpClient.go(
+            EasyMock.capture(capturedRequest),
+            EasyMock.<HttpResponseHandler>anyObject()
+        )
+    )
+            .andReturn(futureResult)
+            .once();
+
+    return capturedRequest;
+  }
+}
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordResourceTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordResourceTest.java
index 89824b046f3..3a58c0cf627 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordResourceTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordResourceTest.java
@@ -40,7 +40,9 @@ import org.apache.druid.indexing.overlord.TaskQueue;
 import org.apache.druid.indexing.overlord.TaskRunner;
 import org.apache.druid.indexing.overlord.TaskRunnerWorkItem;
 import org.apache.druid.indexing.overlord.TaskStorageQueryAdapter;
+import org.apache.druid.indexing.overlord.WorkerTaskRunnerQueryAdapter;
 import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.RE;
 import org.apache.druid.segment.TestHelper;
 import org.apache.druid.server.security.Access;
 import org.apache.druid.server.security.Action;
@@ -51,6 +53,7 @@ import org.apache.druid.server.security.AuthorizerMapper;
 import org.apache.druid.server.security.ForbiddenException;
 import org.apache.druid.server.security.Resource;
 import org.easymock.EasyMock;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
 import org.joda.time.DateTime;
 import org.joda.time.Duration;
 import org.joda.time.Interval;
@@ -79,6 +82,7 @@ public class OverlordResourceTest
   private IndexerMetadataStorageAdapter indexerMetadataStorageAdapter;
   private HttpServletRequest req;
   private TaskRunner taskRunner;
+  private WorkerTaskRunnerQueryAdapter workerTaskRunnerQueryAdapter;
 
   @Rule
   public ExpectedException expectedException = ExpectedException.none();
@@ -91,6 +95,7 @@ public class OverlordResourceTest
     taskStorageQueryAdapter = EasyMock.createStrictMock(TaskStorageQueryAdapter.class);
     indexerMetadataStorageAdapter = EasyMock.createStrictMock(IndexerMetadataStorageAdapter.class);
     req = EasyMock.createStrictMock(HttpServletRequest.class);
+    workerTaskRunnerQueryAdapter = EasyMock.createStrictMock(WorkerTaskRunnerQueryAdapter.class);
 
     EasyMock.expect(taskMaster.getTaskRunner()).andReturn(
         Optional.of(taskRunner)
@@ -124,21 +129,36 @@ public class OverlordResourceTest
         null,
         null,
         null,
-        authMapper
+        authMapper,
+        workerTaskRunnerQueryAdapter
     );
   }
 
   @After
   public void tearDown()
   {
-    EasyMock.verify(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
+    EasyMock.verify(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
   }
 
   @Test
   public void testLeader()
   {
     EasyMock.expect(taskMaster.getCurrentLeader()).andReturn("boz").once();
-    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
 
     final Response response = overlordResource.getLeader();
     Assert.assertEquals("boz", response.getEntity());
@@ -150,7 +170,14 @@ public class OverlordResourceTest
   {
     EasyMock.expect(taskMaster.isLeader()).andReturn(true).once();
     EasyMock.expect(taskMaster.isLeader()).andReturn(false).once();
-    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
 
     // true
     final Response response1 = overlordResource.isLeader();
@@ -207,7 +234,14 @@ public class OverlordResourceTest
         )
     );
 
-    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
 
     List<TaskStatusPlus> responseObjects = (List<TaskStatusPlus>) overlordResource.getWaitingTasks(req)
                                                                                   .getEntity();
@@ -224,7 +258,8 @@ public class OverlordResourceTest
         ImmutableList.of(
             new MockTaskRunnerWorkItem(tasksIds.get(0), null),
             new MockTaskRunnerWorkItem(tasksIds.get(1), null),
-            new MockTaskRunnerWorkItem(tasksIds.get(2), null)));
+            new MockTaskRunnerWorkItem(tasksIds.get(2), null)
+        ));
 
     EasyMock.expect(taskStorageQueryAdapter.getCompletedTaskInfoByCreatedTimeDuration(null, null, null)).andStubReturn(
         ImmutableList.of(
@@ -251,11 +286,18 @@ public class OverlordResourceTest
             )
         )
     );
-    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
     Assert.assertTrue(taskStorageQueryAdapter.getCompletedTaskInfoByCreatedTimeDuration(null, null, null).size() == 3);
     Assert.assertTrue(taskRunner.getRunningTasks().size() == 3);
     List<TaskStatusPlus> responseObjects = (List) overlordResource
-          .getCompleteTasks(null, req).getEntity();
+        .getCompleteTasks(null, req).getEntity();
 
     Assert.assertEquals(2, responseObjects.size());
     Assert.assertEquals(tasksIds.get(1), responseObjects.get(0).getId());
@@ -292,7 +334,14 @@ public class OverlordResourceTest
         )
     );
 
-    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
 
     List<TaskStatusPlus> responseObjects = (List) overlordResource.getRunningTasks(null, req)
                                                                   .getEntity();
@@ -384,7 +433,14 @@ public class OverlordResourceTest
         )
     );
 
-    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
     List<TaskStatusPlus> responseObjects = (List<TaskStatusPlus>) overlordResource
         .getTasks(null, null, null, null, null, req)
         .getEntity();
@@ -396,31 +452,32 @@ public class OverlordResourceTest
   {
     expectAuthorizationTokenCheck();
     //completed tasks
-    EasyMock.expect(taskStorageQueryAdapter.getCompletedTaskInfoByCreatedTimeDuration(null, null, "allow")).andStubReturn(
-        ImmutableList.of(
-            new TaskInfo(
-                "id_5",
-                DateTime.now(ISOChronology.getInstanceUTC()),
-                TaskStatus.success("id_5"),
-                "allow",
-                getTaskWithIdAndDatasource("id_5", "allow")
-            ),
-            new TaskInfo(
-                "id_6",
-                DateTime.now(ISOChronology.getInstanceUTC()),
-                TaskStatus.success("id_6"),
-                "allow",
-                getTaskWithIdAndDatasource("id_6", "allow")
-            ),
-            new TaskInfo(
-                "id_7",
-                DateTime.now(ISOChronology.getInstanceUTC()),
-                TaskStatus.success("id_7"),
-                "allow",
-                getTaskWithIdAndDatasource("id_7", "allow")
+    EasyMock.expect(taskStorageQueryAdapter.getCompletedTaskInfoByCreatedTimeDuration(null, null, "allow"))
+        .andStubReturn(
+            ImmutableList.of(
+                new TaskInfo(
+                    "id_5",
+                    DateTime.now(ISOChronology.getInstanceUTC()),
+                    TaskStatus.success("id_5"),
+                    "allow",
+                    getTaskWithIdAndDatasource("id_5", "allow")
+                ),
+                new TaskInfo(
+                    "id_6",
+                    DateTime.now(ISOChronology.getInstanceUTC()),
+                    TaskStatus.success("id_6"),
+                    "allow",
+                    getTaskWithIdAndDatasource("id_6", "allow")
+                ),
+                new TaskInfo(
+                    "id_7",
+                    DateTime.now(ISOChronology.getInstanceUTC()),
+                    TaskStatus.success("id_7"),
+                    "allow",
+                    getTaskWithIdAndDatasource("id_7", "allow")
+                )
             )
-        )
-    );
+      );
     //active tasks
     EasyMock.expect(taskStorageQueryAdapter.getActiveTaskInfo("allow")).andStubReturn(
         ImmutableList.of(
@@ -471,7 +528,14 @@ public class OverlordResourceTest
             new MockTaskRunnerWorkItem("id_1", null)
         )
     );
-    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
 
     List<TaskStatusPlus> responseObjects = (List<TaskStatusPlus>) overlordResource
         .getTasks(null, "allow", null, null, null, req)
@@ -526,7 +590,14 @@ public class OverlordResourceTest
         )
     );
 
-    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
     List<TaskStatusPlus> responseObjects = (List<TaskStatusPlus>) overlordResource
         .getTasks(
             "waiting",
@@ -586,7 +657,14 @@ public class OverlordResourceTest
     );
 
 
-    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
 
     List<TaskStatusPlus> responseObjects = (List) overlordResource
         .getTasks("running", "allow", null, null, null, req)
@@ -644,7 +722,14 @@ public class OverlordResourceTest
     );
 
 
-    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
 
     List<TaskStatusPlus> responseObjects = (List<TaskStatusPlus>) overlordResource
         .getTasks("pending", null, null, null, null, req)
@@ -685,7 +770,14 @@ public class OverlordResourceTest
             )
         )
     );
-    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
     List<TaskStatusPlus> responseObjects = (List<TaskStatusPlus>) overlordResource
         .getTasks("complete", null, null, null, null, req)
         .getEntity();
@@ -700,33 +792,41 @@ public class OverlordResourceTest
     expectAuthorizationTokenCheck();
     List<String> tasksIds = ImmutableList.of("id_1", "id_2", "id_3");
     Duration duration = new Period("PT86400S").toStandardDuration();
-    EasyMock.expect(taskStorageQueryAdapter.getCompletedTaskInfoByCreatedTimeDuration(null, duration, null)).andStubReturn(
-        ImmutableList.of(
-            new TaskInfo(
-                "id_1",
-                DateTime.now(ISOChronology.getInstanceUTC()),
-                TaskStatus.success("id_1"),
-                "deny",
-                getTaskWithIdAndDatasource("id_1", "deny")
-            ),
-            new TaskInfo(
-                "id_2",
-                DateTime.now(ISOChronology.getInstanceUTC()),
-                TaskStatus.success("id_2"),
-                "allow",
-                getTaskWithIdAndDatasource("id_2", "allow")
-            ),
-            new TaskInfo(
-                "id_3",
-                DateTime.now(ISOChronology.getInstanceUTC()),
-                TaskStatus.success("id_3"),
-                "allow",
-                getTaskWithIdAndDatasource("id_3", "allow")
+    EasyMock.expect(taskStorageQueryAdapter.getCompletedTaskInfoByCreatedTimeDuration(null, duration, null))
+        .andStubReturn(
+            ImmutableList.of(
+                new TaskInfo(
+                    "id_1",
+                    DateTime.now(ISOChronology.getInstanceUTC()),
+                    TaskStatus.success("id_1"),
+                    "deny",
+                    getTaskWithIdAndDatasource("id_1", "deny")
+                ),
+                new TaskInfo(
+                    "id_2",
+                    DateTime.now(ISOChronology.getInstanceUTC()),
+                    TaskStatus.success("id_2"),
+                    "allow",
+                    getTaskWithIdAndDatasource("id_2", "allow")
+                ),
+                new TaskInfo(
+                    "id_3",
+                    DateTime.now(ISOChronology.getInstanceUTC()),
+                    TaskStatus.success("id_3"),
+                    "allow",
+                    getTaskWithIdAndDatasource("id_3", "allow")
+                )
             )
-        )
-    );
+      );
 
-    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
     String interval = "2010-01-01_P1D";
     List<TaskStatusPlus> responseObjects = (List<TaskStatusPlus>) overlordResource
         .getTasks("complete", null, interval, null, null, req)
@@ -765,7 +865,14 @@ public class OverlordResourceTest
             )
         )
     );
-    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
     List<TaskStatusPlus> responseObjects = (List<TaskStatusPlus>) overlordResource
         .getTasks("complete", null, null, null, null, req)
         .getEntity();
@@ -779,7 +886,14 @@ public class OverlordResourceTest
   @Test
   public void testGetTasksNegativeState()
   {
-    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
     Object responseObject = overlordResource
         .getTasks("blah", "ds_test", null, null, null, req)
         .getEntity();
@@ -795,7 +909,14 @@ public class OverlordResourceTest
     expectedException.expect(ForbiddenException.class);
     expectAuthorizationTokenCheck();
 
-    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
     Task task = NoopTask.create();
     overlordResource.taskPost(task, req);
   }
@@ -815,7 +936,14 @@ public class OverlordResourceTest
         )
         .andReturn(2);
 
-    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
 
     final Map<String, Integer> response = (Map<String, Integer>) overlordResource
         .killPendingSegments("allow", new Interval(DateTimes.MIN, DateTimes.nowUtc()).toString(), req)
@@ -837,7 +965,14 @@ public class OverlordResourceTest
     EasyMock.expect(taskStorageQueryAdapter.getTask("othertask"))
             .andReturn(Optional.absent());
 
-    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
 
     final Response response1 = overlordResource.getTaskPayload("mytask");
     final TaskPayloadResponse taskPayloadResponse1 = TestHelper.makeJsonMapper().readValue(
@@ -873,7 +1008,14 @@ public class OverlordResourceTest
     EasyMock.<Collection<? extends TaskRunnerWorkItem>>expect(taskRunner.getKnownTasks())
         .andReturn(ImmutableList.of());
 
-    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
 
     final Response response1 = overlordResource.getTaskStatus("mytask");
     final TaskStatusResponse taskStatusResponse1 = TestHelper.makeJsonMapper().readValue(
@@ -926,7 +1068,15 @@ public class OverlordResourceTest
     mockQueue.shutdown("id_1", "Shutdown request from user");
     EasyMock.expectLastCall();
 
-    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req, mockQueue);
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        mockQueue,
+        workerTaskRunnerQueryAdapter
+    );
 
     final Map<String, Integer> response = (Map<String, Integer>) overlordResource
         .doShutdown("id_1")
@@ -969,7 +1119,15 @@ public class OverlordResourceTest
     mockQueue.shutdown("id_2", "Shutdown request from user");
     EasyMock.expectLastCall();
 
-    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req, mockQueue);
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        mockQueue,
+        workerTaskRunnerQueryAdapter
+    );
 
     final Map<String, Integer> response = (Map<String, Integer>) overlordResource
         .shutdownTasksForDataSource("datasource")
@@ -984,12 +1142,111 @@ public class OverlordResourceTest
     EasyMock.expect(taskMaster.isLeader()).andReturn(true).anyTimes();
     EasyMock.expect(taskMaster.getTaskQueue()).andReturn(Optional.of(taskQueue)).anyTimes();
     EasyMock.expect(taskStorageQueryAdapter.getActiveTaskInfo(EasyMock.anyString())).andReturn(Collections.emptyList());
-    EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
 
     final Response response = overlordResource.shutdownTasksForDataSource("notExisting");
     Assert.assertEquals(Status.NOT_FOUND.getStatusCode(), response.getStatus());
   }
 
+  @Test
+  public void testEnableWorker()
+  {
+    final String host = "worker-host";
+
+    workerTaskRunnerQueryAdapter.enableWorker(host);
+    EasyMock.expectLastCall().once();
+
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
+
+    final Response response = overlordResource.enableWorker(host);
+
+    Assert.assertEquals(HttpResponseStatus.OK.getCode(), response.getStatus());
+    Assert.assertEquals(ImmutableMap.of(host, "enabled"), response.getEntity());
+  }
+
+  @Test
+  public void testDisableWorker()
+  {
+    final String host = "worker-host";
+
+    workerTaskRunnerQueryAdapter.disableWorker(host);
+    EasyMock.expectLastCall().once();
+
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
+
+    final Response response = overlordResource.disableWorker(host);
+
+    Assert.assertEquals(HttpResponseStatus.OK.getCode(), response.getStatus());
+    Assert.assertEquals(ImmutableMap.of(host, "disabled"), response.getEntity());
+  }
+
+  @Test
+  public void testEnableWorkerWhenWorkerAPIRaisesError()
+  {
+    final String host = "worker-host";
+
+    workerTaskRunnerQueryAdapter.enableWorker(host);
+    EasyMock.expectLastCall().andThrow(new RE("Worker API returns error!")).once();
+
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
+
+    final Response response = overlordResource.enableWorker(host);
+
+    Assert.assertEquals(HttpResponseStatus.INTERNAL_SERVER_ERROR.getCode(), response.getStatus());
+    Assert.assertEquals(ImmutableMap.of("error", "Worker API returns error!"), response.getEntity());
+  }
+
+  @Test
+  public void testDisableWorkerWhenWorkerAPIRaisesError()
+  {
+    final String host = "worker-host";
+
+    workerTaskRunnerQueryAdapter.disableWorker(host);
+    EasyMock.expectLastCall().andThrow(new RE("Worker API returns error!")).once();
+
+    EasyMock.replay(
+        taskRunner,
+        taskMaster,
+        taskStorageQueryAdapter,
+        indexerMetadataStorageAdapter,
+        req,
+        workerTaskRunnerQueryAdapter
+    );
+
+    final Response response = overlordResource.disableWorker(host);
+
+    Assert.assertEquals(HttpResponseStatus.INTERNAL_SERVER_ERROR.getCode(), response.getStatus());
+    Assert.assertEquals(ImmutableMap.of("error", "Worker API returns error!"), response.getEntity());
+  }
+
   private void expectAuthorizationTokenCheck()
   {
     AuthenticationResult authenticationResult = new AuthenticationResult("druid", "druid", null, null);
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordTest.java
index 6b34ea4da71..2503014c204 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordTest.java
@@ -52,6 +52,7 @@ import org.apache.druid.indexing.overlord.TaskRunnerListener;
 import org.apache.druid.indexing.overlord.TaskRunnerWorkItem;
 import org.apache.druid.indexing.overlord.TaskStorage;
 import org.apache.druid.indexing.overlord.TaskStorageQueryAdapter;
+import org.apache.druid.indexing.overlord.WorkerTaskRunnerQueryAdapter;
 import org.apache.druid.indexing.overlord.autoscaling.ScalingStats;
 import org.apache.druid.indexing.overlord.config.TaskQueueConfig;
 import org.apache.druid.indexing.overlord.helpers.OverlordHelperManager;
@@ -212,6 +213,7 @@ public class OverlordTest
     Assert.assertEquals(taskMaster.getCurrentLeader(), druidNode.getHostAndPort());
 
     final TaskStorageQueryAdapter taskStorageQueryAdapter = new TaskStorageQueryAdapter(taskStorage);
+    final WorkerTaskRunnerQueryAdapter workerTaskRunnerQueryAdapter = new WorkerTaskRunnerQueryAdapter(taskMaster, null);
     // Test Overlord resource stuff
     overlordResource = new OverlordResource(
         taskMaster,
@@ -220,7 +222,8 @@ public class OverlordTest
         null,
         null,
         null,
-        AuthTestUtils.TEST_AUTHORIZER_MAPPER
+        AuthTestUtils.TEST_AUTHORIZER_MAPPER,
+        workerTaskRunnerQueryAdapter
     );
     Response response = overlordResource.getLeader();
     Assert.assertEquals(druidNode.getHostAndPort(), response.getEntity());
diff --git a/web-console/README.md b/web-console/README.md
index ad02c103431..8ea53e85abf 100644
--- a/web-console/README.md
+++ b/web-console/README.md
@@ -55,6 +55,7 @@ Generated/copied dynamically
 ```
 GET /status
 GET /druid/indexer/v1/supervisor?full
+POST /druid/indexer/v1/worker
 GET /druid/indexer/v1/workers
 GET /druid/coordinator/v1/loadqueue?simple
 GET /druid/coordinator/v1/config
diff --git a/web-console/src/views/segments-view/__snapshots__/segments-view.spec.tsx.snap b/web-console/src/views/segments-view/__snapshots__/segments-view.spec.tsx.snap
index 339d63958e9..ee2f486356e 100644
--- a/web-console/src/views/segments-view/__snapshots__/segments-view.spec.tsx.snap
+++ b/web-console/src/views/segments-view/__snapshots__/segments-view.spec.tsx.snap
@@ -2,10 +2,10 @@
 
 exports[`describe segments-view segments view snapshot 1`] = `
 <div
-  className="servers-view app-view"
+  className="segments-view app-view"
 >
   <ViewControlBar
-    label="Historicals"
+    label="Segments"
   >
     <Blueprint3.Button
       icon="refresh"
@@ -13,256 +13,27 @@ exports[`describe segments-view segments view snapshot 1`] = `
       text="Refresh"
     />
     <Blueprint3.Button
+      hidden={false}
       icon="application"
       onClick={[Function]}
       text="Go to SQL"
     />
-    <Blueprint3.Switch
-      checked={false}
-      label="Group by tier"
-      onChange={[Function]}
-    />
     <TableColumnSelection
       columns={
         Array [
-          "Server",
-          "Tier",
-          "Curr size",
-          "Max size",
-          "Usage",
-          "Load/drop queues",
-          "Host",
-          "Port",
-        ]
-      }
-      onChange={[Function]}
-      tableColumnsHidden={Array []}
-    />
-  </ViewControlBar>
-  <ReactTable
-    AggregatedComponent={[Function]}
-    ExpanderComponent={[Function]}
-    FilterComponent={[Function]}
-    LoadingComponent={[Function]}
-    NoDataComponent={[Function]}
-    PadRowComponent={[Function]}
-    PaginationComponent={[Function]}
-    PivotValueComponent={[Function]}
-    ResizerComponent={[Function]}
-    TableComponent={[Function]}
-    TbodyComponent={[Function]}
-    TdComponent={[Function]}
-    TfootComponent={[Function]}
-    ThComponent={[Function]}
-    TheadComponent={[Function]}
-    TrComponent={[Function]}
-    TrGroupComponent={[Function]}
-    aggregatedKey="_aggregated"
-    className="-striped -highlight"
-    collapseOnDataChange={true}
-    collapseOnPageChange={true}
-    collapseOnSortingChange={true}
-    column={
-      Object {
-        "Aggregated": undefined,
-        "Cell": undefined,
-        "Expander": undefined,
-        "Filter": undefined,
-        "Footer": undefined,
-        "Header": undefined,
-        "Pivot": undefined,
-        "PivotValue": undefined,
-        "aggregate": undefined,
-        "className": "",
-        "filterAll": false,
-        "filterMethod": undefined,
-        "filterable": undefined,
-        "footerClassName": "",
-        "footerStyle": Object {},
-        "getFooterProps": [Function],
-        "getHeaderProps": [Function],
-        "getProps": [Function],
-        "headerClassName": "",
-        "headerStyle": Object {},
-        "minWidth": 100,
-        "resizable": undefined,
-        "show": true,
-        "sortMethod": undefined,
-        "sortable": undefined,
-        "style": Object {},
-      }
-    }
-    columns={
-      Array [
-        Object {
-          "Aggregated": [Function],
-          "Header": "Server",
-          "accessor": "server",
-          "show": true,
-          "width": 300,
-        },
-        Object {
-          "Cell": [Function],
-          "Header": "Tier",
-          "accessor": "tier",
-          "show": true,
-        },
-        Object {
-          "Aggregated": [Function],
-          "Cell": [Function],
-          "Header": "Curr size",
-          "accessor": "curr_size",
-          "filterable": false,
-          "id": "curr_size",
-          "show": true,
-          "width": 100,
-        },
-        Object {
-          "Aggregated": [Function],
-          "Cell": [Function],
-          "Header": "Max size",
-          "accessor": "max_size",
-          "filterable": false,
-          "id": "max_size",
-          "show": true,
-          "width": 100,
-        },
-        Object {
-          "Aggregated": [Function],
-          "Cell": [Function],
-          "Header": "Usage",
-          "accessor": [Function],
-          "filterable": false,
-          "id": "usage",
-          "show": true,
-          "width": 100,
-        },
-        Object {
-          "Aggregated": [Function],
-          "Cell": [Function],
-          "Header": "Load/drop queues",
-          "accessor": [Function],
-          "filterable": false,
-          "id": "queue",
-          "show": true,
-          "width": 400,
-        },
-        Object {
-          "Aggregated": [Function],
-          "Header": "Host",
-          "accessor": "host",
-          "show": true,
-        },
-        Object {
-          "Aggregated": [Function],
-          "Header": "Port",
-          "accessor": [Function],
-          "id": "port",
-          "show": true,
-        },
-      ]
-    }
-    data={Array []}
-    defaultExpanded={Object {}}
-    defaultFilterMethod={[Function]}
-    defaultFiltered={Array []}
-    defaultPageSize={10}
-    defaultResized={Array []}
-    defaultSortDesc={false}
-    defaultSortMethod={[Function]}
-    defaultSorted={Array []}
-    expanderDefaults={
-      Object {
-        "filterable": false,
-        "resizable": false,
-        "sortable": false,
-        "width": 35,
-      }
-    }
-    filterable={true}
-    filtered={Array []}
-    freezeWhenExpanded={false}
-    getLoadingProps={[Function]}
-    getNoDataProps={[Function]}
-    getPaginationProps={[Function]}
-    getProps={[Function]}
-    getResizerProps={[Function]}
-    getTableProps={[Function]}
-    getTbodyProps={[Function]}
-    getTdProps={[Function]}
-    getTfootProps={[Function]}
-    getTfootTdProps={[Function]}
-    getTfootTrProps={[Function]}
-    getTheadFilterProps={[Function]}
-    getTheadFilterThProps={[Function]}
-    getTheadFilterTrProps={[Function]}
-    getTheadGroupProps={[Function]}
-    getTheadGroupThProps={[Function]}
-    getTheadGroupTrProps={[Function]}
-    getTheadProps={[Function]}
-    getTheadThProps={[Function]}
-    getTheadTrProps={[Function]}
-    getTrGroupProps={[Function]}
-    getTrProps={[Function]}
-    groupedByPivotKey="_groupedByPivot"
-    indexKey="_index"
-    loading={true}
-    loadingText="Loading..."
-    multiSort={true}
-    nestingLevelKey="_nestingLevel"
-    nextText="Next"
-    noDataText=""
-    ofText="of"
-    onFetchData={[Function]}
-    onFilteredChange={[Function]}
-    originalKey="_original"
-    pageSizeOptions={
-      Array [
-        5,
-        10,
-        20,
-        25,
-        50,
-        100,
-      ]
-    }
-    pageText="Page"
-    pivotBy={Array []}
-    pivotDefaults={Object {}}
-    pivotIDKey="_pivotID"
-    pivotValKey="_pivotVal"
-    previousText="Previous"
-    resizable={true}
-    resolveData={[Function]}
-    rowsText="rows"
-    showPageJump={true}
-    showPageSizeOptions={true}
-    showPagination={true}
-    showPaginationBottom={true}
-    showPaginationTop={false}
-    sortable={true}
-    style={Object {}}
-    subRowsKey="_subRows"
-  />
-  <div
-    className="control-separator"
-  />
-  <ViewControlBar
-    label="MiddleManagers"
-  >
-    <Blueprint3.Button
-      icon="refresh"
-      onClick={[Function]}
-      text="Refresh"
-    />
-    <TableColumnSelection
-      columns={
-        Array [
-          "Host",
-          "Usage",
-          "Availability groups",
-          "Last completed task time",
-          "Blacklisted until",
+          "Segment ID",
+          "Datasource",
+          "Start",
+          "End",
+          "Version",
+          "Partition",
+          "Size",
+          "Num rows",
+          "Replicas",
+          "Is published",
+          "Is realtime",
+          "Is available",
+          "Is overshadowed",
         ]
       }
       onChange={[Function]}
@@ -325,37 +96,98 @@ exports[`describe segments-view segments view snapshot 1`] = `
     }
     columns={
       Array [
+        Object {
+          "Header": "Segment ID",
+          "accessor": "segment_id",
+          "show": true,
+          "width": 300,
+        },
         Object {
           "Cell": [Function],
-          "Header": "Host",
-          "accessor": [Function],
-          "id": "host",
+          "Header": "Datasource",
+          "accessor": "datasource",
           "show": true,
         },
         Object {
-          "Header": "Usage",
-          "accessor": [Function],
+          "Cell": [Function],
+          "Header": "Start",
+          "accessor": "start",
+          "defaultSortDesc": true,
+          "show": true,
+          "width": 120,
+        },
+        Object {
+          "Cell": [Function],
+          "Header": "End",
+          "accessor": "end",
+          "defaultSortDesc": true,
+          "show": true,
+          "width": 120,
+        },
+        Object {
+          "Header": "Version",
+          "accessor": "version",
+          "defaultSortDesc": true,
+          "show": true,
+          "width": 120,
+        },
+        Object {
+          "Header": "Partition",
+          "accessor": "partition_num",
           "filterable": false,
-          "id": "usage",
           "show": true,
           "width": 60,
         },
         Object {
-          "Header": "Availability groups",
-          "accessor": [Function],
+          "Cell": [Function],
+          "Header": "Size",
+          "accessor": "size",
+          "defaultSortDesc": true,
+          "filterable": false,
+          "show": true,
+        },
+        Object {
+          "Cell": [Function],
+          "Header": "Num rows",
+          "accessor": "num_rows",
+          "defaultSortDesc": true,
+          "filterable": false,
+          "show": true,
+        },
+        Object {
+          "Header": "Replicas",
+          "accessor": "num_replicas",
+          "defaultSortDesc": true,
           "filterable": false,
-          "id": "availabilityGroups",
           "show": true,
           "width": 60,
         },
         Object {
-          "Header": "Last completed task time",
-          "accessor": "lastCompletedTaskTime",
+          "Filter": [Function],
+          "Header": "Is published",
+          "accessor": [Function],
+          "id": "is_published",
           "show": true,
         },
         Object {
-          "Header": "Blacklisted until",
-          "accessor": "blacklistedUntil",
+          "Filter": [Function],
+          "Header": "Is realtime",
+          "accessor": [Function],
+          "id": "is_realtime",
+          "show": true,
+        },
+        Object {
+          "Filter": [Function],
+          "Header": "Is available",
+          "accessor": [Function],
+          "id": "is_available",
+          "show": true,
+        },
+        Object {
+          "Filter": [Function],
+          "Header": "Is overshadowed",
+          "accessor": [Function],
+          "id": "is_overshadowed",
           "show": true,
         },
       ]
@@ -364,11 +196,18 @@ exports[`describe segments-view segments view snapshot 1`] = `
     defaultExpanded={Object {}}
     defaultFilterMethod={[Function]}
     defaultFiltered={Array []}
-    defaultPageSize={10}
+    defaultPageSize={50}
     defaultResized={Array []}
     defaultSortDesc={false}
     defaultSortMethod={[Function]}
-    defaultSorted={Array []}
+    defaultSorted={
+      Array [
+        Object {
+          "desc": true,
+          "id": "start",
+        },
+      ]
+    }
     expanderDefaults={
       Object {
         "filterable": false,
@@ -381,7 +220,7 @@ exports[`describe segments-view segments view snapshot 1`] = `
     filtered={
       Array [
         Object {
-          "id": "host",
+          "id": "datasource",
           "value": "test",
         },
       ]
@@ -413,11 +252,12 @@ exports[`describe segments-view segments view snapshot 1`] = `
     indexKey="_index"
     loading={true}
     loadingText="Loading..."
+    manual={true}
     multiSort={true}
     nestingLevelKey="_nestingLevel"
     nextText="Next"
     noDataText=""
-    ofText="of"
+    ofText=""
     onFetchData={[Function]}
     onFilteredChange={[Function]}
     originalKey="_original"
@@ -432,6 +272,7 @@ exports[`describe segments-view segments view snapshot 1`] = `
       ]
     }
     pageText="Page"
+    pages={10000000}
     pivotDefaults={Object {}}
     pivotIDKey="_pivotID"
     pivotValKey="_pivotVal"
@@ -439,7 +280,7 @@ exports[`describe segments-view segments view snapshot 1`] = `
     resizable={true}
     resolveData={[Function]}
     rowsText="rows"
-    showPageJump={true}
+    showPageJump={false}
     showPageSizeOptions={true}
     showPagination={true}
     showPaginationBottom={true}
diff --git a/web-console/src/views/segments-view/segments-view.spec.tsx b/web-console/src/views/segments-view/segments-view.spec.tsx
index c1dbb7452bc..7b4531f88c5 100644
--- a/web-console/src/views/segments-view/segments-view.spec.tsx
+++ b/web-console/src/views/segments-view/segments-view.spec.tsx
@@ -22,16 +22,16 @@ import { shallow } from 'enzyme';
 import * as enzymeAdapterReact16 from 'enzyme-adapter-react-16';
 import * as React from 'react';
 
-import {ServersView} from '../servers-view/servers-view';
+import {SegmentsView} from '../segments-view/segments-view';
 
 Enzyme.configure({ adapter: new enzymeAdapterReact16() });
 describe('describe segments-view', () => {
   it('segments view snapshot', () => {
     const segmentsView = shallow(
-      <ServersView
-        middleManager={'test'}
+      <SegmentsView
+        datasource={'test'}
+        onlyUnavailable={false}
         goToSql={(initSql: string) => {}}
-        goToTask={(taskId: string) => {}}
         noSqlMode={false}
       />);
     expect(segmentsView).toMatchSnapshot();
diff --git a/web-console/src/views/servers-view/__snapshots__/servers-view.spec.tsx.snap b/web-console/src/views/servers-view/__snapshots__/servers-view.spec.tsx.snap
index a68a29e476d..72048f4f254 100644
--- a/web-console/src/views/servers-view/__snapshots__/servers-view.spec.tsx.snap
+++ b/web-console/src/views/servers-view/__snapshots__/servers-view.spec.tsx.snap
@@ -263,6 +263,8 @@ exports[`describe servers view action servers view 1`] = `
           "Availability groups",
           "Last completed task time",
           "Blacklisted until",
+          "Status",
+          "Actions",
         ]
       }
       onChange={[Function]}
@@ -358,6 +360,21 @@ exports[`describe servers view action servers view 1`] = `
           "accessor": "blacklistedUntil",
           "show": true,
         },
+        Object {
+          "Header": "Status",
+          "accessor": [Function],
+          "id": "status",
+          "show": true,
+        },
+        Object {
+          "Cell": [Function],
+          "Header": "Actions",
+          "accessor": [Function],
+          "filterable": false,
+          "id": "actions",
+          "show": true,
+          "width": 70,
+        },
       ]
     }
     data={Array []}
@@ -448,5 +465,29 @@ exports[`describe servers view action servers view 1`] = `
     style={Object {}}
     subRowsKey="_subRows"
   />
+  <AsyncActionDialog
+    action={null}
+    confirmButtonText="Disable worker"
+    failText="Could not disable worker"
+    intent="danger"
+    onClose={[Function]}
+    successText="Worker has been disabled"
+  >
+    <p>
+      Are you sure you want to disable worker 'null'?
+    </p>
+  </AsyncActionDialog>
+  <AsyncActionDialog
+    action={null}
+    confirmButtonText="Enable worker"
+    failText="Could not enable worker"
+    intent="primary"
+    onClose={[Function]}
+    successText="Worker has been enabled"
+  >
+    <p>
+      Are you sure you want to enable worker 'null'?
+    </p>
+  </AsyncActionDialog>
 </div>
 `;
diff --git a/web-console/src/views/servers-view/servers-view.tsx b/web-console/src/views/servers-view/servers-view.tsx
index 7d1021f5b15..e2c2bdb8a57 100644
--- a/web-console/src/views/servers-view/servers-view.tsx
+++ b/web-console/src/views/servers-view/servers-view.tsx
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-import { Button, Switch } from '@blueprintjs/core';
+import { Button, Icon, Intent, Popover, Position, Switch } from '@blueprintjs/core';
 import { IconNames } from '@blueprintjs/icons';
 import axios from 'axios';
 import { sum } from 'd3-array';
@@ -24,7 +24,12 @@ import * as React from 'react';
 import ReactTable from 'react-table';
 import { Filter } from 'react-table';
 
-import { TableColumnSelection, ViewControlBar } from '../../components/index';
+import {
+  ActionCell,
+  TableColumnSelection,
+  ViewControlBar
+} from '../../components/index';
+import { AsyncActionDialog } from '../../dialogs/index';
 import {
   addFilter,
   formatBytes,
@@ -32,11 +37,12 @@ import {
   queryDruidSql,
   QueryManager, TableColumnSelectionHandler
 } from '../../utils';
+import { BasicAction, basicActionsToMenu } from '../../utils/basic-action';
 
 import './servers-view.scss';
 
 const serverTableColumns: string[] = ['Server', 'Tier', 'Curr size', 'Max size', 'Usage', 'Load/drop queues', 'Host', 'Port'];
-const middleManagerTableColumns: string[] = ['Host', 'Usage', 'Availability groups', 'Last completed task time', 'Blacklisted until'];
+const middleManagerTableColumns: string[] = ['Host', 'Usage', 'Availability groups', 'Last completed task time', 'Blacklisted until', 'Status', 'Actions'];
 
 function formatQueues(segmentsToLoad: number, segmentsToLoadSize: number, segmentsToDrop: number, segmentsToDropSize: number): string {
   const queueParts: string[] = [];
@@ -67,6 +73,9 @@ export interface ServersViewState {
   middleManagers: any[] | null;
   middleManagersError: string | null;
   middleManagerFilter: Filter[];
+
+  middleManagerDisableWorkerHost: string | null;
+  middleManagerEnableWorkerHost: string | null;
 }
 
 interface ServerQueryResultRow {
@@ -110,7 +119,10 @@ export class ServersView extends React.Component<ServersViewProps, ServersViewSt
       middleManagersLoading: true,
       middleManagers: null,
       middleManagersError: null,
-      middleManagerFilter: props.middleManager ? [{ id: 'host', value: props.middleManager }] : []
+      middleManagerFilter: props.middleManager ? [{ id: 'host', value: props.middleManager }] : [],
+
+      middleManagerDisableWorkerHost: null,
+      middleManagerEnableWorkerHost: null
     };
 
     this.serverTableColumnSelectionHandler = new TableColumnSelectionHandler(
@@ -345,69 +357,171 @@ WHERE "server_type" = 'historical'`);
     const { middleManagers, middleManagersLoading, middleManagersError, middleManagerFilter } = this.state;
     const { middleManagerTableColumnSelectionHandler } = this;
 
-    return <ReactTable
-      data={middleManagers || []}
-      loading={middleManagersLoading}
-      noDataText={!middleManagersLoading && middleManagers && !middleManagers.length ? 'No MiddleManagers' : (middleManagersError || '')}
-      filterable
-      filtered={middleManagerFilter}
-      onFilteredChange={(filtered, column) => {
-        this.setState({ middleManagerFilter: filtered });
-      }}
-      columns={[
-        {
-          Header: 'Host',
-          id: 'host',
-          accessor: (row) => row.worker.host,
-          Cell: row => {
-            const value = row.value;
-            return <a onClick={() => { this.setState({ middleManagerFilter: addFilter(middleManagerFilter, 'host', value) }); }}>{value}</a>;
-          },
-          show: middleManagerTableColumnSelectionHandler.showColumn('Host')
-        },
-        {
-          Header: 'Usage',
-          id: 'usage',
-          width: 60,
-          accessor: (row) => `${row.currCapacityUsed} / ${row.worker.capacity}`,
-          filterable: false,
-          show: middleManagerTableColumnSelectionHandler.showColumn('Usage')
-        },
-        {
-          Header: 'Availability groups',
-          id: 'availabilityGroups',
-          width: 60,
-          accessor: (row) => row.availabilityGroups.length,
-          filterable: false,
-          show: middleManagerTableColumnSelectionHandler.showColumn('Availability groups')
-        },
-        {
-          Header: 'Last completed task time',
-          accessor: 'lastCompletedTaskTime',
-          show: middleManagerTableColumnSelectionHandler.showColumn('Last completed task time')
-        },
-        {
-          Header: 'Blacklisted until',
-          accessor: 'blacklistedUntil',
-          show: middleManagerTableColumnSelectionHandler.showColumn('Blacklisted until')
-        }
-      ]}
-      defaultPageSize={10}
-      className="-striped -highlight"
-      SubComponent={rowInfo => {
-        const runningTasks = rowInfo.original.runningTasks;
-        return <div style={{ padding: '20px' }}>
+    return <>
+      <ReactTable
+        data={middleManagers || []}
+        loading={middleManagersLoading}
+        noDataText={!middleManagersLoading && middleManagers && !middleManagers.length ? 'No MiddleManagers' : (middleManagersError || '')}
+        filterable
+        filtered={middleManagerFilter}
+        onFilteredChange={(filtered, column) => {
+          this.setState({ middleManagerFilter: filtered });
+        }}
+        columns={[
           {
-            runningTasks.length ?
-              <>
-                <span>Running tasks:</span>
-                <ul>{runningTasks.map((t: string) => <li key={t}>{t}&nbsp;<a onClick={() => goToTask(t)}>&#x279A;</a></li>)}</ul>
-              </> :
-              <span>No running tasks</span>
+            Header: 'Host',
+            id: 'host',
+            accessor: (row) => row.worker.host,
+            Cell: row => {
+              const value = row.value;
+              return <a onClick={() => { this.setState({ middleManagerFilter: addFilter(middleManagerFilter, 'host', value) }); }}>{value}</a>;
+            },
+            show: middleManagerTableColumnSelectionHandler.showColumn('Host')
+          },
+          {
+            Header: 'Usage',
+            id: 'usage',
+            width: 60,
+            accessor: (row) => `${row.currCapacityUsed} / ${row.worker.capacity}`,
+            filterable: false,
+            show: middleManagerTableColumnSelectionHandler.showColumn('Usage')
+          },
+          {
+            Header: 'Availability groups',
+            id: 'availabilityGroups',
+            width: 60,
+            accessor: (row) => row.availabilityGroups.length,
+            filterable: false,
+            show: middleManagerTableColumnSelectionHandler.showColumn('Availability groups')
+          },
+          {
+            Header: 'Last completed task time',
+            accessor: 'lastCompletedTaskTime',
+            show: middleManagerTableColumnSelectionHandler.showColumn('Last completed task time')
+          },
+          {
+            Header: 'Blacklisted until',
+            accessor: 'blacklistedUntil',
+            show: middleManagerTableColumnSelectionHandler.showColumn('Blacklisted until')
+          },
+          {
+            Header: 'Status',
+            id: 'status',
+            accessor: (row) => row.worker.version === '' ? 'Disabled' : 'Enabled',
+            show: middleManagerTableColumnSelectionHandler.showColumn('Status')
+          },
+          {
+            Header: 'Actions',
+            id: 'actions',
+            width: 70,
+            accessor: (row) => row.worker,
+            filterable: false,
+            Cell: row => {
+              const disabled = row.value.version === '';
+              const workerActions = this.getWorkerActions(row.value.host, disabled);
+              const workerMenu = basicActionsToMenu(workerActions);
+
+              return <ActionCell>
+                {
+                   workerMenu &&
+                   <Popover content={workerMenu} position={Position.BOTTOM_RIGHT}>
+                     <Icon icon={IconNames.WRENCH}/>
+                   </Popover>
+                }
+              </ActionCell>;
+            },
+            show: middleManagerTableColumnSelectionHandler.showColumn('Actions')
           }
-        </div>;
+        ]}
+        defaultPageSize={10}
+        className="-striped -highlight"
+        SubComponent={rowInfo => {
+          const runningTasks = rowInfo.original.runningTasks;
+          return <div style={{ padding: '20px' }}>
+            {
+              runningTasks.length ?
+                <>
+                  <span>Running tasks:</span>
+                  <ul>{runningTasks.map((t: string) => <li key={t}>{t}&nbsp;<a onClick={() => goToTask(t)}>&#x279A;</a></li>)}</ul>
+                </> :
+                <span>No running tasks</span>
+            }
+          </div>;
+        }}
+      />
+      {this.renderDisableWorkerAction()}
+      {this.renderEnableWorkerAction()}
+    </>;
+  }
+
+  private getWorkerActions(workerHost: string, disabled: boolean): BasicAction[] {
+    if (disabled) {
+      return [
+        {
+          icon: IconNames.TICK,
+          title: 'Enable',
+          onAction: () => this.setState({ middleManagerEnableWorkerHost: workerHost })
+        }
+      ];
+    } else {
+      return [
+        {
+          icon: IconNames.DISABLE,
+          title: 'Disable',
+          onAction: () => this.setState({ middleManagerDisableWorkerHost: workerHost })
+        }
+      ];
+    }
+  }
+
+  renderDisableWorkerAction() {
+    const { middleManagerDisableWorkerHost } = this.state;
+
+    return <AsyncActionDialog
+      action={
+        middleManagerDisableWorkerHost ? async () => {
+          const resp = await axios.post(`/druid/indexer/v1/worker/${middleManagerDisableWorkerHost}/disable`, {});
+          return resp.data;
+        } : null
+      }
+      confirmButtonText="Disable worker"
+      successText="Worker has been disabled"
+      failText="Could not disable worker"
+      intent={Intent.DANGER}
+      onClose={(success) => {
+        this.setState({ middleManagerDisableWorkerHost: null });
+        if (success) this.middleManagerQueryManager.rerunLastQuery();
       }}
-    />;
+    >
+      <p>
+        {`Are you sure you want to disable worker '${middleManagerDisableWorkerHost}'?`}
+      </p>
+    </AsyncActionDialog>;
+  }
+
+  renderEnableWorkerAction() {
+    const { middleManagerEnableWorkerHost } = this.state;
+
+    return <AsyncActionDialog
+      action={
+        middleManagerEnableWorkerHost ? async () => {
+          const resp = await axios.post(`/druid/indexer/v1/worker/${middleManagerEnableWorkerHost}/enable`, {});
+          return resp.data;
+        } : null
+      }
+      confirmButtonText="Enable worker"
+      successText="Worker has been enabled"
+      failText="Could not enable worker"
+      intent={Intent.PRIMARY}
+      onClose={(success) => {
+        this.setState({ middleManagerEnableWorkerHost: null });
+        if (success) this.middleManagerQueryManager.rerunLastQuery();
+      }}
+    >
+      <p>
+        {`Are you sure you want to enable worker '${middleManagerEnableWorkerHost}'?`}
+      </p>
+    </AsyncActionDialog>;
   }
 
   render() {