@Override
  public ParDoFn create(
      PipelineOptions options,
      CloudObject cloudUserFn,
      List<SideInputInfo> sideInputInfos,
      TupleTag<?> mainOutputTag,
      Map<TupleTag<?>, Integer> outputTupleTagsToReceiverIndices,
      DataflowExecutionContext<?> executionContext,
      DataflowOperationContext operationContext)
      throws Exception {
    DataflowStepContext stepContext = executionContext.getStepContext(operationContext);
    checkArgument(
        stepContext instanceof StreamingModeExecutionContext.StreamingModeStepContext,
        "stepContext must be a StreamingModeStepContext to use StreamingPCollectionViewWriterFn");

    Coder<?> coder =
        CloudObjects.coderFromCloudObject(
            CloudObject.fromSpec(Structs.getObject(cloudUserFn, PropertyNames.ENCODING)));
    checkState(
        coder instanceof FullWindowedValueCoder,
        "Expected to received an instanceof an %s but got %s",
        FullWindowedValueCoder.class.getSimpleName(),
        coder);
    FullWindowedValueCoder<?> windowedValueCoder = (FullWindowedValueCoder<?>) coder;
    return new StreamingPCollectionViewWriterParDoFn(
        (StreamingModeExecutionContext.StreamingModeStepContext) stepContext,
        new TupleTag<>(Structs.getString(cloudUserFn, WorkerPropertyNames.SIDE_INPUT_ID)),
        (Coder) windowedValueCoder.getValueCoder(),
        (Coder) windowedValueCoder.getWindowCoder());
  }
@Test
  public void testConstruction() throws Exception {
    DataflowOperationContext mockOperationContext = Mockito.mock(DataflowOperationContext.class);
    DataflowExecutionContext mockExecutionContext = Mockito.mock(DataflowExecutionContext.class);
    DataflowStepContext mockStepContext =
        Mockito.mock(StreamingModeExecutionContext.StepContext.class);
    when(mockExecutionContext.getStepContext(mockOperationContext)).thenReturn(mockStepContext);

    CloudObject coder =
        CloudObjects.asCloudObject(
            WindowedValue.getFullCoder(BigEndianIntegerCoder.of(), GlobalWindow.Coder.INSTANCE),
            /*sdkComponents=*/ null);
    ParDoFn parDoFn =
        new StreamingPCollectionViewWriterDoFnFactory()
            .create(
                null /* pipeline options */,
                CloudObject.fromSpec(
                    ImmutableMap.of(
                        PropertyNames.OBJECT_TYPE_NAME,
                        "StreamingPCollectionViewWriterDoFn",
                        PropertyNames.ENCODING,
                        coder,
                        WorkerPropertyNames.SIDE_INPUT_ID,
                        "test-side-input-id")),
                null /* side input infos */,
                null /* main output tag */,
                null /* output tag to receiver index */,
                mockExecutionContext,
                mockOperationContext);
    assertThat(parDoFn, instanceOf(StreamingPCollectionViewWriterParDoFn.class));
  }