List<BatchInterface> makeRewriteBatches(LinkedList<RewriteOp> rewrites) throws IOException {
    List<BatchInterface> batches = new ArrayList<>();
    @Nullable BatchInterface opBatch = null;
    boolean useSeparateRewriteDataBatch = this.rewriteDataOpBatchLimit != MAX_REQUESTS_PER_BATCH;
    Iterator<RewriteOp> it = rewrites.iterator();
    List<RewriteOp> deferredRewriteDataOps = new ArrayList<>();
    while (it.hasNext()) {
      RewriteOp rewrite = it.next();
      if (!rewrite.getReadyToEnqueue()) {
        it.remove();
        continue;
      }
      if (useSeparateRewriteDataBatch && !rewrite.isMetadataOperation()) {
        deferredRewriteDataOps.add(rewrite);
      } else {
        if (opBatch != null && opBatch.size() >= MAX_REQUESTS_PER_BATCH) {
          opBatch = null;
        }
        if (opBatch == null) {
          opBatch = batchRequestSupplier.get();
          batches.add(opBatch);
        }
        rewrite.enqueue(opBatch);
      }
    }
    for (RewriteOp rewrite : deferredRewriteDataOps) {
      if (opBatch != null && opBatch.size() >= this.rewriteDataOpBatchLimit) {
        opBatch = null;
      }
      if (opBatch == null) {
        opBatch = batchRequestSupplier.get();
        batches.add(opBatch);
      }
      rewrite.enqueue(opBatch);
    }
    return batches;
  }
@Test
  public void testMakeRewriteBatches() throws IOException {
    GcsUtil gcsUtil = gcsOptionsWithTestCredential().getGcsUtil();

    // Small number of files fits in 1 batch
    List<BatchInterface> batches =
        gcsUtil.makeRewriteBatches(
            gcsUtil.makeRewriteOps(makeStrings("s", 3), makeStrings("d", 3), false, false, false));
    assertThat(batches.size(), equalTo(1));
    assertThat(sumBatchSizes(batches), equalTo(3));

    // 1 batch of files fits in 1 batch
    batches =
        gcsUtil.makeRewriteBatches(
            gcsUtil.makeRewriteOps(
                makeStrings("s", 100), makeStrings("d", 100), false, false, false));
    assertThat(batches.size(), equalTo(1));
    assertThat(sumBatchSizes(batches), equalTo(100));

    // A little more than 5 batches of files fits in 6 batches
    batches =
        gcsUtil.makeRewriteBatches(
            gcsUtil.makeRewriteOps(
                makeStrings("s", 501), makeStrings("d", 501), false, false, false));
    assertThat(batches.size(), equalTo(6));
    assertThat(sumBatchSizes(batches), equalTo(501));
  }