public Iterator<List<Partition>> getNewPartsIterator(int batchSize) {
    return iteratePartitions(new_parts, batchSize);
  }
@Test
  public void testGetNewPartitionsIterator() {
    int batch = 0;
    List<Partition> actual = new ArrayList<>();
    Map<Long, List<Partition>> idToParts = new HashMap<>();
    Iterator<List<Partition>> iterator = event.getNewPartsIterator(batchSize);
    while (iterator.hasNext()) {
      List<Partition> partitions = iterator.next();
      Assert.assertTrue(batchSize <=0 || partitions.size() <= batchSize);
      Long writeId = null;
      for (Partition part : partitions) {
        if (writeId == null) {
          writeId = part.getWriteId();
        } else {
          Assert.assertEquals(writeId.longValue(), part.getWriteId());
        }
      }
      idToParts.putIfAbsent(writeId, new ArrayList<>());
      idToParts.get(writeId).addAll(partitions);
      batch++;
      actual.addAll(partitions);
    }
    Assert.assertEquals(5, idToParts.get(1L).size());
    Assert.assertEquals(5, idToParts.get(2L).size());
    Assert.assertEquals(writeIdToParts, idToParts);
    Assert.assertEquals(expectedBatch, batch);
    Assert.assertEquals(expectedParts, actual);
  }