DoFnInstanceManagers { public static DoFnInstanceManager cloningPool(DoFnInfo<?, ?> info) { return new ConcurrentQueueInstanceManager(info); }  static DoFnInstanceManager cloningPool(DoFnInfo<?, ?> info); static DoFnInstanceManager singleInstance(DoFnInfo<?, ?> info);  }
@Test public void testCloningPoolReusesAfterComplete() throws Exception { DoFnInfo<?, ?> info = DoFnInfo.forFn( initialFn, WindowingStrategy.globalDefault(), null , null , new TupleTag<>(PropertyNames.OUTPUT) , DoFnSchemaInformation.create(), Collections.emptyMap()); DoFnInstanceManager mgr = DoFnInstanceManagers.cloningPool(info); 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())); }
@Test public void testCloningPoolTearsDownAfterAbort() throws Exception { DoFnInfo<?, ?> info = DoFnInfo.forFn( initialFn, WindowingStrategy.globalDefault(), null , null , new TupleTag<>(PropertyNames.OUTPUT) , DoFnSchemaInformation.create(), Collections.emptyMap()); DoFnInstanceManager mgr = DoFnInstanceManagers.cloningPool(info); 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)); }
@Test public void testCloningPoolMultipleOutstanding() throws Exception { DoFnInfo<?, ?> info = DoFnInfo.forFn( initialFn, WindowingStrategy.globalDefault(), null , null , new TupleTag<>(PropertyNames.OUTPUT) , DoFnSchemaInformation.create(), Collections.emptyMap()); DoFnInstanceManager mgr = DoFnInstanceManagers.cloningPool(info); DoFnInfo<?, ?> firstInfo = mgr.get(); DoFnInfo<?, ?> secondInfo = mgr.get(); assertThat(firstInfo, not(Matchers.<DoFnInfo<?, ?>>theInstance(secondInfo))); assertThat(firstInfo.getDoFn(), not(theInstance(secondInfo.getDoFn()))); }