IsmSink extends Sink<WindowedValue<IsmRecord<V>>> { @Override public SinkWriter<WindowedValue<IsmRecord<V>>> writer() throws IOException { return new IsmSinkWriter(FileSystems.create(resourceId, MimeTypes.BINARY)); } IsmSink(ResourceId resourceId, IsmRecordCoder<V> coder, long bloomFilterSizeLimitBytes); @Override SinkWriter<WindowedValue<IsmRecord<V>>> writer();  }
@Test public void testWriteEmptyKeyWithValueLargerThanBlockSize() throws Throwable { IsmSink<byte[]> sink = new IsmSink<>( FileSystems.matchNewResource(tmpFolder.newFile().getPath(), false), IsmRecordCoder.of( 1, 0, 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(); }
@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))); }
@Test public void testWriteNonContiguousShardsIsError() 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(new byte[] {0x00}, EMPTY), EMPTY))); sinkWriter.add( new ValueInEmptyWindows<>(IsmRecord.of(ImmutableList.of(new byte[] {0x01}, EMPTY), EMPTY))); expectedException.expect(IllegalStateException.class); expectedException.expectMessage("for shard which already exists"); sinkWriter.add( new ValueInEmptyWindows<>(IsmRecord.of(ImmutableList.of(new byte[] {0x00}, EMPTY), EMPTY))); }
@Test public void testWriteEqualKeysIsError() 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[] {0x01}), EMPTY))); }
@Test public void testWriteKeyWhichIsProperPrefixOfPreviousSecondaryKeyIsError() 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[] {0x00, 0x00}), 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))); }