Allow tests to permit cluster state application failures

DaveCTurner · DaveCTurner · commit 51b9901cc7e9 · 2019-09-25T10:40:22.000+01:00
diff --git a/server/src/main/java/org/elasticsearch/cluster/service/ClusterApplierService.java b/server/src/main/java/org/elasticsearch/cluster/service/ClusterApplierService.java
@@ -449,7 +449,7 @@ private void runTask(UpdateTask task) {
                 }
                 // failing to apply a cluster state with an exception indicates a bug in validation or in one of the appliers; if we
                 // continue we will retry with the same cluster state but that might not help.
-                assert false;
+                assert applicationMayFail();
                 task.listener.onFailure(task.source, e);
             }
         }
@@ -664,4 +664,8 @@ protected long currentTimeInMillis() {
         return threadPool.relativeTimeInMillis();
     }
 
+    // overridden by tests that need to check behaviour in the event of an application failure
+    protected boolean applicationMayFail() {
+        return false;
+    }
 }
diff --git a/server/src/test/java/org/elasticsearch/cluster/coordination/CoordinatorTests.java b/server/src/test/java/org/elasticsearch/cluster/coordination/CoordinatorTests.java
@@ -585,6 +585,7 @@ public void testAckListenerReceivesNackFromFollower() {
             final ClusterNode follower0 = cluster.getAnyNodeExcept(leader);
             final ClusterNode follower1 = cluster.getAnyNodeExcept(leader, follower0);
 
+            follower0.allowClusterStateApplicationFailure();
             follower0.setClusterStateApplyResponse(ClusterStateApplyResponse.FAIL);
             AckCollector ackCollector = leader.submitValue(randomLong());
             cluster.stabilise(DEFAULT_CLUSTER_STATE_UPDATE_DELAY);
@@ -604,6 +605,7 @@ public void testAckListenerReceivesNackFromLeader() {
             final ClusterNode follower1 = cluster.getAnyNodeExcept(leader, follower0);
             final long startingTerm = leader.coordinator.getCurrentTerm();
 
+            leader.allowClusterStateApplicationFailure();
             leader.setClusterStateApplyResponse(ClusterStateApplyResponse.FAIL);
             AckCollector ackCollector = leader.submitValue(randomLong());
             cluster.runFor(DEFAULT_CLUSTER_STATE_UPDATE_DELAY, "committing value");
diff --git a/server/src/test/java/org/elasticsearch/cluster/service/ClusterApplierServiceTests.java b/server/src/test/java/org/elasticsearch/cluster/service/ClusterApplierServiceTests.java
@@ -359,6 +359,7 @@ public void testClusterStateApplierBubblesUpExceptionsInApplier() throws Interru
         clusterApplierService.addStateApplier(event -> {
             throw new RuntimeException("dummy exception");
         });
+        clusterApplierService.allowClusterStateApplicationFailure();
 
         CountDownLatch latch = new CountDownLatch(1);
         clusterApplierService.onNewClusterState("test", () -> ClusterState.builder(clusterApplierService.state()).build(),
@@ -387,6 +388,7 @@ public void testClusterStateApplierBubblesUpExceptionsInSettingsApplier() throws
         AtomicReference<Throwable> error = new AtomicReference<>();
         clusterApplierService.clusterSettings.addSettingsUpdateConsumer(EnableAllocationDecider.CLUSTER_ROUTING_ALLOCATION_ENABLE_SETTING,
             v -> {});
+        clusterApplierService.allowClusterStateApplicationFailure();
 
         CountDownLatch latch = new CountDownLatch(1);
         clusterApplierService.onNewClusterState("test", () -> ClusterState.builder(clusterApplierService.state())
@@ -497,6 +499,7 @@ static class TimedClusterApplierService extends ClusterApplierService {
 
         final ClusterSettings clusterSettings;
         volatile Long currentTimeOverride = null;
+        boolean applicationMayFail;
 
         TimedClusterApplierService(Settings settings, ClusterSettings clusterSettings, ThreadPool threadPool) {
             super("test_node", settings, clusterSettings, threadPool);
@@ -507,6 +510,15 @@ static class TimedClusterApplierService extends ClusterApplierService {
         protected long currentTimeInMillis() {
             return Objects.requireNonNullElseGet(currentTimeOverride, super::currentTimeInMillis);
         }
+
+        @Override
+        protected boolean applicationMayFail() {
+            return this.applicationMayFail;
+        }
+
+        void allowClusterStateApplicationFailure() {
+            this.applicationMayFail = true;
+        }
     }
 
 }
diff --git a/test/framework/src/main/java/org/elasticsearch/cluster/coordination/AbstractCoordinatorTestCase.java b/test/framework/src/main/java/org/elasticsearch/cluster/coordination/AbstractCoordinatorTestCase.java
@@ -1172,6 +1172,10 @@ void applyInitialConfiguration() {
             private boolean isNotUsefullyBootstrapped() {
                 return getLocalNode().isMasterNode() == false || coordinator.isInitialConfigurationSet() == false;
             }
+
+            void allowClusterStateApplicationFailure() {
+                clusterApplierService.allowClusterStateApplicationFailure();
+            }
         }
 
         private List<TransportAddress> provideSeedHosts(SeedHostsProvider.HostsResolver ignored) {
@@ -1282,6 +1286,7 @@ static class DisruptableClusterApplierService extends ClusterApplierService {
         private final String nodeName;
         private final DeterministicTaskQueue deterministicTaskQueue;
         ClusterStateApplyResponse clusterStateApplyResponse = ClusterStateApplyResponse.SUCCEED;
+        private boolean applicationMayFail;
 
         DisruptableClusterApplierService(String nodeName, Settings settings, ClusterSettings clusterSettings,
                                          DeterministicTaskQueue deterministicTaskQueue, Function<Runnable, Runnable> runnableWrapper) {
@@ -1326,6 +1331,15 @@ public void onNewClusterState(String source, Supplier<ClusterState> clusterState
         protected void connectToNodesAndWait(ClusterState newClusterState) {
             // don't do anything, and don't block
         }
+
+        @Override
+        protected boolean applicationMayFail() {
+            return this.applicationMayFail;
+        }
+
+        void allowClusterStateApplicationFailure() {
+            this.applicationMayFail = true;
+        }
     }
 
     protected DiscoveryNode createDiscoveryNode(int nodeIndex, boolean masterEligible) {

Original file line number	Diff line number	Diff line change
`@@ -449,7 +449,7 @@ private void runTask(UpdateTask task) {`
`449`	`449`	`}`
`450`	`450`	`// failing to apply a cluster state with an exception indicates a bug in validation or in one of the appliers; if we`
`451`	`451`	`// continue we will retry with the same cluster state but that might not help.`
`452`		`- assert false;`
	`452`	`+ assert applicationMayFail();`
`453`	`453`	`task.listener.onFailure(task.source, e);`
`454`	`454`	`}`
`455`	`455`	`}`
`@@ -664,4 +664,8 @@ protected long currentTimeInMillis() {`
`664`	`664`	`return threadPool.relativeTimeInMillis();`
`665`	`665`	`}`
`666`	`666`
	`667`	`+ // overridden by tests that need to check behaviour in the event of an application failure`
	`668`	`+ protected boolean applicationMayFail() {`
	`669`	`+ return false;`
	`670`	`+ }`
`667`	`671`	`}`