@Override
  public SinkWriter<WindowedValue<IsmRecord<V>>> writer() throws IOException {
    return new IsmSinkWriter(FileSystems.create(resourceId, MimeTypes.BINARY));
  }
@Test
  public void testWriteOutOfOrderKeysWithSameShardKeyIsError() throws Throwable {
    IsmSink<byte[]> sink =
        new IsmSink<>(
            FileSystems.matchNewResource(tmpFolder.newFile().getPath(), false),
            CODER,
            BLOOM_FILTER_SIZE_LIMIT);
    SinkWriter<WindowedValue<IsmRecord<byte[]>>> sinkWriter = sink.writer();
    sinkWriter.add(
        new ValueInEmptyWindows<>(IsmRecord.of(ImmutableList.of(EMPTY, new byte[] {0x01}), EMPTY)));

    expectedException.expect(IllegalArgumentException.class);
    expectedException.expectMessage("expects keys to be written in strictly increasing order");
    sinkWriter.add(
        new ValueInEmptyWindows<>(IsmRecord.of(ImmutableList.of(EMPTY, new byte[] {0x00}), EMPTY)));
  }