diff --git a/CHANGES.txt b/CHANGES.txt
index 43578775ed..ad42e1c1dd 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -3,6 +3,7 @@
  * Adding missing configs in system_views.settings to be backward compatible (CASSANDRA-20863)
  * Heap dump should not be generated on handled exceptions (CASSANDRA-20974)
 Merged from 4.1:
+ * Disk usage guardrail cannot be disabled when failure threshold is reached (CASSANDRA-21057)
  * ReadCommandController should close fast to avoid deadlock when building secondary index (CASSANDRA-19564)
 Merged from 4.0:
  * Backport fix to nodetool gcstats output for direct memory (CASSANDRA-21037)
diff --git a/src/java/org/apache/cassandra/service/disk/usage/DiskUsageMonitor.java b/src/java/org/apache/cassandra/service/disk/usage/DiskUsageMonitor.java
index fc0bb3a554..fad5945483 100644
--- a/src/java/org/apache/cassandra/service/disk/usage/DiskUsageMonitor.java
+++ b/src/java/org/apache/cassandra/service/disk/usage/DiskUsageMonitor.java
@@ -59,6 +59,7 @@ public class DiskUsageMonitor
     private final Supplier<Multimap<FileStore, Directories.DataDirectory>> dataDirectoriesSupplier;
 
     private volatile DiskUsageState localState = DiskUsageState.NOT_AVAILABLE;
+    private volatile boolean enabled;
 
     @VisibleForTesting
     public DiskUsageMonitor()
@@ -79,14 +80,27 @@ public class DiskUsageMonitor
     {
         // start the scheduler regardless guardrail is enabled, so we can enable it later without a restart
         ScheduledExecutors.scheduledTasks.scheduleAtFixedRate(() -> {
-
-            if (!Guardrails.localDataDiskUsage.enabled(null))
-                return;
-
-            updateLocalState(getDiskUsage(), notifier);
+            boolean currentEnabled = Guardrails.localDataDiskUsage.enabled(null);
+            boolean oldEnabled = enabled;
+            enabled = currentEnabled;
+            boolean isDisabled = !currentEnabled && oldEnabled;
+            if (isDisabled)
+            {
+                onDiskUsageGuardrailDisabled(notifier);
+            }
+            if (currentEnabled)
+            {
+                updateLocalState(getDiskUsage(), notifier);
+            }
         }, 0, CassandraRelevantProperties.DISK_USAGE_MONITOR_INTERVAL_MS.getLong(), TimeUnit.MILLISECONDS);
     }
 
+    private void onDiskUsageGuardrailDisabled(Consumer<DiskUsageState> notifier)
+    {
+        localState = DiskUsageState.NOT_AVAILABLE;
+        notifier.accept(DiskUsageState.NOT_AVAILABLE);
+    }
+
     @VisibleForTesting
     public void updateLocalState(double usageRatio, Consumer<DiskUsageState> notifier)
     {
diff --git a/test/distributed/org/apache/cassandra/distributed/test/guardrails/GuardrailDiskUsageTest.java b/test/distributed/org/apache/cassandra/distributed/test/guardrails/GuardrailDiskUsageTest.java
index dd205bf87a..3bbd0e4a4a 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/guardrails/GuardrailDiskUsageTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/guardrails/GuardrailDiskUsageTest.java
@@ -194,6 +194,83 @@ public class GuardrailDiskUsageTest extends GuardrailTester
         }
     }
 
+    @Test
+    public void testDiskUsageNodetoolDisableWhenDiskIsFullShouldEnableWrites()
+    {
+        schemaChange("CREATE TABLE %s (k int PRIMARY KEY, v int)");
+        String insert = format("INSERT INTO %s(k, v) VALUES (?, 0)");
+
+        // With both nodes in SPACIOUS state, we can write without warnings nor failures
+        for (int i = 0; i < NUM_ROWS; i++)
+        {
+            ResultSet rs = driverSession.execute(insert, i);
+            Assertions.assertThat(rs.getExecutionInfo().getWarnings()).isEmpty();
+        }
+
+        // If the STUFFED node becomes FULL, the writes targeting that node will fail, while the writes targeting
+        // the node that remains SPACIOUS will keep succeeding without warnings
+        DiskStateInjection.setState(getCluster(), 2, DiskUsageState.FULL);
+        int numFailures = 0;
+        for (int i = 0; i < NUM_ROWS; i++)
+        {
+            try
+            {
+                ResultSet rs = driverSession.execute(insert, i);
+                Assertions.assertThat(rs.getExecutionInfo().getWarnings()).isEmpty();
+            }
+            catch (InvalidQueryException e)
+            {
+                Assertions.assertThat(e).hasMessageContaining(FAIL_MESSAGE);
+                numFailures++;
+            }
+        }
+        Assertions.assertThat(numFailures).isGreaterThan(0).isLessThan(NUM_ROWS);
+
+        // After disabling the guardrail, we should be able to write again.
+        cluster.get(2).runOnInstance(() -> Guardrails.instance.setDataDiskUsagePercentageThreshold(-1, -1));
+        int stateDissemenationTimeoutSec = 2 * 60; // 2 minutes.
+        Util.spinAssertEquals(true,
+            () -> cluster.get(1).callOnInstance(() -> !DiskUsageBroadcaster.instance.hasStuffedOrFullNode()),
+            stateDissemenationTimeoutSec
+        );
+
+        for (int i = 0; i < NUM_ROWS; i++)
+        {
+            ResultSet rs = driverSession.execute(insert, i);
+            Assertions.assertThat(rs.getExecutionInfo().getWarnings()).isEmpty();
+        }
+
+        // Re-enabling the guardrail should again cause writes to fail
+        cluster.get(2).runOnInstance(() -> Guardrails.instance.setDataDiskUsagePercentageThreshold(98, 99));
+        Util.spinAssertEquals(true,
+            () -> cluster.get(1).callOnInstance(() -> DiskUsageBroadcaster.instance.hasStuffedOrFullNode()),
+            stateDissemenationTimeoutSec
+        );
+        numFailures = 0;
+        for (int i = 0; i < NUM_ROWS; i++)
+        {
+            try
+            {
+                ResultSet rs = driverSession.execute(insert, i);
+                Assertions.assertThat(rs.getExecutionInfo().getWarnings()).isEmpty();
+            }
+            catch (InvalidQueryException e)
+            {
+                Assertions.assertThat(e).hasMessageContaining(FAIL_MESSAGE);
+                numFailures++;
+            }
+        }
+        Assertions.assertThat(numFailures).isGreaterThan(0).isLessThan(NUM_ROWS);
+
+        // Finally, if both nodes go back to SPACIOUS, all queries will succeed again
+        DiskStateInjection.setState(getCluster(), 2, DiskUsageState.SPACIOUS);
+        for (int i = 0; i < NUM_ROWS; i++)
+        {
+            ResultSet rs = driverSession.execute(insert, i);
+            Assertions.assertThat(rs.getExecutionInfo().getWarnings()).isEmpty();
+        }
+    }
+
     /**
      * ByteBuddy rule to override the disk usage state of each node.
      */