BalanceProcedureScheduler { @VisibleForTesting static String allocateJobId() { return "job-" + UUID.randomUUID(); } BalanceProcedureScheduler(Configuration conf); synchronized void init(boolean recoverJobs); synchronized void submit(BalanceJob job); BalanceJob remove(BalanceJob job); BalanceJob findJob(BalanceJob job); Collection<BalanceJob> getAllJobs(); void waitUntilDone(BalanceJob job); boolean isRunning(); synchronized void shutDown(); synchronized void shutDownAndWait(int timeout); @VisibleForTesting void setJournal(BalanceJournal journal); static final Logger LOG; }
@Test(timeout = 60000) public void testJobSerializeAndDeserialize() throws Exception { BalanceJob.Builder builder = new BalanceJob.Builder<RecordProcedure>(); for (int i = 0; i < 5; i++) { RecordProcedure r = new RecordProcedure("record-" + i, 1000L); builder.nextProcedure(r); } builder.nextProcedure(new RetryProcedure("retry", 1000, 3)); BalanceJob<RecordProcedure> job = builder.build(); job.setId(BalanceProcedureScheduler.allocateJobId()); ByteArrayOutputStream bao = new ByteArrayOutputStream(); job.write(new DataOutputStream(bao)); bao.flush(); ByteArrayInputStream bai = new ByteArrayInputStream(bao.toByteArray()); BalanceJob newJob = new BalanceJob.Builder<>().build(); newJob.readFields(new DataInputStream(bai)); assertEquals(job, newJob); }