public static DoFnInstanceManager cloningPool(DoFnInfo<?, ?> info, PipelineOptions options) {
    return new ConcurrentQueueInstanceManager(info, options);
  }
@Test
  public void testCloningPoolTearsDownAfterAbort() 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();

    mgr.abort(retrievedInfo);
    TestFn fn = (TestFn) retrievedInfo.getDoFn();
    assertThat(fn.tornDown, is(true));

    DoFnInfo<?, ?> afterAbortInfo = mgr.get();
    assertThat(afterAbortInfo, not(Matchers.<DoFnInfo<?, ?>>theInstance(retrievedInfo)));
    assertThat(afterAbortInfo.getDoFn(), not(theInstance(retrievedInfo.getDoFn())));
    assertThat(((TestFn) afterAbortInfo.getDoFn()).tornDown, is(false));
  }