### Question:
WaitForTopologySettleStage extends AbstractDistStage { @Override public DistStageAck executeOnWorker() { if (!isServiceRunning()) { return successfulResponse(); } long startTime = TimeService.currentTimeMillis(); for (; ; ) { long now = TimeService.currentTimeMillis(); if (now >= startTime + timeout) { return errorResponse("The topology has not settled within timeout."); } boolean settled = true; if (checkEvents.contains(HistoryType.TOPOLOGY)) { settled = settled && checkEventHistory(history.getTopologyChangeHistory(cacheName), "Topology change", now); } if (checkEvents.contains(HistoryType.REHASH)) { settled = settled && checkEventHistory(history.getRehashHistory(cacheName), "Rehash", now); } if (checkEvents.contains(HistoryType.CACHE_STATUS)) { settled = settled && checkEventHistory(history.getCacheStatusChangeHistory(cacheName), "Cache status change", now); } if (checkMembership) { List<Clustered.Membership> membershipHistory = clustered == null ? null : clustered.getMembershipHistory(); settled = settled && checkMembership(membershipHistory, now); } if (settled) { return successfulResponse(); } else { try { Thread.sleep(1000); } catch (InterruptedException e) { return errorResponse("Waiting interrupted", e); } } } }  @Override DistStageAck executeOnWorker(); @Property(doc = "How long period without any change are we looking for. Default is 10 seconds.", converter = TimeConverter.class)
   public long period; @Property(doc = "How long should we wait until we give up with error, 0 means indefinitely. Default is 10 minutes.", converter = TimeConverter.class)
   public long timeout; @Property(doc = "Name of the cache where we detect the events. Default is the default cache.")
   public String cacheName; @Property(doc = "Type of events to check in this stage. Default are TOPOLOGY, REHASH, CACHE_STATUS (see org.radargun.traits.TopologyHistory.HistoryType).")
   public EnumSet<TopologyHistory.HistoryType> checkEvents; @Property(doc = "Wait for cluster membership to settle. Default is true (if the Clustered trait is supported).")
   public boolean checkMembership; }### Answer:
@Test(timeOut = 10_000) public void testExecuteOnWorkerMembershipChanges() throws NoSuchFieldException, IllegalAccessException { WaitForTopologySettleStage stage = initStage(EnumSet.noneOf(HistoryType.class), true); stage.period = 10l; Clustered clustered = mock(Clustered.class); setClassProperty(WaitForTopologySettleStage.class, stage, "clustered", clustered); doReturn(true).when(stage).isServiceRunning(); Clustered.Membership membershipEvent1 = new Clustered.Membership(new Date(1), mock(List.class)); Clustered.Membership membershipEvent2 = new Clustered.Membership(new Date(2), mock(List.class)); List<Clustered.Membership> membershipList = new ArrayList<>(2); membershipList.add(membershipEvent1); membershipList.add(membershipEvent2); doReturn(membershipList).when(clustered).getMembershipHistory(); PowerMockito.mockStatic(TimeService.class); PowerMockito.when(TimeService.currentTimeMillis()).thenReturn(10l).thenReturn(10l).thenReturn(13l); DistStageAck distStageAck = stage.executeOnWorker(); assertFalse(distStageAck.isError()); verify(clustered, times(2)).getMembershipHistory(); }