### Question:
CassandraScheduler implements Scheduler { @Override public void disconnected(final SchedulerDriver driver) { if (LOGGER.isDebugEnabled()) { LOGGER.debug("disconnected(driver : {})", driver); } } CassandraScheduler(
        @NotNull final PersistedCassandraFrameworkConfiguration configuration,
        @NotNull final CassandraCluster cassandraCluster,
        @NotNull final Clock clock
    ); @Override void registered(final SchedulerDriver driver, final FrameworkID frameworkId, final MasterInfo masterInfo); @Override void reregistered(final SchedulerDriver driver, final MasterInfo masterInfo); @Override void resourceOffers(final SchedulerDriver driver, final List<Offer> offers); @Override void offerRescinded(final SchedulerDriver driver, final OfferID offerId); @Override void statusUpdate(final SchedulerDriver driver, final TaskStatus status); @Override void frameworkMessage(final SchedulerDriver driver, final ExecutorID executorId, final SlaveID slaveId, final byte[] data); @Override void disconnected(final SchedulerDriver driver); @Override void slaveLost(final SchedulerDriver driver, final SlaveID slaveId); @Override void executorLost(final SchedulerDriver driver, final ExecutorID executorId, final SlaveID slaveId, final int status); @Override void error(final SchedulerDriver driver, final String message); @NotNull static String getTaskName(@Nullable final String taskName, @NotNull final String taskIdValue);  }### Answer:
@Test public void testReregistration() throws Exception { threeNodeCluster(); List<CassandraFrameworkProtos.CassandraNode> nodes = cluster.getClusterState().nodes(); assertThat(nodes).hasSize(3); for (final CassandraFrameworkProtos.CassandraNode node : nodes) { assertNotNull(node); final CassandraFrameworkProtos.CassandraNodeExecutor exec = node.getCassandraNodeExecutor(); assertNotNull(exec); assertThat(nodes).isNotEmpty(); for (final CassandraFrameworkProtos.FileDownload down : exec.getDownloadList()) { assertThat(down.getDownloadUrl()).startsWith("http: } } scheduler.disconnected(driver); cluster = new CassandraCluster(new SystemClock(), "http: new ExecutorCounter(state, 0L), new PersistedCassandraClusterState(state), new PersistedCassandraClusterHealthCheckHistory(state), new PersistedCassandraClusterJobs(state), configuration, new SeedManager(configuration, new ObjectMapper(), new SystemClock()) ); clusterState = cluster.getClusterState(); scheduler = new CassandraScheduler(configuration, cluster, clock); driver = new MockSchedulerDriver(scheduler); driver.callReRegistered(); nodes = cluster.getClusterState().nodes(); assertThat(nodes).hasSize(3); for (final CassandraFrameworkProtos.CassandraNode node : nodes) { assertNotNull(node); final CassandraFrameworkProtos.CassandraNodeExecutor exec = node.getCassandraNodeExecutor(); assertNotNull(exec); assertThat(nodes).isNotEmpty(); for (final CassandraFrameworkProtos.FileDownload down : exec.getDownloadList()) { assertThat(down.getDownloadUrl()).startsWith("http: } } }