@VisibleForTesting
  static String allocateJobId() {
    return "job-" + UUID.randomUUID();
  }
@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);
  }