diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java b/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java
index 953e52a..89d6849 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/Cluster.java
@@ -627,14 +627,12 @@
         double cpuAvailable = resourcesAvailable.getTotalCpu();
 
         if (cpuAdded > cpuAvailable) {
-            if (LOG.isTraceEnabled()) {
-                LOG.trace("Could not schedule {}:{} on {} not enough CPU {} > {}",
-                        td.getName(),
-                        exec,
-                        ws,
-                        cpuAdded,
-                        cpuAvailable);
-            }
+            LOG.debug("Could not schedule {}:{} on {} not enough CPU {} > {}",
+                    td.getName(),
+                    exec,
+                    ws,
+                    cpuAdded,
+                    cpuAvailable);
             return false;
         }
 
@@ -642,25 +640,21 @@
         double memoryAvailable = resourcesAvailable.getTotalMemoryMb();
 
         if (memoryAdded > memoryAvailable) {
-            if (LOG.isTraceEnabled()) {
-                LOG.trace("Could not schedule {}:{} on {} not enough Mem {} > {}",
-                          td.getName(),
-                          exec,
-                          ws,
-                          memoryAdded,
-                          memoryAvailable);
-            }
+            LOG.debug("Could not schedule {}:{} on {} not enough Mem {} > {}",
+                      td.getName(),
+                      exec,
+                      ws,
+                      memoryAdded,
+                      memoryAvailable);
             return false;
         }
         if (afterOnHeap > maxHeap) {
-            if (LOG.isTraceEnabled()) {
-                LOG.trace("Could not schedule {}:{} on {} HEAP would be too large {} > {}",
-                          td.getName(),
-                          exec,
-                          ws,
-                          afterOnHeap,
-                          maxHeap);
-            }
+            LOG.debug("Could not schedule {}:{} on {} HEAP would be too large {} > {}",
+                      td.getName(),
+                      exec,
+                      ws,
+                      afterOnHeap,
+                      maxHeap);
             return false;
         }
         return true;
diff --git a/storm-server/src/main/java/org/apache/storm/utils/ServerUtils.java b/storm-server/src/main/java/org/apache/storm/utils/ServerUtils.java
index 6928eb1..a70ae86 100644
--- a/storm-server/src/main/java/org/apache/storm/utils/ServerUtils.java
+++ b/storm-server/src/main/java/org/apache/storm/utils/ServerUtils.java
@@ -708,9 +708,27 @@
             double memoryRequirement = entry.getValue().getOnHeapMemoryMb();
             totalMemoryRequired += memoryRequirement * parallelism;
         }
+
+        double ackerMem = getTotalAckerExecutorMemoryUsageForTopo(topology, topoConf);
+        totalMemoryRequired += ackerMem;
+
         return totalMemoryRequired;
     }
 
+    private static double getTotalAckerExecutorMemoryUsageForTopo(
+            StormTopology topology, Map<String, Object> topologyConf)
+            throws InvalidTopologyException {
+        topology = StormCommon.systemTopology(topologyConf, topology);
+        Map<String, NormalizedResourceRequest> boltResources = ResourceUtils.getBoltsResources(topology, topologyConf);
+        NormalizedResourceRequest entry = boltResources.get(Acker.ACKER_COMPONENT_ID);
+        if (entry == null) {
+            return 0.0d;
+        }
+        Map<String, Integer> componentParallelism = getComponentParallelism(topologyConf, topology);
+        int parallelism = componentParallelism.getOrDefault(Acker.ACKER_COMPONENT_ID, 1);
+        return entry.getTotalMemoryMb() * parallelism;
+    }
+
     public static Map<String, Integer> getComponentParallelism(Map<String, Object> topoConf, StormTopology topology)
         throws InvalidTopologyException {
         Map<String, Integer> ret = new HashMap<>();
@@ -1398,6 +1416,9 @@
         topology = StormCommon.systemTopology(topologyConf, topology);
         Map<String, NormalizedResourceRequest> boltResources = ResourceUtils.getBoltsResources(topology, topologyConf);
         NormalizedResourceRequest entry = boltResources.get(Acker.ACKER_COMPONENT_ID);
+        if (entry == null) {
+            return 0.0d;
+        }
         return entry.getTotalMemoryMb();
     }
 
