@Override
  public NativeReader<?> create(
      CloudObject spec,
      @Nullable Coder<?> coder,
      @Nullable PipelineOptions options,
      @Nullable DataflowExecutionContext executionContext,
      DataflowOperationContext operationContext)
      throws Exception {
    coder = checkArgumentNotNull(coder);
    executionContext = checkArgumentNotNull(executionContext);
    return createImpl(spec, coder, options, executionContext, operationContext);
  }
@Test
  public void testFactory() throws Exception {
    WindowedValueCoder<?> coder =
        WindowedValue.getFullCoder(
            IsmRecordCoder.of(
                1, 0, ImmutableList.<Coder<?>>of(StringUtf8Coder.of()), VarLongCoder.of()),
            GlobalWindow.Coder.INSTANCE);

    String tmpFile = tmpFolder.newFile().getPath();
    ResourceId tmpResourceId = FileSystems.matchSingleFileSpec(tmpFile).resourceId();
    @SuppressWarnings("rawtypes")
    IsmReader<?> ismReader =
        (IsmReader)
            new IsmReaderFactory()
                .create(
                    createSpecForFilename(tmpFile),
                    coder,
                    options,
                    executionContext,
                    operationContext);
    assertEquals(coder.getValueCoder(), ismReader.getCoder());
    assertEquals(tmpResourceId, ismReader.getResourceId());
  }