@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 testFactoryReturnsCachedInstance() throws Exception {
    Coder<?> coder =
        WindowedValue.getFullCoder(
            IsmRecordCoder.of(
                1, 0, ImmutableList.<Coder<?>>of(StringUtf8Coder.of()), VarLongCoder.of()),
            GlobalWindow.Coder.INSTANCE);

    String tmpFile = tmpFolder.newFile().getPath();
    String anotherTmpFile = tmpFolder.newFile().getPath();

    @SuppressWarnings("rawtypes")
    IsmReader<?> ismReader =
        (IsmReader)
            new IsmReaderFactory()
                .create(
                    createSpecForFilename(tmpFile),
                    coder,
                    options,
                    executionContext,
                    operationContext);
    assertSame(
        ismReader,
        new IsmReaderFactory()
            .create(
                createSpecForFilename(tmpFile),
                coder,
                options,
                executionContext,
                operationContext));
    assertNotSame(
        ismReader,
        new IsmReaderFactory()
            .create(
                createSpecForFilename(anotherTmpFile),
                coder,
                options,
                executionContext,
                operationContext));
  }