StreamingPCollectionViewWriterDoFnFactory implements ParDoFnFactory { @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()); }  @Override ParDoFn create(
      PipelineOptions options,
      CloudObject cloudUserFn,
      List<SideInputInfo> sideInputInfos,
      TupleTag<?> mainOutputTag,
      Map<TupleTag<?>, Integer> outputTupleTagsToReceiverIndices,
      DataflowExecutionContext<?> executionContext,
      DataflowOperationContext operationContext);  }
@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), null); ParDoFn parDoFn = new StreamingPCollectionViewWriterDoFnFactory() .create( null , CloudObject.fromSpec( ImmutableMap.of( PropertyNames.OBJECT_TYPE_NAME, "StreamingPCollectionViewWriterDoFn", PropertyNames.ENCODING, coder, WorkerPropertyNames.SIDE_INPUT_ID, "test-side-input-id")), null , null , null , mockExecutionContext, mockOperationContext); assertThat(parDoFn, instanceOf(StreamingPCollectionViewWriterParDoFn.class)); }