diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignor.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignor.java
index d48589423dc78..1ff9819e538d4 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignor.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignor.java
@@ -240,6 +240,13 @@ ClusterAssignment performTaskAssignment(
             previousRevocation.tasks().clear();
         }
 
+        final long now = time.milliseconds();
+        // Before any operations, check if the rebalance delay has expired. In that case,
+        // we will reset the revoking rebalance state.
+        if (scheduledRebalance > 0 && now >= scheduledRebalance) {
+            resetPreviousRevocationsState();
+        }
+
         // Derived set: The set of deleted connectors-and-tasks is a derived set from the set
         // difference of previous - configured
         ConnectorsAndTasks deleted = diff(previousAssignment, configured);
@@ -324,15 +331,14 @@ ClusterAssignment performTaskAssignment(
                 // have converged to a balanced load. We can reset the rebalance clock
                 log.debug("Previous round had revocations but this round didn't. Probably, the cluster has reached a " +
                         "balanced load. Resetting the exponential backoff clock");
-                revokedInPrevious = false;
-                numSuccessiveRevokingRebalances = 0;
+                resetPreviousRevocationsState();
             } else {
                 // no-op
                 log.debug("No revocations in previous and current round.");
             }
         } else {
             log.debug("Delayed rebalance is active. Delaying {}ms before revoking connectors and tasks: {}", delay, toRevoke);
-            revokedInPrevious = false;
+            resetPreviousRevocationsState();
         }
 
         // The complete set of connectors and tasks that should be newly-assigned during this round
@@ -442,8 +448,8 @@ private ConnectorsAndTasks duplicatedAssignments(Map<String, ConnectorsAndTasks>
     protected void handleLostAssignments(ConnectorsAndTasks lostAssignments,
                                          ConnectorsAndTasks.Builder lostAssignmentsToReassign,
                                          List<WorkerLoad> completeWorkerAssignment) {
-        // There are no lost assignments and there have been no successive revoking rebalances
-        if (lostAssignments.isEmpty() && !revokedInPrevious) {
+
+        if (lostAssignments.isEmpty()) {
             resetDelay();
             return;
         }
@@ -502,9 +508,6 @@ protected void handleLostAssignments(ConnectorsAndTasks lostAssignments,
                 lostAssignmentsToReassign.addTasks(lostAssignments.tasks());
             }
             resetDelay();
-            // Resetting the flag as now we can permit successive revoking rebalances.
-            // since we have gone through the full rebalance delay
-            revokedInPrevious = false;
         } else {
             candidateWorkersForReassignment
                     .addAll(candidateWorkersForReassignment(completeWorkerAssignment));
@@ -535,6 +538,11 @@ private void resetDelay() {
         delay = 0;
     }
 
+    private void resetPreviousRevocationsState() {
+        revokedInPrevious = false;
+        numSuccessiveRevokingRebalances = 0;
+    }
+
     private Set<String> candidateWorkersForReassignment(List<WorkerLoad> completeWorkerAssignment) {
         return completeWorkerAssignment.stream()
                 .filter(WorkerLoad::isEmpty)
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignorTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignorTest.java
index 9ec5368a8bbdb..9e4d9eae593e6 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignorTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/IncrementalCooperativeAssignorTest.java
@@ -142,6 +142,7 @@ public void testAssignmentsWhenWorkersJoinAfterRevocations()  {
         // We should revoke.
         addNewEmptyWorkers("worker2");
         performStandardRebalance();
+        assertEquals(0, assignor.delay); // First revoking rebalance
         assertWorkers("worker1", "worker2");
         assertConnectorAllocations(0, 2);
         assertTaskAllocations(0, 6);
@@ -151,7 +152,7 @@ public void testAssignmentsWhenWorkersJoinAfterRevocations()  {
         // in this round
         addNewEmptyWorkers("worker3");
         performStandardRebalance();
-        assertTrue(assignor.delay > 0);
+        assertEquals(40, assignor.delay); // First successive revoking rebalance.
         assertWorkers("worker1", "worker2", "worker3");
         assertConnectorAllocations(0, 1, 2);
         assertTaskAllocations(3, 3, 6);
@@ -161,30 +162,38 @@ public void testAssignmentsWhenWorkersJoinAfterRevocations()  {
         time.sleep(assignor.delay);
         addNewEmptyWorkers("worker4");
         performStandardRebalance();
+        assertEquals(0, assignor.delay); // First revoking rebalance after delay
         assertWorkers("worker1", "worker2", "worker3", "worker4");
         assertConnectorAllocations(0, 0, 1, 1);
         assertTaskAllocations(0, 3, 3, 3);
 
         // Fifth assignment and a fifth worker joining after a revoking rebalance.
-        // We shouldn't revoke and set a delay > initial interval
+        // We shouldn't revoke and set a delay equal to initial interval
         addNewEmptyWorkers("worker5");
         performStandardRebalance();
-        assertTrue(assignor.delay > 40);
+        assertEquals(40, assignor.delay); // First successive revoking rebalance after delay
         assertWorkers("worker1", "worker2", "worker3", "worker4", "worker5");
         assertConnectorAllocations(0, 0, 1, 1, 1);
         assertTaskAllocations(1, 2, 3, 3, 3);
 
-        // Sixth assignment with sixth worker joining after the expiry.
-        // Should revoke
-        time.sleep(assignor.delay);
+        // Sixth assignment with sixth worker joining within rebalance interval
+        // There should not be any revocations and delay should be extended further.
         addNewEmptyWorkers("worker6");
         performStandardRebalance();
-        assertDelay(0);
+        assertTrue(assignor.delay > 40); // Second successive revoking rebalance after delay
+        assertWorkers("worker1", "worker2", "worker3", "worker4", "worker5", "worker6");
+        assertConnectorAllocations(0, 0, 0, 1, 1, 1);
+        assertTaskAllocations(0, 1, 2, 3, 3, 3);
+
+        // Follow up rebalance after delay expires.
+        time.sleep(assignor.delay);
+        performStandardRebalance();
+        assertEquals(0, assignor.delay); // Delay reset
         assertWorkers("worker1", "worker2", "worker3", "worker4", "worker5", "worker6");
         assertConnectorAllocations(0, 0, 0, 1, 1, 1);
         assertTaskAllocations(0, 1, 2, 2, 2, 2);
 
-        // Follow up rebalance since there were revocations
+        // Final rebalance because of revocations in the previous round
         performStandardRebalance();
         assertWorkers("worker1", "worker2", "worker3", "worker4", "worker5", "worker6");
         assertConnectorAllocations(0, 0, 0, 1, 1, 1);