diff --git a/storm-server/src/test/java/org/apache/storm/scheduler/blacklist/TestBlacklistScheduler.java b/storm-server/src/test/java/org/apache/storm/scheduler/blacklist/TestBlacklistScheduler.java
index bed28ce..abc5f25 100644
--- a/storm-server/src/test/java/org/apache/storm/scheduler/blacklist/TestBlacklistScheduler.java
+++ b/storm-server/src/test/java/org/apache/storm/scheduler/blacklist/TestBlacklistScheduler.java
@@ -20,10 +20,19 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Sets;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.config.Configurator;
 import org.apache.storm.Config;
 import org.apache.storm.DaemonConfig;
+import org.apache.storm.daemon.StormCommon;
+import org.apache.storm.generated.Bolt;
+import org.apache.storm.generated.InvalidTopologyException;
+import org.apache.storm.generated.SpoutSpec;
+import org.apache.storm.generated.StormTopology;
 import org.apache.storm.scheduler.Cluster;
 import org.apache.storm.scheduler.DefaultScheduler;
+import org.apache.storm.scheduler.ExecutorDetails;
 import org.apache.storm.scheduler.INimbus;
 import org.apache.storm.scheduler.IScheduler;
 import org.apache.storm.scheduler.SchedulerAssignmentImpl;
@@ -31,11 +40,13 @@
 import org.apache.storm.scheduler.Topologies;
 import org.apache.storm.scheduler.TopologyDetails;
 import org.apache.storm.scheduler.resource.ResourceAwareScheduler;
+import org.apache.storm.scheduler.resource.TestUtilsForResourceAwareScheduler;
 import org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategy;
 import org.apache.storm.scheduler.resource.strategies.scheduling.DefaultResourceAwareStrategyOld;
 import org.apache.storm.scheduler.resource.strategies.scheduling.GenericResourceAwareStrategy;
 import org.apache.storm.scheduler.resource.strategies.scheduling.GenericResourceAwareStrategyOld;
 import org.apache.storm.scheduler.resource.strategies.scheduling.RoundRobinResourceAwareStrategy;
+import org.apache.storm.topology.TopologyBuilder;
 import org.apache.storm.utils.Utils;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
@@ -74,6 +85,118 @@
     }
 
     @Test
