@Override
  public SinkWriter<WindowedValue<IsmRecord<V>>> writer() throws IOException {
    return new IsmSinkWriter(FileSystems.create(resourceId, MimeTypes.BINARY));
  }
@Test
  public void testWriteEmptyKeyWithValueLargerThanBlockSize() throws Throwable {
    IsmSink<byte[]> sink =
        new IsmSink<>(
            FileSystems.matchNewResource(tmpFolder.newFile().getPath(), false),
            IsmRecordCoder.of(
                1, // We hash using only the window
                0, // There are no metadata records
                // We specifically use a coder that encodes to 0 bytes.
                ImmutableList.<Coder<?>>of(VoidCoder.of()),
                ByteArrayCoder.of()),
            BLOOM_FILTER_SIZE_LIMIT);
    SinkWriter<WindowedValue<IsmRecord<byte[]>>> sinkWriter = sink.writer();
    sinkWriter.add(
        new ValueInEmptyWindows<>(
            IsmRecord.of(
                Arrays.asList(new Object[] {null}), new byte[IsmSink.BLOCK_SIZE_BYTES * 2])));
    sinkWriter.close();
  }