@Override
  protected SideInputReader getSideInputReaderForViews(
      Iterable<? extends PCollectionView<?>> views) {
    return StreamingModeSideInputReader.of(views, this);
  }
@Test
  public void testSideInputReaderReconstituted() {
    Pipeline p = Pipeline.create();

    PCollectionView<String> preview1 = p.apply(Create.of("")).apply(View.asSingleton());
    PCollectionView<String> preview2 = p.apply(Create.of("")).apply(View.asSingleton());
    PCollectionView<String> preview3 = p.apply(Create.of("")).apply(View.asSingleton());

    SideInputReader sideInputReader =
        executionContext.getSideInputReaderForViews(Arrays.asList(preview1, preview2));

    assertTrue(sideInputReader.contains(preview1));
    assertTrue(sideInputReader.contains(preview2));
    assertFalse(sideInputReader.contains(preview3));

    PCollectionView<String> view1 = SerializableUtils.ensureSerializable(preview1);
    PCollectionView<String> view2 = SerializableUtils.ensureSerializable(preview2);
    PCollectionView<String> view3 = SerializableUtils.ensureSerializable(preview3);

    assertTrue(sideInputReader.contains(view1));
    assertTrue(sideInputReader.contains(view2));
    assertFalse(sideInputReader.contains(view3));
  }