DoFnInstanceManagers { public static DoFnInstanceManager singleInstance(DoFnInfo<?, ?> info) { return new SingleInstanceManager(info); }  static DoFnInstanceManager cloningPool(DoFnInfo<?, ?> info); static DoFnInstanceManager singleInstance(DoFnInfo<?, ?> info);  }
@Test public void testInstanceReturnsInstance() throws Exception { DoFnInfo<?, ?> info = DoFnInfo.forFn( initialFn, WindowingStrategy.globalDefault(), null , null , new TupleTag<>(PropertyNames.OUTPUT) , DoFnSchemaInformation.create(), Collections.emptyMap()); DoFnInstanceManager mgr = DoFnInstanceManagers.singleInstance(info); assertThat(mgr.peek(), Matchers.<DoFnInfo<?, ?>>theInstance(info)); assertThat(mgr.get(), Matchers.<DoFnInfo<?, ?>>theInstance(info)); assertThat(mgr.get(), Matchers.<DoFnInfo<?, ?>>theInstance(info)); assertThat(mgr.peek(), Matchers.<DoFnInfo<?, ?>>theInstance(info)); }
@Test public void testInstanceIgnoresAbort() throws Exception { DoFnInfo<?, ?> info = DoFnInfo.forFn( initialFn, WindowingStrategy.globalDefault(), null , null , new TupleTag<>(PropertyNames.OUTPUT) , DoFnSchemaInformation.create(), Collections.emptyMap()); DoFnInstanceManager mgr = DoFnInstanceManagers.singleInstance(info); mgr.abort(mgr.get()); mgr.abort(mgr.get()); assertThat(mgr.get(), Matchers.<DoFnInfo<?, ?>>theInstance(info)); assertThat(mgr.get().getDoFn(), theInstance(initialFn)); }