+    public void testBlacklistResumeWhenAckersWontFit() throws InvalidTopologyException {
+        // 3 supervisors exist with 4 slots, 2 are blacklisted
+        // topology with given worker heap size would fit in 4 slots if ignoring ackers, needs 5 slots with ackers.
+        // verify that one of the supervisors will be resumed and topology will schedule.
+
+        Config config = new Config();
+        config.putAll(Utils.readDefaultConfig());
+        config.put(DaemonConfig.BLACKLIST_SCHEDULER_TOLERANCE_TIME, 300);
+        config.put(DaemonConfig.BLACKLIST_SCHEDULER_TOLERANCE_COUNT, 3);
+        config.put(DaemonConfig.BLACKLIST_SCHEDULER_RESUME_TIME, 1800);
+        config.put(DaemonConfig.STORM_WORKER_MIN_CPU_PCORE_PERCENT, 100);
+        config.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, 10);
+        config.put(DaemonConfig.BLACKLIST_SCHEDULER_ASSUME_SUPERVISOR_BAD_BASED_ON_BAD_SLOT, false);
+        config.put(DaemonConfig.BLACKLIST_SCHEDULER_TOLERANCE_COUNT, 3);
+        config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 128);
+        config.put(DaemonConfig.BLACKLIST_SCHEDULER_STRATEGY, "org.apache.storm.scheduler.blacklist.strategies.RasBlacklistStrategy");
+        config.put(Config.TOPOLOGY_RAS_ACKER_EXECUTORS_PER_WORKER, 1);
+        config.setNumWorkers(1);
+        config.put(Config.TOPOLOGY_ACKER_EXECUTORS, 4);
+        config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, "org.apache.storm.scheduler.resource.strategies.scheduling.GenericResourceAwareStrategy");
+        config.put(Config.TOPOLOGY_WORKER_MAX_HEAP_SIZE_MB, 512);
+        config.put(Config.WORKER_HEAP_MEMORY_MB, 768);
+        config.put(Config.TOPOLOGY_NAME, "testTopology");
+
+        INimbus iNimbus = new TestUtilsForBlacklistScheduler.INimbusTest();
+        StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
+        ResourceMetrics resourceMetrics = new ResourceMetrics(metricsRegistry);
+        Map<String, SupervisorDetails> supMap = TestUtilsForBlacklistScheduler.genSupervisors(3, 4, 400.0d, 4096.0d);
+        Topologies noTopologies = new Topologies();
+        Cluster cluster = new Cluster(iNimbus, resourceMetrics, supMap, new HashMap<>(), noTopologies, config);
+
+        scheduler = new BlacklistScheduler(new ResourceAwareScheduler());
+        scheduler.prepare(config, metricsRegistry);
+        scheduler.schedule(noTopologies, cluster);
+
+        Map<String, SupervisorDetails> removedSup0 = TestUtilsForBlacklistScheduler.removeSupervisorFromSupervisors(supMap, "sup-0");
+        Map<String, SupervisorDetails> removedSup0Sup1 = TestUtilsForBlacklistScheduler.removeSupervisorFromSupervisors(removedSup0, "sup-1");
+
+        cluster = new Cluster(iNimbus, resourceMetrics, removedSup0Sup1,
+                TestUtilsForBlacklistScheduler.assignmentMapToImpl(cluster.getAssignments()), noTopologies, config);
+        scheduler.schedule(noTopologies, cluster);
+        scheduler.schedule(noTopologies, cluster);
+        scheduler.schedule(noTopologies, cluster);
+
+        // 2 supervisors blacklisted at this point.  Let's schedule the topology.
+
+        Map<String, TopologyDetails> topoMap = new HashMap<>();
+        TopologyDetails topo1 = createResourceTopo(config);
+        topoMap.put(topo1.getId(), topo1);
+        Topologies topologies = new Topologies(topoMap);
+
+        cluster = new Cluster(iNimbus, resourceMetrics, supMap,
+                TestUtilsForBlacklistScheduler.assignmentMapToImpl(cluster.getAssignments()), topologies, config);
+        boolean enableTraceLogging = false; // for scheduling debug
+        if (enableTraceLogging) {
+            Configurator.setAllLevels(LogManager.getRootLogger().getName(), Level.TRACE);
+        }
+        scheduler.schedule(topologies, cluster);
+
+        // topology should be fully scheduled with 1 host remaining blacklisted
+        String topoScheduleStatus = cluster.getStatus("testTopology-id");
+        assertTrue(topoScheduleStatus.contains("Running - Fully Scheduled"));
+        assertEquals(1, cluster.getBlacklistedHosts().size());
+    }
+
+    public TopologyDetails createResourceTopo(Config conf) throws InvalidTopologyException {
+        int spoutParallelism = 6;
+        int boltParallelism = 5;
+        TopologyBuilder builder = new TopologyBuilder();
+        builder.setSpout("word", new TestUtilsForResourceAwareScheduler.TestSpout(),
+                spoutParallelism);
+        builder.setBolt("exclaim1", new TestUtilsForResourceAwareScheduler.TestBolt(),
+                boltParallelism).shuffleGrouping("word");
+        builder.setBolt("exclaim2", new TestUtilsForResourceAwareScheduler.TestBolt(),
+                boltParallelism).shuffleGrouping("exclaim1");
+
+        StormTopology stormTopology = builder.createTopology();
+
+        TopologyDetails topo = new TopologyDetails("testTopology-id", conf, stormTopology, 0,
+                genExecsAndComps(StormCommon.systemTopology(conf, stormTopology)), currentTime, "user");
+        return topo;
+    }
+
+    public static Map<ExecutorDetails, String> genExecsAndComps(StormTopology topology) {
+        Map<ExecutorDetails, String> retMap = new HashMap<>();
+        int startTask = 1;
+        int endTask = 1;
+        for (Map.Entry<String, SpoutSpec> entry : topology.get_spouts().entrySet()) {
+            SpoutSpec spout = entry.getValue();
+            String spoutId = entry.getKey();
+            int spoutParallelism = spout.get_common().get_parallelism_hint();
+            for (int i = 0; i < spoutParallelism; i++) {
+                retMap.put(new ExecutorDetails(startTask, endTask), spoutId);
+                startTask++;
+                endTask++;
+            }
+        }
+
+        for (Map.Entry<String, Bolt> entry : topology.get_bolts().entrySet()) {
+            String boltId = entry.getKey();
+            Bolt bolt = entry.getValue();
+            int boltParallelism = bolt.get_common().get_parallelism_hint();
+            for (int i = 0; i < boltParallelism; i++) {
+                retMap.put(new ExecutorDetails(startTask, endTask), boltId);
+                startTask++;
+                endTask++;
+            }
+        }
+        return retMap;
+    }
+
+    @Test
     public void TestBadSupervisor() {
         INimbus iNimbus = new TestUtilsForBlacklistScheduler.INimbusTest();
 
diff --git a/storm-server/src/test/java/org/apache/storm/scheduler/blacklist/TestUtilsForBlacklistScheduler.java b/storm-server/src/test/java/org/apache/storm/scheduler/blacklist/TestUtilsForBlacklistScheduler.java
index 7d2f3fe..403cb99 100644
--- a/storm-server/src/test/java/org/apache/storm/scheduler/blacklist/TestUtilsForBlacklistScheduler.java
+++ b/storm-server/src/test/java/org/apache/storm/scheduler/blacklist/TestUtilsForBlacklistScheduler.java
@@ -18,6 +18,7 @@
 package org.apache.storm.scheduler.blacklist;
 
 import org.apache.storm.Config;
+import org.apache.storm.Constants;
 import org.apache.storm.generated.Bolt;
 import org.apache.storm.generated.SpoutSpec;
 import org.apache.storm.generated.StormTopology;
@@ -112,6 +113,22 @@
         return retList;
     }
 
+    public static Map<String, SupervisorDetails> genSupervisors(int numSup, int numPorts, double cpu, double memory) {
+        Map<String, Double> totalResources = new HashMap<>();
+        totalResources.put(Constants.COMMON_CPU_RESOURCE_NAME, cpu);
+        totalResources.put(Constants.COMMON_TOTAL_MEMORY_RESOURCE_NAME, memory);
+        Map<String, SupervisorDetails> retList = new HashMap<>();
+        for (int i = 0; i < numSup; i++) {
+            List<Number> ports = new LinkedList<>();
+            for (int j = 0; j < numPorts; j++) {
+                ports.add(j);
+            }
+            SupervisorDetails sup = new SupervisorDetails("sup-" + i, "host-" + i, null, ports, totalResources);
+            retList.put(sup.getId(), sup);
+        }
+        return retList;
+    }
+
 
     public static TopologyDetails getTopology(String name, Map<String, Object> config, int numSpout, int numBolt,
                                               int spoutParallelism, int boltParallelism, int launchTime, boolean blacklistEnable) {