public <InputT, OutputT> DoFn<InputT, OutputT> get() throws Exception {
    Thread currentThread = Thread.currentThread();
    return (DoFn<InputT, OutputT>) outstanding.get(currentThread);
  }
@Test
  public void getMultipleThreadsDifferentInstances() throws Exception {
    CountDownLatch startSignal = new CountDownLatch(1);
    ExecutorService executor = Executors.newCachedThreadPool();
    List<Future<TestFn>> futures = new ArrayList<>();
    for (int i = 0; i < 10; i++) {
      futures.add(executor.submit(new GetFnCallable(mgr, startSignal)));
    }
    startSignal.countDown();
    List<TestFn> fns = new ArrayList<>();
    for (Future<TestFn> future : futures) {
      fns.add(future.get(1L, TimeUnit.SECONDS));
    }

    for (TestFn fn : fns) {
      assertThat(fn.setupCalled, is(true));
      int sameInstances = 0;
      for (TestFn otherFn : fns) {
        if (otherFn == fn) {
          sameInstances++;
        }
      }
      assertThat(sameInstances, equalTo(1));
    }
  }