FnApiWindowMappingFn extends WindowMappingFn<TargetWindowT> { @Override public TargetWindowT getSideInputWindow(BoundedWindow mainWindow) { try { TargetWindowT rval = (TargetWindowT) sideInputMappingCache.get( CacheKey.create(windowMappingFn, mainWindow), () -> loadIfNeeded(windowMappingFn, mainWindow)); return rval; } catch (ExecutionException e) { throw new IllegalStateException( String.format("Unable to load side input window for %s", mainWindow), e); } } FnApiWindowMappingFn(
      IdGenerator idGenerator,
      InstructionRequestHandler instructionRequestHandler,
      ApiServiceDescriptor dataServiceApiServiceDescriptor,
      FnDataService beamFnDataService,
      FunctionSpec windowMappingFn,
      Coder<BoundedWindow> mainInputWindowCoder,
      Coder<TargetWindowT> sideInputWindowCoder); @Override TargetWindowT getSideInputWindow(BoundedWindow mainWindow);  }
@Test public void testWindowMapping() throws Exception { TestSdkHarness testSdkHarness = new TestSdkHarness(GlobalWindow.INSTANCE); FnApiWindowMappingFn windowMappingFn = new FnApiWindowMappingFn( IdGenerators.decrementingLongs(), testSdkHarness, DATA_SERVICE, testSdkHarness, WINDOW_MAPPING_SPEC, IntervalWindowCoder.of(), GlobalWindow.Coder.INSTANCE); BoundedWindow inputWindow = new IntervalWindow(Instant.now(), Duration.standardMinutes(1)); assertEquals(GlobalWindow.INSTANCE, windowMappingFn.getSideInputWindow(inputWindow)); assertEquals(inputWindow, ((KV) testSdkHarness.getInputValues().get(0).getValue()).getValue()); BoundedWindow inputWindow2 = new IntervalWindow(Instant.now(), Duration.standardMinutes(2)); assertEquals(GlobalWindow.INSTANCE, windowMappingFn.getSideInputWindow(inputWindow2)); assertEquals(inputWindow2, ((KV) testSdkHarness.getInputValues().get(1).getValue()).getValue()); assertEquals(GlobalWindow.INSTANCE, windowMappingFn.getSideInputWindow(inputWindow)); assertEquals(2, testSdkHarness.getInputValues().size()); }