public static DoFnInstanceManager cloningPool(DoFnInfo<?, ?> info, PipelineOptions options) {
    return new ConcurrentQueueInstanceManager(info, options);
  }
@Test
  public void testCloningPoolReusesAfterComplete() throws Exception {
    DoFnInfo<?, ?> info =
        DoFnInfo.forFn(
            initialFn,
            WindowingStrategy.globalDefault(),
            null /* side input views */,
            null /* input coder */,
            new TupleTag<>(PropertyNames.OUTPUT) /* main output id */,
            DoFnSchemaInformation.create(),
            Collections.emptyMap());

    DoFnInstanceManager mgr = DoFnInstanceManagers.cloningPool(info, options);
    DoFnInfo<?, ?> retrievedInfo = mgr.get();
    assertThat(retrievedInfo, not(Matchers.<DoFnInfo<?, ?>>theInstance(info)));
    assertThat(retrievedInfo.getDoFn(), not(theInstance(info.getDoFn())));

    mgr.complete(retrievedInfo);
    DoFnInfo<?, ?> afterCompleteInfo = mgr.get();
    assertThat(afterCompleteInfo, Matchers.<DoFnInfo<?, ?>>theInstance(retrievedInfo));
    assertThat(afterCompleteInfo.getDoFn(), theInstance(retrievedInfo.getDoFn()));
  }