swap out snappy for LZ4

client/pom.xml

@@ -182,8 +182,8 @@
             <artifactId>bytebuffer-collections</artifactId>
         </dependency>
         <dependency>
-          <groupId>org.xerial.snappy</groupId>
+          <groupId>net.jpountz.lz4</groupId>
-          <artifactId>snappy-java</artifactId>
+          <artifactId>lz4</artifactId>
         </dependency>
 
         <!-- Tests -->

client/src/main/java/com/metamx/druid/client/cache/LZ4Transcoder.java

@@ -0,0 +1,81 @@
+/*
+ * Druid - a distributed column store.
+ * Copyright (C) 2013  Metamarkets Group Inc.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ */
+
+package com.metamx.druid.client.cache;
+
+import com.google.common.primitives.Ints;
+import net.jpountz.lz4.LZ4Compressor;
+import net.jpountz.lz4.LZ4Decompressor;
+import net.jpountz.lz4.LZ4Factory;
+import net.spy.memcached.transcoders.SerializingTranscoder;
+
+import java.nio.ByteBuffer;
+
+public class LZ4Transcoder extends SerializingTranscoder
+{
+
+  private final LZ4Factory lz4Factory;
+
+  public LZ4Transcoder()
+  {
+    super();
+    lz4Factory = LZ4Factory.fastestJavaInstance();
+  }
+
+  public LZ4Transcoder(int max)
+  {
+    super(max);
+    lz4Factory = LZ4Factory.fastestJavaInstance();
+  }
+
+  @Override
+  protected byte[] compress(byte[] in)
+  {
+    if (in == null) {
+      throw new NullPointerException("Can't compress null");
+    }
+
+    LZ4Compressor compressor = lz4Factory.fastCompressor();
+
+    byte[] out = new byte[compressor.maxCompressedLength(in.length)];
+    int compressedLength = compressor.compress(in, 0, in.length, out, 0);
+
+    getLogger().debug("Compressed %d bytes to %d", in.length, compressedLength);
+
+    return ByteBuffer.allocate(Ints.BYTES + compressedLength)
+                     .putInt(in.length)
+                     .put(out, 0, compressedLength)
+                     .array();
+  }
+
+  @Override
+  protected byte[] decompress(byte[] in)
+  {
+    byte[] out = null;
+    if(in != null) {
+      LZ4Decompressor decompressor = lz4Factory.decompressor();
+
+      int size = ByteBuffer.wrap(in).getInt();
+
+      out = new byte[size];
+      decompressor.decompress(in, Ints.BYTES, out, 0, out.length);
+    }
+    return out == null ? null : out;
+  }
+}

client/src/main/java/com/metamx/druid/client/cache/MemcachedCache.java

@@ -53,7 +53,7 @@ public class MemcachedCache implements Cache
   public static MemcachedCache create(final MemcachedCacheConfig config)
   {
     try {
-      SnappyTranscoder transcoder = new SnappyTranscoder(config.getMaxObjectSize());
+      LZ4Transcoder transcoder = new LZ4Transcoder(config.getMaxObjectSize());
 
       // always use compression
       transcoder.setCompressionThreshold(0);

client/src/main/java/com/metamx/druid/client/cache/SnappyTranscoder.java

@@ -1,50 +0,0 @@
-package com.metamx.druid.client.cache;
-
-import net.spy.memcached.transcoders.SerializingTranscoder;
-import org.xerial.snappy.Snappy;
-
-import java.io.IOException;
-
-public class SnappyTranscoder extends SerializingTranscoder
-{
-  public SnappyTranscoder()
-  {
-    super();
-  }
-
-  public SnappyTranscoder(int max)
-  {
-    super(max);
-  }
-
-  @Override
-  protected byte[] compress(byte[] in)
-  {
-    if (in == null) {
-      throw new NullPointerException("Can't compress null");
-    }
-
-    byte[] out;
-    try {
-      out = Snappy.compress(in);
-    } catch(IOException e) {
-      throw new RuntimeException("IO exception compressing data", e);
-    }
-    getLogger().debug("Compressed %d bytes to %d", in.length, out.length);
-    return out;
-  }
-
-  @Override
-  protected byte[] decompress(byte[] in)
-  {
-    byte[] out = null;
-    if(in != null) {
-      try {
-        out = Snappy.uncompress(in);
-      } catch (IOException e) {
-        getLogger().warn("Failed to decompress data", e);
-      }
-    }
-    return out == null ? null : out;
-  }
-}

client/src/test/java/com/metamx/druid/client/cache/MemcachedCacheTest.java

@@ -52,8 +52,8 @@ import java.util.concurrent.TimeoutException;
  */
 public class MemcachedCacheTest
 {
-  private static final byte[] HI = "hi".getBytes();
+  private static final byte[] HI = "hiiiiiiiiiiiiiiiiiii".getBytes();
-  private static final byte[] HO = "ho".getBytes();
+  private static final byte[] HO = "hooooooooooooooooooo".getBytes();
   private MemcachedCache cache;
 
   @Before
@@ -124,7 +124,13 @@ public class MemcachedCacheTest
 class MockMemcachedClient implements MemcachedClientIF
 {
   private final ConcurrentMap<String, CachedData> theMap = new ConcurrentHashMap<String, CachedData>();
-  private final Transcoder<Object> transcoder = new SerializingTranscoder();
+  private final SerializingTranscoder transcoder;
+
+  public MockMemcachedClient()
+  {
+    transcoder = new LZ4Transcoder();
+    transcoder.setCompressionThreshold(0);
+  }
 
   @Override
   public Collection<SocketAddress> getAvailableServers()

pom.xml

@@ -321,9 +321,9 @@
                 <version>1.9.11</version>
             </dependency>
             <dependency>
-                <groupId>org.xerial.snappy</groupId>
+                <groupId>net.jpountz.lz4</groupId>
-                <artifactId>snappy-java</artifactId>
+                <artifactId>lz4</artifactId>
-                <version>1.0.5</version>
+                <version>1.1.2</version>
             </dependency>