You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by lc...@apache.org on 2016/07/06 17:58:29 UTC
[1/2] incubator-beam git commit: AvroSource: use a 64K buffer size
for Snappy codec
Repository: incubator-beam
Updated Branches:
refs/heads/master 3bb78cb8e -> a7e8151dd
AvroSource: use a 64K buffer size for Snappy codec
commons-compress defaults to a 32K buffer size for Snappy.
However, Avro uses xerial.snappy to write, which has a 64K buffer size.
When the buffer size is too small, decoding data from Snappy can cause
an EOF exception rather than finishing data.
This fixes BEAM-422.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/46ab9b1d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/46ab9b1d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/46ab9b1d
Branch: refs/heads/master
Commit: 46ab9b1d64b7f916aa6459e523484072b97ea0ad
Parents: 3bb78cb
Author: Dan Halperin <dh...@google.com>
Authored: Sat Jul 2 02:18:07 2016 -0700
Committer: Luke Cwik <lc...@google.com>
Committed: Wed Jul 6 10:57:45 2016 -0700
----------------------------------------------------------------------
.../main/java/org/apache/beam/sdk/io/AvroSource.java | 2 +-
.../java/org/apache/beam/sdk/io/AvroSourceTest.java | 15 +++++++++++----
2 files changed, 12 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/46ab9b1d/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroSource.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroSource.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroSource.java
index 22e9dfe..d1e0c4d 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroSource.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroSource.java
@@ -393,7 +393,7 @@ public class AvroSource<T> extends BlockBasedSource<T> {
ByteArrayInputStream byteStream = new ByteArrayInputStream(data);
switch (codec) {
case DataFileConstants.SNAPPY_CODEC:
- return new SnappyCompressorInputStream(byteStream);
+ return new SnappyCompressorInputStream(byteStream, 1 << 16 /* Avro uses 64KB blocks */);
case DataFileConstants.DEFLATE_CODEC:
// nowrap == true: Do not expect ZLIB header or checksum, as Avro does not write them.
Inflater inflater = new Inflater(true);
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/46ab9b1d/sdks/java/core/src/test/java/org/apache/beam/sdk/io/AvroSourceTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/AvroSourceTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/AvroSourceTest.java
index 13f8e7f..af17a95 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/io/AvroSourceTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/io/AvroSourceTest.java
@@ -133,10 +133,17 @@ public class AvroSourceTest {
@Test
public void testReadWithDifferentCodecs() throws Exception {
// Test reading files generated using all codecs.
- String codecs[] = {DataFileConstants.NULL_CODEC, DataFileConstants.BZIP2_CODEC,
- DataFileConstants.DEFLATE_CODEC, DataFileConstants.SNAPPY_CODEC,
- DataFileConstants.XZ_CODEC};
- List<Bird> expected = createRandomRecords(DEFAULT_RECORD_COUNT);
+ String codecs[] = {
+ DataFileConstants.NULL_CODEC,
+ DataFileConstants.BZIP2_CODEC,
+ DataFileConstants.DEFLATE_CODEC,
+ DataFileConstants.SNAPPY_CODEC,
+ DataFileConstants.XZ_CODEC,
+ };
+ // As Avro's default block size is 64KB, write 64K records to ensure at least one full block.
+ // We could make this smaller than 64KB assuming each record is at least B bytes, but then the
+ // test could silently stop testing the failure condition from BEAM-422.
+ List<Bird> expected = createRandomRecords(1 << 16);
for (String codec : codecs) {
String filename = generateTestFile(
[2/2] incubator-beam git commit: [BEAM-422] AvroSource: use a 64K
buffer size for Snappy codec
Posted by lc...@apache.org.
[BEAM-422] AvroSource: use a 64K buffer size for Snappy codec
This closes #583
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/a7e8151d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/a7e8151d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/a7e8151d
Branch: refs/heads/master
Commit: a7e8151ddc249047994c1835eab79f271b7f3b3b
Parents: 3bb78cb 46ab9b1
Author: Luke Cwik <lc...@google.com>
Authored: Wed Jul 6 10:58:19 2016 -0700
Committer: Luke Cwik <lc...@google.com>
Committed: Wed Jul 6 10:58:19 2016 -0700
----------------------------------------------------------------------
.../main/java/org/apache/beam/sdk/io/AvroSource.java | 2 +-
.../java/org/apache/beam/sdk/io/AvroSourceTest.java | 15 +++++++++++----
2 files changed, 12 insertions(+), 5 deletions(-)
----------------------------------------------------------------------