@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());
  // Serialize.
  ByteArrayOutputStream bao = new ByteArrayOutputStream();
  job.write(new DataOutputStream(bao));
  bao.flush();
  ByteArrayInputStream bai = new ByteArrayInputStream(bao.toByteArray());
  // Deserialize.
  BalanceJob newJob = new BalanceJob.Builder<>().build();
  newJob.readFields(new DataInputStream(bai));
  assertEquals(job, newJob);
}