From 57023c8ba93a5c43793c4e109e1c020d00eebc80 Mon Sep 17 00:00:00 2001
From: Shay Banon <kimchy@gmail.com>
Date: Wed, 4 Jul 2012 17:14:12 +0200
Subject: [PATCH] Compression: Support snappy as a compression option, closes
 #2081.

---
 pom.xml                                       |  10 +-
 src/main/assemblies/common-bin.xml            |   1 +
 .../client/transport/TransportClient.java     |   3 +
 .../elasticsearch/cluster/ClusterState.java   |   2 +-
 .../common/compress/CompressedIndexInput.java |  23 ++--
 .../compress/CompressedIndexOutput.java       |  13 +-
 .../compress/CompressedStreamInput.java       |   9 +-
 .../compress/CompressedStreamOutput.java      |  13 +-
 .../common/compress/Compressor.java           |   3 +
 .../common/compress/CompressorContext.java    |  25 ++++
 .../common/compress/CompressorFactory.java    |  51 ++++++-
 .../compress/lzf/LZFCompressedIndexInput.java |   5 +-
 .../lzf/LZFCompressedIndexOutput.java         |   5 +-
 .../lzf/LZFCompressedStreamInput.java         |   7 +-
 .../lzf/LZFCompressedStreamOutput.java        |   5 +-
 .../common/compress/lzf/LZFCompressor.java    |   5 +
 .../compress/lzf/LZFCompressorContext.java    |  29 ++++
 .../snappy/SnappyCompressedIndexInput.java    |  67 +++++++++
 .../snappy/SnappyCompressedIndexOutput.java   |  64 +++++++++
 .../snappy/SnappyCompressedStreamInput.java   |  72 ++++++++++
 .../snappy/SnappyCompressedStreamOutput.java  |  64 +++++++++
 .../compress/snappy/SnappyCompressor.java     | 128 ++++++++++++++++++
 .../snappy/SnappyCompressorContext.java       |  43 ++++++
 .../snappy/UnavailableSnappyCompressor.java   |  77 +++++++++++
 .../compress/snappy/xerial/XerialSnappy.java  |  45 ++++++
 .../XerialSnappyCompressedIndexInput.java     |  49 +++++++
 .../XerialSnappyCompressedIndexOutput.java    |  51 +++++++
 .../XerialSnappyCompressedStreamInput.java    |  54 ++++++++
 .../XerialSnappyCompressedStreamOutput.java   |  51 +++++++
 .../snappy/xerial/XerialSnappyCompressor.java |  68 ++++++++++
 .../org/elasticsearch/common/io/Streams.java  |   4 +-
 .../common/io/stream/AdapterStreamOutput.java |   2 +-
 .../common/io/stream/CachedStreamOutput.java  |  31 ++---
 .../common/io/stream/HandlesStreamOutput.java |  10 +-
 .../zen/ping/multicast/MulticastZenPing.java  |   2 +-
 .../publish/PublishClusterStateAction.java    |   2 +-
 .../gateway/blobstore/BlobStoreGateway.java   |   4 +-
 .../state/meta/LocalGatewayMetaState.java     |   4 +-
 .../state/shards/LocalGatewayShardsState.java |   2 +-
 .../index/mapper/core/BinaryFieldMapper.java  |   2 +-
 .../mapper/internal/SourceFieldMapper.java    |  12 +-
 .../index/translog/fs/FsTranslog.java         |   2 +-
 .../node/internal/InternalNode.java           |   3 +
 .../action/support/RestXContentBuilder.java   |   2 +-
 .../transport/local/LocalTransport.java       |   2 +-
 .../local/LocalTransportChannel.java          |   7 +-
 .../netty/NettyTransportChannel.java          |   5 +-
 .../transport/support/TransportStreams.java   |   8 +-
 .../compress/LuceneCompressionBenchmark.java  |  29 ++--
 .../compress/SearchSourceCompressTests.java   |  29 +++-
 .../CompressIndexInputOutputTests.java        |  86 +++++++++---
 .../compress/CompressedStringTests.java       |  14 +-
 .../common/io/streams/BytesStreamsTests.java  |   2 +-
 53 files changed, 1173 insertions(+), 133 deletions(-)
 create mode 100644 src/main/java/org/elasticsearch/common/compress/CompressorContext.java
 create mode 100644 src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressorContext.java
 create mode 100644 src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressedIndexInput.java
 create mode 100644 src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressedIndexOutput.java
 create mode 100644 src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressedStreamInput.java
 create mode 100644 src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressedStreamOutput.java
 create mode 100644 src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressor.java
 create mode 100644 src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressorContext.java
 create mode 100644 src/main/java/org/elasticsearch/common/compress/snappy/UnavailableSnappyCompressor.java
 create mode 100644 src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappy.java
 create mode 100644 src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappyCompressedIndexInput.java
 create mode 100644 src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappyCompressedIndexOutput.java
 create mode 100644 src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappyCompressedStreamInput.java
 create mode 100644 src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappyCompressedStreamOutput.java
 create mode 100644 src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappyCompressor.java

diff --git a/pom.xml b/pom.xml
index b3ccc199223..fa9b30bd4ed 100644
--- a/pom.xml
+++ b/pom.xml
@@ -184,6 +184,14 @@
             <optional>true</optional>
         </dependency>
 
+        <dependency>
+            <groupId>org.xerial.snappy</groupId>
+            <artifactId>snappy-java</artifactId>
+            <version>1.0.4.1</version>
+            <scope>compile</scope>
+            <optional>true</optional>
+        </dependency>
+
         <!-- We don't use this since the publish pom is then messed up -->
         <!--
         <dependency>
@@ -494,7 +502,7 @@
                                 </data>
                                 <data>
                                     <src>${project.build.directory}/lib</src>
-                                    <includes>lucene*, log4j*, jna*</includes>
+                                    <includes>lucene*, log4j*, jna*, snappy-java-*</includes>
                                     <type>directory</type>
                                     <mapper>
                                         <type>perm</type>
diff --git a/src/main/assemblies/common-bin.xml b/src/main/assemblies/common-bin.xml
index 5ef827435c8..8ed5f3fc222 100644
--- a/src/main/assemblies/common-bin.xml
+++ b/src/main/assemblies/common-bin.xml
@@ -7,6 +7,7 @@
                 <include>org.apache.lucene:lucene*</include>
                 <include>log4j:log4j</include>
                 <include>net.java.dev.jna:jna</include>
+                <include>org.xerial.snappy:snappy-java</include>
             </includes>
         </dependencySet>
         <dependencySet>
diff --git a/src/main/java/org/elasticsearch/client/transport/TransportClient.java b/src/main/java/org/elasticsearch/client/transport/TransportClient.java
index 21472577c54..977e44d6bd9 100644
--- a/src/main/java/org/elasticsearch/client/transport/TransportClient.java
+++ b/src/main/java/org/elasticsearch/client/transport/TransportClient.java
@@ -50,6 +50,7 @@ import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.CacheRecycler;
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.component.LifecycleComponent;
+import org.elasticsearch.common.compress.CompressorFactory;
 import org.elasticsearch.common.inject.Injector;
 import org.elasticsearch.common.inject.ModulesBuilder;
 import org.elasticsearch.common.io.CachedStreams;
@@ -155,6 +156,8 @@ public class TransportClient extends AbstractClient {
         this.pluginsService = new PluginsService(tuple.v1(), tuple.v2());
         this.settings = pluginsService.updatedSettings();
 
+        CompressorFactory.configure(settings);
+
         ModulesBuilder modules = new ModulesBuilder();
         modules.add(new PluginsModule(settings, pluginsService));
         modules.add(new EnvironmentModule(environment));
diff --git a/src/main/java/org/elasticsearch/cluster/ClusterState.java b/src/main/java/org/elasticsearch/cluster/ClusterState.java
index 69eecccc10c..90352269b7c 100644
--- a/src/main/java/org/elasticsearch/cluster/ClusterState.java
+++ b/src/main/java/org/elasticsearch/cluster/ClusterState.java
@@ -224,7 +224,7 @@ public class ClusterState {
         public static byte[] toBytes(ClusterState state) throws IOException {
             CachedStreamOutput.Entry cachedEntry = CachedStreamOutput.popEntry();
             try {
-                BytesStreamOutput os = cachedEntry.cachedBytes();
+                BytesStreamOutput os = cachedEntry.bytes();
                 writeTo(state, os);
                 return os.copiedByteArray();
             } finally {
diff --git a/src/main/java/org/elasticsearch/common/compress/CompressedIndexInput.java b/src/main/java/org/elasticsearch/common/compress/CompressedIndexInput.java
index 3ef0a203ba6..c127f8dd612 100644
--- a/src/main/java/org/elasticsearch/common/compress/CompressedIndexInput.java
+++ b/src/main/java/org/elasticsearch/common/compress/CompressedIndexInput.java
@@ -27,31 +27,34 @@ import java.io.IOException;
 
 /**
  */
-public abstract class CompressedIndexInput extends IndexInput {
+public abstract class CompressedIndexInput<T extends CompressorContext> extends IndexInput {
 
     private IndexInput in;
+    protected final T context;
 
     private int version;
-    private long uncompressedLength;
+    private long totalUncompressedLength;
     private BigLongArray offsets;
 
     private boolean closed;
 
     protected byte[] uncompressed;
+    protected int uncompressedLength;
     private int position = 0;
     private int valid = 0;
     private int currentOffsetIdx;
     private long currentUncompressedChunkPointer;
 
-    public CompressedIndexInput(IndexInput in) throws IOException {
+    public CompressedIndexInput(IndexInput in, T context) throws IOException {
         super("compressed(" + in.toString() + ")");
         this.in = in;
+        this.context = context;
         readHeader(in);
         this.version = in.readInt();
         long metaDataPosition = in.readLong();
         long headerLength = in.getFilePointer();
         in.seek(metaDataPosition);
-        this.uncompressedLength = in.readVLong();
+        this.totalUncompressedLength = in.readVLong();
         int size = in.readVInt();
         offsets = new BigLongArray(size);
         for (int i = 0; i < size; i++) {
@@ -133,7 +136,7 @@ public abstract class CompressedIndexInput extends IndexInput {
 
     @Override
     public void seek(long pos) throws IOException {
-        int idx = (int) (pos / uncompressed.length);
+        int idx = (int) (pos / uncompressedLength);
         if (idx >= offsets.size) {
             // set the next "readyBuffer" to EOF
             currentOffsetIdx = idx;
@@ -151,12 +154,12 @@ public abstract class CompressedIndexInput extends IndexInput {
             currentOffsetIdx = idx - 1; // we are going to increase it in readyBuffer...
             readyBuffer();
         }
-        position = (int) (pos % uncompressed.length);
+        position = (int) (pos % uncompressedLength);
     }
 
     @Override
     public long length() {
-        return uncompressedLength;
+        return totalUncompressedLength;
     }
 
     @Override
@@ -187,7 +190,7 @@ public abstract class CompressedIndexInput extends IndexInput {
             return false;
         }
         currentOffsetIdx++;
-        currentUncompressedChunkPointer = ((long) currentOffsetIdx) * uncompressed.length;
+        currentUncompressedChunkPointer = ((long) currentOffsetIdx) * uncompressedLength;
         position = 0;
         return (position < valid);
     }
@@ -203,8 +206,8 @@ public abstract class CompressedIndexInput extends IndexInput {
     public Object clone() {
         // we clone and we need to make sure we keep the same positions!
         CompressedIndexInput cloned = (CompressedIndexInput) super.clone();
-        cloned.uncompressed = new byte[uncompressed.length];
-        System.arraycopy(uncompressed, 0, cloned.uncompressed, 0, uncompressed.length);
+        cloned.uncompressed = new byte[uncompressedLength];
+        System.arraycopy(uncompressed, 0, cloned.uncompressed, 0, uncompressedLength);
         cloned.in = (IndexInput) cloned.in.clone();
         return cloned;
     }
diff --git a/src/main/java/org/elasticsearch/common/compress/CompressedIndexOutput.java b/src/main/java/org/elasticsearch/common/compress/CompressedIndexOutput.java
index 5ae4c83d2b4..6d5dbba8da5 100644
--- a/src/main/java/org/elasticsearch/common/compress/CompressedIndexOutput.java
+++ b/src/main/java/org/elasticsearch/common/compress/CompressedIndexOutput.java
@@ -28,11 +28,13 @@ import java.io.IOException;
 
 /**
  */
-public abstract class CompressedIndexOutput extends IndexOutput {
+public abstract class CompressedIndexOutput<T extends CompressorContext> extends IndexOutput {
 
     final IndexOutput out;
+    protected final T context;
 
     protected byte[] uncompressed;
+    protected int uncompressedLength;
     private int position = 0;
 
     private long uncompressedPosition;
@@ -43,8 +45,9 @@ public abstract class CompressedIndexOutput extends IndexOutput {
     // need to have a growing segment long array list here...
     private TLongArrayList offsets = new TLongArrayList();
 
-    public CompressedIndexOutput(IndexOutput out) throws IOException {
+    public CompressedIndexOutput(IndexOutput out, T context) throws IOException {
         this.out = out;
+        this.context = context;
         writeHeader(out);
         out.writeInt(0); // version
         metaDataPointer = out.getFilePointer();
@@ -57,7 +60,7 @@ public abstract class CompressedIndexOutput extends IndexOutput {
 
     @Override
     public void writeByte(byte b) throws IOException {
-        if (position >= uncompressed.length) {
+        if (position >= uncompressedLength) {
             flushBuffer();
         }
         uncompressedPosition++;
@@ -70,7 +73,7 @@ public abstract class CompressedIndexOutput extends IndexOutput {
         if (length == 0) {
             return;
         }
-        final int BUFFER_LEN = uncompressed.length;
+        final int BUFFER_LEN = uncompressedLength;
 
         // simple case first: buffering only (for trivially short writes)
         int free = BUFFER_LEN - position;
@@ -109,7 +112,7 @@ public abstract class CompressedIndexOutput extends IndexOutput {
 
     @Override
     public void copyBytes(DataInput input, long length) throws IOException {
-        final int BUFFER_LEN = uncompressed.length;
+        final int BUFFER_LEN = uncompressedLength;
 
         // simple case first: buffering only (for trivially short writes)
         int free = BUFFER_LEN - position;
diff --git a/src/main/java/org/elasticsearch/common/compress/CompressedStreamInput.java b/src/main/java/org/elasticsearch/common/compress/CompressedStreamInput.java
index 4acd6cf358a..8587f67dfeb 100644
--- a/src/main/java/org/elasticsearch/common/compress/CompressedStreamInput.java
+++ b/src/main/java/org/elasticsearch/common/compress/CompressedStreamInput.java
@@ -23,13 +23,13 @@ import org.elasticsearch.common.io.stream.StreamInput;
 
 import java.io.EOFException;
 import java.io.IOException;
-import java.io.InputStream;
 
 /**
  */
-public abstract class CompressedStreamInput extends StreamInput {
+public abstract class CompressedStreamInput<T extends CompressorContext> extends StreamInput {
 
     private final StreamInput in;
+    protected final CompressorContext context;
 
     private boolean closed;
 
@@ -37,8 +37,9 @@ public abstract class CompressedStreamInput extends StreamInput {
     private int position = 0;
     private int valid = 0;
 
-    public CompressedStreamInput(StreamInput in) throws IOException {
+    public CompressedStreamInput(StreamInput in, T context) throws IOException {
         this.in = in;
+        this.context = context;
         readHeader(in);
     }
 
@@ -169,6 +170,6 @@ public abstract class CompressedStreamInput extends StreamInput {
     /**
      * Uncompress the data into the out array, returning the size uncompressed
      */
-    protected abstract int uncompress(InputStream in, byte[] out) throws IOException;
+    protected abstract int uncompress(StreamInput in, byte[] out) throws IOException;
 
 }
diff --git a/src/main/java/org/elasticsearch/common/compress/CompressedStreamOutput.java b/src/main/java/org/elasticsearch/common/compress/CompressedStreamOutput.java
index eccb224c112..7f95828a1ca 100644
--- a/src/main/java/org/elasticsearch/common/compress/CompressedStreamOutput.java
+++ b/src/main/java/org/elasticsearch/common/compress/CompressedStreamOutput.java
@@ -25,23 +25,26 @@ import java.io.IOException;
 
 /**
  */
-public abstract class CompressedStreamOutput extends StreamOutput {
+public abstract class CompressedStreamOutput<T extends CompressorContext> extends StreamOutput {
 
     private final StreamOutput out;
+    protected final T context;
 
     protected byte[] uncompressed;
+    protected int uncompressedLength;
     private int position = 0;
 
     private boolean closed;
 
-    public CompressedStreamOutput(StreamOutput out) throws IOException {
+    public CompressedStreamOutput(StreamOutput out, T context) throws IOException {
         this.out = out;
+        this.context = context;
         writeHeader(out);
     }
 
     @Override
     public void write(int b) throws IOException {
-        if (position >= uncompressed.length) {
+        if (position >= uncompressedLength) {
             flushBuffer();
         }
         uncompressed[position++] = (byte) b;
@@ -49,7 +52,7 @@ public abstract class CompressedStreamOutput extends StreamOutput {
 
     @Override
     public void writeByte(byte b) throws IOException {
-        if (position >= uncompressed.length) {
+        if (position >= uncompressedLength) {
             flushBuffer();
         }
         uncompressed[position++] = b;
@@ -61,7 +64,7 @@ public abstract class CompressedStreamOutput extends StreamOutput {
         if (length == 0) {
             return;
         }
-        final int BUFFER_LEN = uncompressed.length;
+        final int BUFFER_LEN = uncompressedLength;
 
         // simple case first: buffering only (for trivially short writes)
         int free = BUFFER_LEN - position;
diff --git a/src/main/java/org/elasticsearch/common/compress/Compressor.java b/src/main/java/org/elasticsearch/common/compress/Compressor.java
index 4946a52d0a1..4675de6af53 100644
--- a/src/main/java/org/elasticsearch/common/compress/Compressor.java
+++ b/src/main/java/org/elasticsearch/common/compress/Compressor.java
@@ -23,6 +23,7 @@ import org.apache.lucene.store.IndexInput;
 import org.apache.lucene.store.IndexOutput;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.settings.Settings;
 import org.jboss.netty.buffer.ChannelBuffer;
 
 import java.io.IOException;
@@ -33,6 +34,8 @@ public interface Compressor {
 
     String type();
 
+    void configure(Settings settings);
+
     boolean isCompressed(byte[] data, int offset, int length);
 
     boolean isCompressed(ChannelBuffer buffer);
diff --git a/src/main/java/org/elasticsearch/common/compress/CompressorContext.java b/src/main/java/org/elasticsearch/common/compress/CompressorContext.java
new file mode 100644
index 00000000000..49e1eb77f04
--- /dev/null
+++ b/src/main/java/org/elasticsearch/common/compress/CompressorContext.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.common.compress;
+
+/**
+ */
+public interface CompressorContext {
+}
diff --git a/src/main/java/org/elasticsearch/common/compress/CompressorFactory.java b/src/main/java/org/elasticsearch/common/compress/CompressorFactory.java
index a4cf752cea7..9b39e6a0e71 100644
--- a/src/main/java/org/elasticsearch/common/compress/CompressorFactory.java
+++ b/src/main/java/org/elasticsearch/common/compress/CompressorFactory.java
@@ -20,14 +20,22 @@
 package org.elasticsearch.common.compress;
 
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
 import org.apache.lucene.store.IndexInput;
 import org.elasticsearch.common.BytesHolder;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.collect.MapBuilder;
 import org.elasticsearch.common.compress.lzf.LZFCompressor;
+import org.elasticsearch.common.compress.snappy.UnavailableSnappyCompressor;
+import org.elasticsearch.common.compress.snappy.xerial.XerialSnappy;
+import org.elasticsearch.common.compress.snappy.xerial.XerialSnappyCompressor;
+import org.elasticsearch.common.logging.Loggers;
+import org.elasticsearch.common.settings.Settings;
 import org.jboss.netty.buffer.ChannelBuffer;
 
 import java.io.IOException;
+import java.util.List;
+import java.util.Locale;
 
 /**
  */
@@ -37,20 +45,57 @@ public class CompressorFactory {
 
     private static final Compressor[] compressors;
     private static final ImmutableMap<String, Compressor> compressorsByType;
+    private static Compressor defaultCompressor;
 
     static {
-        compressors = new Compressor[1];
-        compressors[0] = LZF;
+        List<Compressor> compressorsX = Lists.newArrayList();
+        compressorsX.add(LZF);
+        boolean addedSnappy = false;
+        if (XerialSnappy.available) {
+            compressorsX.add(new XerialSnappyCompressor());
+            addedSnappy = true;
+        } else {
+            Loggers.getLogger(CompressorFactory.class).debug("failed to load xerial snappy-java", XerialSnappy.failure);
+        }
+        if (!addedSnappy) {
+            compressorsX.add(new UnavailableSnappyCompressor());
+        }
+
+        compressors = compressorsX.toArray(new Compressor[compressorsX.size()]);
 
         MapBuilder<String, Compressor> compressorsByTypeX = MapBuilder.newMapBuilder();
         for (Compressor compressor : compressors) {
             compressorsByTypeX.put(compressor.type(), compressor);
         }
         compressorsByType = compressorsByTypeX.immutableMap();
+
+        defaultCompressor = LZF;
+    }
+
+    public static synchronized void configure(Settings settings) {
+        for (Compressor compressor : compressors) {
+            compressor.configure(settings);
+        }
+        String defaultType = settings.get("compress.default.type", "lzf").toLowerCase(Locale.ENGLISH);
+        boolean found = false;
+        for (Compressor compressor : compressors) {
+            if (defaultType.equalsIgnoreCase(compressor.type())) {
+                defaultCompressor = compressor;
+                found = true;
+                break;
+            }
+        }
+        if (!found) {
+            Loggers.getLogger(CompressorFactory.class).warn("failed to find default type [{}]", defaultType);
+        }
+    }
+
+    public static synchronized void setDefaultCompressor(Compressor defaultCompressor) {
+        CompressorFactory.defaultCompressor = defaultCompressor;
     }
 
     public static Compressor defaultCompressor() {
-        return LZF;
+        return defaultCompressor;
     }
 
     public static boolean isCompressed(byte[] data) {
diff --git a/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressedIndexInput.java b/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressedIndexInput.java
index c9d14f7ad6a..95b12604653 100644
--- a/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressedIndexInput.java
+++ b/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressedIndexInput.java
@@ -30,17 +30,18 @@ import java.util.Arrays;
 
 /**
  */
-public class LZFCompressedIndexInput extends CompressedIndexInput {
+public class LZFCompressedIndexInput extends CompressedIndexInput<LZFCompressorContext> {
 
     private final ChunkDecoder decoder;
     // scratch area buffer
     private byte[] inputBuffer;
 
     public LZFCompressedIndexInput(IndexInput in, ChunkDecoder decoder) throws IOException {
-        super(in);
+        super(in, LZFCompressorContext.INSTANCE);
 
         this.decoder = decoder;
         this.uncompressed = new byte[LZFChunk.MAX_CHUNK_LEN];
+        this.uncompressedLength = LZFChunk.MAX_CHUNK_LEN;
         this.inputBuffer = new byte[LZFChunk.MAX_CHUNK_LEN];
     }
 
diff --git a/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressedIndexOutput.java b/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressedIndexOutput.java
index 364098f476b..84e8a62f41c 100644
--- a/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressedIndexOutput.java
+++ b/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressedIndexOutput.java
@@ -30,15 +30,16 @@ import java.io.IOException;
 
 /**
  */
-public class LZFCompressedIndexOutput extends CompressedIndexOutput {
+public class LZFCompressedIndexOutput extends CompressedIndexOutput<LZFCompressorContext> {
 
     private final BufferRecycler recycler;
     private final ChunkEncoder encoder;
 
     public LZFCompressedIndexOutput(IndexOutput out) throws IOException {
-        super(out);
+        super(out, LZFCompressorContext.INSTANCE);
         this.recycler = BufferRecycler.instance();
         this.uncompressed = this.recycler.allocOutputBuffer(LZFChunk.MAX_CHUNK_LEN);
+        this.uncompressedLength = LZFChunk.MAX_CHUNK_LEN;
         this.encoder = new ChunkEncoder(LZFChunk.MAX_CHUNK_LEN);
     }
 
diff --git a/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressedStreamInput.java b/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressedStreamInput.java
index 78b8ae985b5..68b20359cb6 100644
--- a/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressedStreamInput.java
+++ b/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressedStreamInput.java
@@ -26,11 +26,10 @@ import org.elasticsearch.common.compress.CompressedStreamInput;
 import org.elasticsearch.common.io.stream.StreamInput;
 
 import java.io.IOException;
-import java.io.InputStream;
 
 /**
  */
-public class LZFCompressedStreamInput extends CompressedStreamInput {
+public class LZFCompressedStreamInput extends CompressedStreamInput<LZFCompressorContext> {
 
     private final BufferRecycler recycler;
 
@@ -40,7 +39,7 @@ public class LZFCompressedStreamInput extends CompressedStreamInput {
     private byte[] inputBuffer;
 
     public LZFCompressedStreamInput(StreamInput in, ChunkDecoder decoder) throws IOException {
-        super(in);
+        super(in, LZFCompressorContext.INSTANCE);
         this.recycler = BufferRecycler.instance();
         this.decoder = decoder;
 
@@ -54,7 +53,7 @@ public class LZFCompressedStreamInput extends CompressedStreamInput {
     }
 
     @Override
-    public int uncompress(InputStream in, byte[] out) throws IOException {
+    public int uncompress(StreamInput in, byte[] out) throws IOException {
         return decoder.decodeChunk(in, inputBuffer, out);
     }
 
diff --git a/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressedStreamOutput.java b/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressedStreamOutput.java
index af58fde006d..76af131241b 100644
--- a/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressedStreamOutput.java
+++ b/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressedStreamOutput.java
@@ -29,15 +29,16 @@ import java.io.IOException;
 
 /**
  */
-public class LZFCompressedStreamOutput extends CompressedStreamOutput {
+public class LZFCompressedStreamOutput extends CompressedStreamOutput<LZFCompressorContext> {
 
     private final BufferRecycler recycler;
     private final ChunkEncoder encoder;
 
     public LZFCompressedStreamOutput(StreamOutput out) throws IOException {
-        super(out);
+        super(out, LZFCompressorContext.INSTANCE);
         this.recycler = BufferRecycler.instance();
         this.uncompressed = this.recycler.allocOutputBuffer(LZFChunk.MAX_CHUNK_LEN);
+        this.uncompressedLength = LZFChunk.MAX_CHUNK_LEN;
         this.encoder = new ChunkEncoder(LZFChunk.MAX_CHUNK_LEN);
     }
 
diff --git a/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressor.java b/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressor.java
index c2ff2300562..373c1445ebb 100644
--- a/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressor.java
+++ b/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressor.java
@@ -29,6 +29,7 @@ import org.elasticsearch.common.compress.*;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.logging.Loggers;
+import org.elasticsearch.common.settings.Settings;
 import org.jboss.netty.buffer.ChannelBuffer;
 
 import java.io.IOException;
@@ -53,6 +54,10 @@ public class LZFCompressor implements Compressor {
         return TYPE;
     }
 
+    @Override
+    public void configure(Settings settings) {
+    }
+
     @Override
     public boolean isCompressed(byte[] data, int offset, int length) {
         return length >= 3 &&
diff --git a/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressorContext.java b/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressorContext.java
new file mode 100644
index 00000000000..e4a36884601
--- /dev/null
+++ b/src/main/java/org/elasticsearch/common/compress/lzf/LZFCompressorContext.java
@@ -0,0 +1,29 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.common.compress.lzf;
+
+import org.elasticsearch.common.compress.CompressorContext;
+
+/**
+ */
+public class LZFCompressorContext implements CompressorContext {
+
+    public static final LZFCompressorContext INSTANCE = new LZFCompressorContext();
+}
diff --git a/src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressedIndexInput.java b/src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressedIndexInput.java
new file mode 100644
index 00000000000..bdfeccdf304
--- /dev/null
+++ b/src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressedIndexInput.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.common.compress.snappy;
+
+import org.apache.lucene.store.IndexInput;
+import org.elasticsearch.common.compress.CompressedIndexInput;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+/**
+ */
+public abstract class SnappyCompressedIndexInput extends CompressedIndexInput<SnappyCompressorContext> {
+
+    protected int chunkSize;
+
+    protected int maxCompressedChunkLength;
+
+    protected byte[] inputBuffer;
+
+    public SnappyCompressedIndexInput(IndexInput in, SnappyCompressorContext context) throws IOException {
+        super(in, context);
+        this.uncompressed = new byte[chunkSize];
+        this.uncompressedLength = chunkSize;
+        this.inputBuffer = new byte[Math.max(chunkSize, maxCompressedChunkLength)];
+    }
+
+    @Override
+    protected void readHeader(IndexInput in) throws IOException {
+        byte[] header = new byte[SnappyCompressor.HEADER.length];
+        in.readBytes(header, 0, header.length);
+        if (!Arrays.equals(header, SnappyCompressor.HEADER)) {
+            throw new IOException("wrong snappy compressed header [" + Arrays.toString(header) + "]");
+        }
+        this.chunkSize = in.readVInt();
+        this.maxCompressedChunkLength = in.readVInt();
+    }
+
+    @Override
+    protected void doClose() throws IOException {
+        // nothing to do here
+    }
+
+    @Override
+    public Object clone() {
+        SnappyCompressedIndexInput cloned = (SnappyCompressedIndexInput) super.clone();
+        cloned.inputBuffer = new byte[inputBuffer.length];
+        return cloned;
+    }
+}
diff --git a/src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressedIndexOutput.java b/src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressedIndexOutput.java
new file mode 100644
index 00000000000..402e92b4def
--- /dev/null
+++ b/src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressedIndexOutput.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.common.compress.snappy;
+
+import com.ning.compress.BufferRecycler;
+import org.apache.lucene.store.IndexOutput;
+import org.elasticsearch.common.compress.CompressedIndexOutput;
+
+import java.io.IOException;
+
+/**
+ */
+public abstract class SnappyCompressedIndexOutput extends CompressedIndexOutput<SnappyCompressorContext> {
+
+    protected final BufferRecycler recycler;
+
+    protected byte[] compressedBuffer;
+
+    public SnappyCompressedIndexOutput(IndexOutput out, SnappyCompressorContext context) throws IOException {
+        super(out, context);
+        this.recycler = BufferRecycler.instance();
+        this.uncompressed = this.recycler.allocOutputBuffer(context.compressChunkLength());
+        this.uncompressedLength = context.compressChunkLength();
+        this.compressedBuffer = recycler.allocEncodingBuffer(context.compressMaxCompressedChunkLength());
+    }
+
+    @Override
+    protected void writeHeader(IndexOutput out) throws IOException {
+        out.writeBytes(SnappyCompressor.HEADER, SnappyCompressor.HEADER.length);
+        out.writeVInt(context.compressChunkLength());
+        out.writeVInt(context.compressMaxCompressedChunkLength());
+    }
+
+    @Override
+    protected void doClose() throws IOException {
+        byte[] buf = uncompressed;
+        if (buf != null) {
+            uncompressed = null;
+            recycler.releaseOutputBuffer(buf);
+        }
+        buf = compressedBuffer;
+        if (buf != null) {
+            compressedBuffer = null;
+            recycler.releaseEncodeBuffer(buf);
+        }
+    }
+}
diff --git a/src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressedStreamInput.java b/src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressedStreamInput.java
new file mode 100644
index 00000000000..4c73a5b6985
--- /dev/null
+++ b/src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressedStreamInput.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.common.compress.snappy;
+
+import com.ning.compress.BufferRecycler;
+import org.elasticsearch.common.compress.CompressedStreamInput;
+import org.elasticsearch.common.io.stream.StreamInput;
+
+import java.io.IOException;
+import java.util.Arrays;
+
+/**
+ */
+public abstract class SnappyCompressedStreamInput extends CompressedStreamInput<SnappyCompressorContext> {
+
+    protected final BufferRecycler recycler;
+
+    protected int chunkSize;
+
+    protected int maxCompressedChunkLength;
+
+    protected byte[] inputBuffer;
+
+    public SnappyCompressedStreamInput(StreamInput in, SnappyCompressorContext context) throws IOException {
+        super(in, context);
+        this.recycler = BufferRecycler.instance();
+        this.uncompressed = recycler.allocDecodeBuffer(Math.max(chunkSize, maxCompressedChunkLength));
+        this.inputBuffer = recycler.allocInputBuffer(Math.max(chunkSize, maxCompressedChunkLength));
+    }
+
+    @Override
+    public void readHeader(StreamInput in) throws IOException {
+        byte[] header = new byte[SnappyCompressor.HEADER.length];
+        in.readBytes(header, 0, header.length);
+        if (!Arrays.equals(header, SnappyCompressor.HEADER)) {
+            throw new IOException("wrong snappy compressed header [" + Arrays.toString(header) + "]");
+        }
+        this.chunkSize = in.readVInt();
+        this.maxCompressedChunkLength = in.readVInt();
+    }
+
+    @Override
+    protected void doClose() throws IOException {
+        byte[] buf = uncompressed;
+        if (buf != null) {
+            uncompressed = null;
+            recycler.releaseDecodeBuffer(uncompressed);
+        }
+        buf = inputBuffer;
+        if (buf != null) {
+            inputBuffer = null;
+            recycler.releaseInputBuffer(inputBuffer);
+        }
+    }
+}
diff --git a/src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressedStreamOutput.java b/src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressedStreamOutput.java
new file mode 100644
index 00000000000..8498db801f6
--- /dev/null
+++ b/src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressedStreamOutput.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.common.compress.snappy;
+
+import com.ning.compress.BufferRecycler;
+import org.elasticsearch.common.compress.CompressedStreamOutput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+
+import java.io.IOException;
+
+/**
+ */
+public abstract class SnappyCompressedStreamOutput extends CompressedStreamOutput<SnappyCompressorContext> {
+
+    protected final BufferRecycler recycler;
+
+    protected byte[] compressedBuffer;
+
+    public SnappyCompressedStreamOutput(StreamOutput out, SnappyCompressorContext context) throws IOException {
+        super(out, context);
+        this.recycler = BufferRecycler.instance();
+        this.uncompressed = this.recycler.allocOutputBuffer(context.compressChunkLength());
+        this.uncompressedLength = context.compressChunkLength();
+        this.compressedBuffer = recycler.allocEncodingBuffer(context.compressMaxCompressedChunkLength());
+    }
+
+    @Override
+    public void writeHeader(StreamOutput out) throws IOException {
+        out.writeBytes(SnappyCompressor.HEADER);
+        out.writeVInt(context.compressChunkLength());
+        out.writeVInt(context.compressMaxCompressedChunkLength());
+    }
+
+    @Override
+    protected void doClose() throws IOException {
+        byte[] buf = uncompressed;
+        if (buf != null) {
+            uncompressed = null;
+            recycler.releaseOutputBuffer(buf);
+        }
+        buf = compressedBuffer;
+        if (buf != null) {
+            compressedBuffer = null;
+            recycler.releaseEncodeBuffer(buf);
+        }
+    }
+}
diff --git a/src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressor.java b/src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressor.java
new file mode 100644
index 00000000000..5194a2a6af5
--- /dev/null
+++ b/src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressor.java
@@ -0,0 +1,128 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.common.compress.snappy;
+
+import org.apache.lucene.store.IndexInput;
+import org.elasticsearch.common.compress.Compressor;
+import org.elasticsearch.common.io.Streams;
+import org.elasticsearch.common.io.stream.BytesStreamInput;
+import org.elasticsearch.common.io.stream.CachedStreamOutput;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeValue;
+import org.jboss.netty.buffer.ChannelBuffer;
+
+import java.io.IOException;
+
+/**
+ */
+public abstract class SnappyCompressor implements Compressor {
+
+    public static final byte[] HEADER = {'s', 'n', 'a', 'p', 'p', 'y', 0};
+
+    protected SnappyCompressorContext compressorContext;
+
+    // default block size (32k)
+    static final int DEFAULT_CHUNK_SIZE = 1 << 15;
+
+    protected SnappyCompressor() {
+        this.compressorContext = new SnappyCompressorContext(DEFAULT_CHUNK_SIZE, maxCompressedLength(DEFAULT_CHUNK_SIZE));
+    }
+
+    @Override
+    public void configure(Settings settings) {
+        int chunkLength = (int) settings.getAsBytesSize("compress.snappy.chunk_size", new ByteSizeValue(compressorContext.compressChunkLength())).bytes();
+        int maxCompressedChunkLength = maxCompressedLength(chunkLength);
+        this.compressorContext = new SnappyCompressorContext(chunkLength, maxCompressedChunkLength);
+    }
+
+    protected abstract int maxCompressedLength(int length);
+
+    @Override
+    public boolean isCompressed(byte[] data, int offset, int length) {
+        if (length < HEADER.length) {
+            return false;
+        }
+        for (int i = 0; i < HEADER.length; i++) {
+            if (data[offset + i] != HEADER[i]) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    @Override
+    public boolean isCompressed(ChannelBuffer buffer) {
+        if (buffer.readableBytes() < HEADER.length) {
+            return false;
+        }
+        int offset = buffer.readerIndex();
+        for (int i = 0; i < HEADER.length; i++) {
+            if (buffer.getByte(offset + i) != HEADER[i]) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    @Override
+    public boolean isCompressed(IndexInput in) throws IOException {
+        long currentPointer = in.getFilePointer();
+        // since we have some metdata before the first compressed header, we check on our specific header
+        if (in.length() - currentPointer < (HEADER.length)) {
+            return false;
+        }
+        for (int i = 0; i < HEADER.length; i++) {
+            if (in.readByte() != HEADER[i]) {
+                in.seek(currentPointer);
+                return false;
+            }
+        }
+        in.seek(currentPointer);
+        return true;
+    }
+
+    @Override
+    public byte[] compress(byte[] data, int offset, int length) throws IOException {
+        // this needs to be the same format as regular streams reading from it!
+        CachedStreamOutput.Entry entry = CachedStreamOutput.popEntry();
+        try {
+            StreamOutput compressed = entry.bytes(this);
+            compressed.writeBytes(data, offset, length);
+            compressed.close();
+            return entry.bytes().copiedByteArray();
+        } finally {
+            CachedStreamOutput.pushEntry(entry);
+        }
+    }
+
+    @Override
+    public byte[] uncompress(byte[] data, int offset, int length) throws IOException {
+        StreamInput compressed = streamInput(new BytesStreamInput(data, offset, length, false));
+        CachedStreamOutput.Entry entry = CachedStreamOutput.popEntry();
+        try {
+            Streams.copy(compressed, entry.bytes());
+            return entry.bytes().copiedByteArray();
+        } finally {
+            CachedStreamOutput.pushEntry(entry);
+        }
+    }
+}
diff --git a/src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressorContext.java b/src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressorContext.java
new file mode 100644
index 00000000000..68d95baec5b
--- /dev/null
+++ b/src/main/java/org/elasticsearch/common/compress/snappy/SnappyCompressorContext.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.common.compress.snappy;
+
+import org.elasticsearch.common.compress.CompressorContext;
+
+/**
+ */
+public class SnappyCompressorContext implements CompressorContext {
+
+    private final int compressChunkLength;
+    private final int compressMaxCompressedChunkLength;
+
+    public SnappyCompressorContext(int compressChunkLength, int compressMaxCompressedChunkLength) {
+        this.compressChunkLength = compressChunkLength;
+        this.compressMaxCompressedChunkLength = compressMaxCompressedChunkLength;
+    }
+
+    public int compressChunkLength() {
+        return compressChunkLength;
+    }
+
+    public int compressMaxCompressedChunkLength() {
+        return compressMaxCompressedChunkLength;
+    }
+}
diff --git a/src/main/java/org/elasticsearch/common/compress/snappy/UnavailableSnappyCompressor.java b/src/main/java/org/elasticsearch/common/compress/snappy/UnavailableSnappyCompressor.java
new file mode 100644
index 00000000000..2c586dbeaa6
--- /dev/null
+++ b/src/main/java/org/elasticsearch/common/compress/snappy/UnavailableSnappyCompressor.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.common.compress.snappy;
+
+import org.apache.lucene.store.IndexInput;
+import org.apache.lucene.store.IndexOutput;
+import org.elasticsearch.ElasticSearchIllegalStateException;
+import org.elasticsearch.common.compress.CompressedIndexInput;
+import org.elasticsearch.common.compress.CompressedIndexOutput;
+import org.elasticsearch.common.compress.CompressedStreamInput;
+import org.elasticsearch.common.compress.CompressedStreamOutput;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+
+import java.io.IOException;
+
+/**
+ */
+public class UnavailableSnappyCompressor extends SnappyCompressor {
+
+    @Override
+    public String type() {
+        return "snappy";
+    }
+
+    @Override
+    protected int maxCompressedLength(int length) {
+        return length;
+    }
+
+    @Override
+    public byte[] uncompress(byte[] data, int offset, int length) throws IOException {
+        throw new ElasticSearchIllegalStateException("snappy unavailable");
+    }
+
+    @Override
+    public byte[] compress(byte[] data, int offset, int length) throws IOException {
+        throw new ElasticSearchIllegalStateException("snappy unavailable");
+    }
+
+    @Override
+    public CompressedStreamInput streamInput(StreamInput in) throws IOException {
+        throw new ElasticSearchIllegalStateException("snappy unavailable");
+    }
+
+    @Override
+    public CompressedStreamOutput streamOutput(StreamOutput out) throws IOException {
+        throw new ElasticSearchIllegalStateException("snappy unavailable");
+    }
+
+    @Override
+    public CompressedIndexInput indexInput(IndexInput in) throws IOException {
+        throw new ElasticSearchIllegalStateException("snappy unavailable");
+    }
+
+    @Override
+    public CompressedIndexOutput indexOutput(IndexOutput out) throws IOException {
+        throw new ElasticSearchIllegalStateException("snappy unavailable");
+    }
+}
diff --git a/src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappy.java b/src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappy.java
new file mode 100644
index 00000000000..64d3bf8f17f
--- /dev/null
+++ b/src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappy.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.common.compress.snappy.xerial;
+
+import org.xerial.snappy.Snappy;
+
+/**
+ */
+public class XerialSnappy {
+
+    public static final boolean available;
+    public static final Throwable failure;
+
+    static {
+        Throwable failureX = null;
+        boolean availableX;
+        try {
+            byte[] tests = Snappy.compress("test");
+            Snappy.uncompressString(tests);
+            availableX = true;
+        } catch (Throwable e) {
+            availableX = false;
+            failureX = e;
+        }
+        available = availableX;
+        failure = failureX;
+    }
+}
diff --git a/src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappyCompressedIndexInput.java b/src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappyCompressedIndexInput.java
new file mode 100644
index 00000000000..0ff387145ca
--- /dev/null
+++ b/src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappyCompressedIndexInput.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.common.compress.snappy.xerial;
+
+import org.apache.lucene.store.IndexInput;
+import org.elasticsearch.common.compress.snappy.SnappyCompressedIndexInput;
+import org.elasticsearch.common.compress.snappy.SnappyCompressorContext;
+import org.xerial.snappy.Snappy;
+
+import java.io.IOException;
+
+/**
+ */
+public class XerialSnappyCompressedIndexInput extends SnappyCompressedIndexInput {
+
+    public XerialSnappyCompressedIndexInput(IndexInput in, SnappyCompressorContext context) throws IOException {
+        super(in, context);
+    }
+
+    @Override
+    protected int uncompress(IndexInput in, byte[] out) throws IOException {
+        boolean compressed = in.readByte() != 0;
+        int length = in.readVInt();
+        if (!compressed) {
+            in.readBytes(out, 0, length);
+            return length;
+        } else {
+            in.readBytes(inputBuffer, 0, length);
+            return Snappy.rawUncompress(inputBuffer, 0, length, out, 0);
+        }
+    }
+}
diff --git a/src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappyCompressedIndexOutput.java b/src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappyCompressedIndexOutput.java
new file mode 100644
index 00000000000..69dc115afdc
--- /dev/null
+++ b/src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappyCompressedIndexOutput.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.common.compress.snappy.xerial;
+
+import org.apache.lucene.store.IndexOutput;
+import org.elasticsearch.common.compress.snappy.SnappyCompressedIndexOutput;
+import org.elasticsearch.common.compress.snappy.SnappyCompressorContext;
+import org.xerial.snappy.Snappy;
+
+import java.io.IOException;
+
+/**
+ */
+public class XerialSnappyCompressedIndexOutput extends SnappyCompressedIndexOutput {
+
+    public XerialSnappyCompressedIndexOutput(IndexOutput out, SnappyCompressorContext context) throws IOException {
+        super(out, context);
+    }
+
+    @Override
+    protected void compress(byte[] data, int offset, int len, IndexOutput out) throws IOException {
+        int compressedLength = Snappy.rawCompress(data, offset, len, compressedBuffer, 0);
+        // use uncompressed input if less than 12.5% compression
+        if (compressedLength >= (len - (len / 8))) {
+            out.writeByte((byte) 0);
+            out.writeVInt(len);
+            out.writeBytes(data, offset, len);
+        } else {
+            out.writeByte((byte) 1);
+            out.writeVInt(compressedLength);
+            out.writeBytes(compressedBuffer, 0, compressedLength);
+        }
+    }
+}
diff --git a/src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappyCompressedStreamInput.java b/src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappyCompressedStreamInput.java
new file mode 100644
index 00000000000..079e02fae42
--- /dev/null
+++ b/src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappyCompressedStreamInput.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.common.compress.snappy.xerial;
+
+import org.elasticsearch.common.compress.snappy.SnappyCompressedStreamInput;
+import org.elasticsearch.common.compress.snappy.SnappyCompressorContext;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.xerial.snappy.Snappy;
+
+import java.io.IOException;
+
+/**
+ */
+public class XerialSnappyCompressedStreamInput extends SnappyCompressedStreamInput {
+
+    public XerialSnappyCompressedStreamInput(StreamInput in, SnappyCompressorContext context) throws IOException {
+        super(in, context);
+    }
+
+    @Override
+    protected int uncompress(StreamInput in, byte[] out) throws IOException {
+        int compressedByte = in.read();
+        // if we are on the "tip", just return 0 uncompressed data...
+        if (compressedByte == -1) {
+            return 0;
+        }
+        boolean compressed = compressedByte == 1;
+        int length = in.readVInt();
+        if (!compressed) {
+            in.readBytes(out, 0, length);
+            return length;
+        } else {
+            in.readBytes(inputBuffer, 0, length);
+            return Snappy.rawUncompress(inputBuffer, 0, length, out, 0);
+        }
+    }
+}
diff --git a/src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappyCompressedStreamOutput.java b/src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappyCompressedStreamOutput.java
new file mode 100644
index 00000000000..bc3e71a05c2
--- /dev/null
+++ b/src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappyCompressedStreamOutput.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.common.compress.snappy.xerial;
+
+import org.elasticsearch.common.compress.snappy.SnappyCompressedStreamOutput;
+import org.elasticsearch.common.compress.snappy.SnappyCompressorContext;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.xerial.snappy.Snappy;
+
+import java.io.IOException;
+
+/**
+ */
+public class XerialSnappyCompressedStreamOutput extends SnappyCompressedStreamOutput {
+
+    public XerialSnappyCompressedStreamOutput(StreamOutput out, SnappyCompressorContext context) throws IOException {
+        super(out, context);
+    }
+
+    @Override
+    protected void compress(byte[] data, int offset, int len, StreamOutput out) throws IOException {
+        int compressedLength = Snappy.rawCompress(data, offset, len, compressedBuffer, 0);
+        // use uncompressed input if less than 12.5% compression
+        if (compressedLength >= (len - (len / 8))) {
+            out.writeBoolean(false);
+            out.writeVInt(len);
+            out.writeBytes(data, offset, len);
+        } else {
+            out.writeBoolean(true);
+            out.writeVInt(compressedLength);
+            out.writeBytes(compressedBuffer, 0, compressedLength);
+        }
+    }
+}
diff --git a/src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappyCompressor.java b/src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappyCompressor.java
new file mode 100644
index 00000000000..6890376abea
--- /dev/null
+++ b/src/main/java/org/elasticsearch/common/compress/snappy/xerial/XerialSnappyCompressor.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.common.compress.snappy.xerial;
+
+import org.apache.lucene.store.IndexInput;
+import org.apache.lucene.store.IndexOutput;
+import org.elasticsearch.common.compress.CompressedIndexInput;
+import org.elasticsearch.common.compress.CompressedIndexOutput;
+import org.elasticsearch.common.compress.CompressedStreamInput;
+import org.elasticsearch.common.compress.CompressedStreamOutput;
+import org.elasticsearch.common.compress.snappy.SnappyCompressor;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.xerial.snappy.Snappy;
+
+import java.io.IOException;
+
+/**
+ */
+public class XerialSnappyCompressor extends SnappyCompressor {
+
+    @Override
+    public String type() {
+        return "snappy";
+    }
+
+    @Override
+    protected int maxCompressedLength(int length) {
+        return Snappy.maxCompressedLength(length);
+    }
+
+    @Override
+    public CompressedStreamInput streamInput(StreamInput in) throws IOException {
+        return new XerialSnappyCompressedStreamInput(in, compressorContext);
+    }
+
+    @Override
+    public CompressedStreamOutput streamOutput(StreamOutput out) throws IOException {
+        return new XerialSnappyCompressedStreamOutput(out, compressorContext);
+    }
+
+    @Override
+    public CompressedIndexInput indexInput(IndexInput in) throws IOException {
+        return new XerialSnappyCompressedIndexInput(in, compressorContext);
+    }
+
+    @Override
+    public CompressedIndexOutput indexOutput(IndexOutput out) throws IOException {
+        return new XerialSnappyCompressedIndexOutput(out, compressorContext);
+    }
+}
diff --git a/src/main/java/org/elasticsearch/common/io/Streams.java b/src/main/java/org/elasticsearch/common/io/Streams.java
index ef142068154..8e4e9324b28 100644
--- a/src/main/java/org/elasticsearch/common/io/Streams.java
+++ b/src/main/java/org/elasticsearch/common/io/Streams.java
@@ -32,8 +32,6 @@ import java.io.*;
  * <p/>
  * <p>Mainly for use within the framework,
  * but also useful for application code.
- *
- *
  */
 public abstract class Streams {
 
@@ -164,7 +162,7 @@ public abstract class Streams {
     public static byte[] copyToByteArray(InputStream in) throws IOException {
         CachedStreamOutput.Entry cachedEntry = CachedStreamOutput.popEntry();
         try {
-            BytesStreamOutput out = cachedEntry.cachedBytes();
+            BytesStreamOutput out = cachedEntry.bytes();
             copy(in, out);
             return out.copiedByteArray();
         } finally {
diff --git a/src/main/java/org/elasticsearch/common/io/stream/AdapterStreamOutput.java b/src/main/java/org/elasticsearch/common/io/stream/AdapterStreamOutput.java
index b5d4940d5b1..2f74b07c746 100644
--- a/src/main/java/org/elasticsearch/common/io/stream/AdapterStreamOutput.java
+++ b/src/main/java/org/elasticsearch/common/io/stream/AdapterStreamOutput.java
@@ -34,7 +34,7 @@ public class AdapterStreamOutput extends StreamOutput {
         this.out = out;
     }
 
-    public void reset(StreamOutput out) throws IOException {
+    public void setOut(StreamOutput out) {
         this.out = out;
     }
 
diff --git a/src/main/java/org/elasticsearch/common/io/stream/CachedStreamOutput.java b/src/main/java/org/elasticsearch/common/io/stream/CachedStreamOutput.java
index 3436e2a9a48..3126f8519b0 100644
--- a/src/main/java/org/elasticsearch/common/io/stream/CachedStreamOutput.java
+++ b/src/main/java/org/elasticsearch/common/io/stream/CachedStreamOutput.java
@@ -47,35 +47,28 @@ public class CachedStreamOutput {
             this.handles = handles;
         }
 
-        /**
-         * Returns the underlying bytes without any resetting.
-         */
+        public void reset() {
+            bytes.reset();
+            handles.setOut(bytes);
+            handles.clear();
+        }
+
         public BytesStreamOutput bytes() {
             return bytes;
         }
 
-        /**
-         * Returns cached bytes that are also reset.
-         */
-        public BytesStreamOutput cachedBytes() {
-            bytes.reset();
-            return bytes;
-        }
-
-        public StreamOutput cachedHandles() throws IOException {
-            handles.reset(bytes);
+        public StreamOutput handles() throws IOException {
             return handles;
         }
 
-        public StreamOutput cachedBytes(Compressor compressor) throws IOException {
-            bytes.reset();
+        public StreamOutput bytes(Compressor compressor) throws IOException {
             return compressor.streamOutput(bytes);
         }
 
-        public StreamOutput cachedHandles(Compressor compressor) throws IOException {
-            bytes.reset();
+        public StreamOutput handles(Compressor compressor) throws IOException {
             StreamOutput compressed = compressor.streamOutput(bytes);
-            handles.reset(compressed);
+            handles.clear();
+            handles.setOut(compressed);
             return handles;
         }
     }
@@ -118,10 +111,12 @@ public class CachedStreamOutput {
             return newEntry();
         }
         counter.decrementAndGet();
+        entry.reset();
         return entry;
     }
 
     public static void pushEntry(Entry entry) {
+        entry.reset();
         if (entry.bytes().underlyingBytes().length > BYTES_LIMIT) {
             return;
         }
diff --git a/src/main/java/org/elasticsearch/common/io/stream/HandlesStreamOutput.java b/src/main/java/org/elasticsearch/common/io/stream/HandlesStreamOutput.java
index 2ad85045109..c99817b9c08 100644
--- a/src/main/java/org/elasticsearch/common/io/stream/HandlesStreamOutput.java
+++ b/src/main/java/org/elasticsearch/common/io/stream/HandlesStreamOutput.java
@@ -80,12 +80,14 @@ public class HandlesStreamOutput extends AdapterStreamOutput {
     public void reset() throws IOException {
         handles.clear();
         identityHandles.clear();
-        out.reset();
+        if (out != null) {
+            out.reset();
+        }
     }
 
-    public void reset(StreamOutput out) throws IOException {
-        super.reset(out);
-        reset();
+    public void clear() {
+        handles.clear();
+        identityHandles.clear();
     }
 
     /**
diff --git a/src/main/java/org/elasticsearch/discovery/zen/ping/multicast/MulticastZenPing.java b/src/main/java/org/elasticsearch/discovery/zen/ping/multicast/MulticastZenPing.java
index bbd83e06e70..ffd30a9136b 100644
--- a/src/main/java/org/elasticsearch/discovery/zen/ping/multicast/MulticastZenPing.java
+++ b/src/main/java/org/elasticsearch/discovery/zen/ping/multicast/MulticastZenPing.java
@@ -268,7 +268,7 @@ public class MulticastZenPing extends AbstractLifecycleComponent<ZenPing> implem
         synchronized (sendMutex) {
             CachedStreamOutput.Entry cachedEntry = CachedStreamOutput.popEntry();
             try {
-                StreamOutput out = cachedEntry.cachedHandles();
+                StreamOutput out = cachedEntry.handles();
                 out.writeBytes(INTERNAL_HEADER);
                 Version.writeVersion(Version.CURRENT, out);
                 out.writeInt(id);
diff --git a/src/main/java/org/elasticsearch/discovery/zen/publish/PublishClusterStateAction.java b/src/main/java/org/elasticsearch/discovery/zen/publish/PublishClusterStateAction.java
index 9c817b52b50..c874867acdf 100644
--- a/src/main/java/org/elasticsearch/discovery/zen/publish/PublishClusterStateAction.java
+++ b/src/main/java/org/elasticsearch/discovery/zen/publish/PublishClusterStateAction.java
@@ -69,7 +69,7 @@ public class PublishClusterStateAction extends AbstractComponent {
         CachedStreamOutput.Entry cachedEntry = CachedStreamOutput.popEntry();
         byte[] clusterStateInBytes;
         try {
-            StreamOutput stream = cachedEntry.cachedHandles(CompressorFactory.defaultCompressor());
+            StreamOutput stream = cachedEntry.handles(CompressorFactory.defaultCompressor());
             ClusterState.Builder.writeTo(clusterState, stream);
             stream.close();
             clusterStateInBytes = cachedEntry.bytes().copiedByteArray();
diff --git a/src/main/java/org/elasticsearch/gateway/blobstore/BlobStoreGateway.java b/src/main/java/org/elasticsearch/gateway/blobstore/BlobStoreGateway.java
index d72a9d7a7a7..7eedb9f8ce4 100644
--- a/src/main/java/org/elasticsearch/gateway/blobstore/BlobStoreGateway.java
+++ b/src/main/java/org/elasticsearch/gateway/blobstore/BlobStoreGateway.java
@@ -155,9 +155,9 @@ public abstract class BlobStoreGateway extends SharedStorageGateway {
         try {
             StreamOutput streamOutput;
             if (compress) {
-                streamOutput = cachedEntry.cachedBytes(CompressorFactory.defaultCompressor());
+                streamOutput = cachedEntry.bytes(CompressorFactory.defaultCompressor());
             } else {
-                streamOutput = cachedEntry.cachedBytes();
+                streamOutput = cachedEntry.bytes();
             }
             XContentBuilder builder = XContentFactory.contentBuilder(XContentType.JSON, streamOutput);
             builder.startObject();
diff --git a/src/main/java/org/elasticsearch/gateway/local/state/meta/LocalGatewayMetaState.java b/src/main/java/org/elasticsearch/gateway/local/state/meta/LocalGatewayMetaState.java
index 203e4bba2eb..9b1802183fe 100644
--- a/src/main/java/org/elasticsearch/gateway/local/state/meta/LocalGatewayMetaState.java
+++ b/src/main/java/org/elasticsearch/gateway/local/state/meta/LocalGatewayMetaState.java
@@ -322,7 +322,7 @@ public class LocalGatewayMetaState extends AbstractComponent implements ClusterS
         logger.trace("[{}] writing state, reason [{}]", indexMetaData.index(), reason);
         CachedStreamOutput.Entry cachedEntry = CachedStreamOutput.popEntry();
         try {
-            XContentBuilder builder = XContentFactory.contentBuilder(format, cachedEntry.cachedBytes());
+            XContentBuilder builder = XContentFactory.contentBuilder(format, cachedEntry.bytes());
             builder.startObject();
             IndexMetaData.Builder.toXContent(indexMetaData, builder, formatParams);
             builder.endObject();
@@ -373,7 +373,7 @@ public class LocalGatewayMetaState extends AbstractComponent implements ClusterS
 
         CachedStreamOutput.Entry cachedEntry = CachedStreamOutput.popEntry();
         try {
-            XContentBuilder builder = XContentFactory.contentBuilder(format, cachedEntry.cachedBytes());
+            XContentBuilder builder = XContentFactory.contentBuilder(format, cachedEntry.bytes());
             builder.startObject();
             MetaData.Builder.toXContent(globalMetaData, builder, formatParams);
             builder.endObject();
diff --git a/src/main/java/org/elasticsearch/gateway/local/state/shards/LocalGatewayShardsState.java b/src/main/java/org/elasticsearch/gateway/local/state/shards/LocalGatewayShardsState.java
index 066cb7acd4e..e7a192d92db 100644
--- a/src/main/java/org/elasticsearch/gateway/local/state/shards/LocalGatewayShardsState.java
+++ b/src/main/java/org/elasticsearch/gateway/local/state/shards/LocalGatewayShardsState.java
@@ -268,7 +268,7 @@ public class LocalGatewayShardsState extends AbstractComponent implements Cluste
         logger.trace("[{}][{}] writing shard state, reason [{}]", shardId.index().name(), shardId.id(), reason);
         CachedStreamOutput.Entry cachedEntry = CachedStreamOutput.popEntry();
         try {
-            XContentBuilder builder = XContentFactory.contentBuilder(XContentType.JSON, cachedEntry.cachedBytes());
+            XContentBuilder builder = XContentFactory.contentBuilder(XContentType.JSON, cachedEntry.bytes());
             builder.prettyPrint();
             builder.startObject();
             builder.field("version", shardStateInfo.version);
diff --git a/src/main/java/org/elasticsearch/index/mapper/core/BinaryFieldMapper.java b/src/main/java/org/elasticsearch/index/mapper/core/BinaryFieldMapper.java
index b0531363da4..f89bc60255c 100644
--- a/src/main/java/org/elasticsearch/index/mapper/core/BinaryFieldMapper.java
+++ b/src/main/java/org/elasticsearch/index/mapper/core/BinaryFieldMapper.java
@@ -170,7 +170,7 @@ public class BinaryFieldMapper extends AbstractFieldMapper<byte[]> {
             if (compress != null && compress && !CompressorFactory.isCompressed(value, 0, value.length)) {
                 if (compressThreshold == -1 || value.length > compressThreshold) {
                     CachedStreamOutput.Entry cachedEntry = CachedStreamOutput.popEntry();
-                    StreamOutput streamOutput = cachedEntry.cachedBytes(CompressorFactory.defaultCompressor());
+                    StreamOutput streamOutput = cachedEntry.bytes(CompressorFactory.defaultCompressor());
                     streamOutput.writeBytes(value, 0, value.length);
                     streamOutput.close();
                     // we copy over the byte array, since we need to push back the cached entry
diff --git a/src/main/java/org/elasticsearch/index/mapper/internal/SourceFieldMapper.java b/src/main/java/org/elasticsearch/index/mapper/internal/SourceFieldMapper.java
index dbe76d60895..0d5f43cf6c5 100644
--- a/src/main/java/org/elasticsearch/index/mapper/internal/SourceFieldMapper.java
+++ b/src/main/java/org/elasticsearch/index/mapper/internal/SourceFieldMapper.java
@@ -255,9 +255,9 @@ public class SourceFieldMapper extends AbstractFieldMapper<byte[]> implements In
             CachedStreamOutput.Entry cachedEntry = CachedStreamOutput.popEntry();
             StreamOutput streamOutput;
             if (compress != null && compress && (compressThreshold == -1 || dataLength > compressThreshold)) {
-                streamOutput = cachedEntry.cachedBytes(CompressorFactory.defaultCompressor());
+                streamOutput = cachedEntry.bytes(CompressorFactory.defaultCompressor());
             } else {
-                streamOutput = cachedEntry.cachedBytes();
+                streamOutput = cachedEntry.bytes();
             }
             XContentType contentType = formatContentType;
             if (contentType == null) {
@@ -277,11 +277,11 @@ public class SourceFieldMapper extends AbstractFieldMapper<byte[]> implements In
                 try {
                     XContentType contentType = XContentFactory.xContentType(data, dataOffset, dataLength);
                     if (formatContentType != null && formatContentType != contentType) {
-                        XContentBuilder builder = XContentFactory.contentBuilder(formatContentType, cachedEntry.cachedBytes(CompressorFactory.defaultCompressor()));
+                        XContentBuilder builder = XContentFactory.contentBuilder(formatContentType, cachedEntry.bytes(CompressorFactory.defaultCompressor()));
                         builder.copyCurrentStructure(XContentFactory.xContent(contentType).createParser(data, dataOffset, dataLength));
                         builder.close();
                     } else {
-                        StreamOutput streamOutput = cachedEntry.cachedBytes(CompressorFactory.defaultCompressor());
+                        StreamOutput streamOutput = cachedEntry.bytes(CompressorFactory.defaultCompressor());
                         streamOutput.writeBytes(data, dataOffset, dataLength);
                         streamOutput.close();
                     }
@@ -307,7 +307,7 @@ public class SourceFieldMapper extends AbstractFieldMapper<byte[]> implements In
                     // we need to reread and store back, compressed....
                     CachedStreamOutput.Entry cachedEntry = CachedStreamOutput.popEntry();
                     try {
-                        StreamOutput streamOutput = cachedEntry.cachedBytes(CompressorFactory.defaultCompressor());
+                        StreamOutput streamOutput = cachedEntry.bytes(CompressorFactory.defaultCompressor());
                         XContentBuilder builder = XContentFactory.contentBuilder(formatContentType, streamOutput);
                         builder.copyCurrentStructure(XContentFactory.xContent(contentType).createParser(compressedStreamInput));
                         builder.close();
@@ -329,7 +329,7 @@ public class SourceFieldMapper extends AbstractFieldMapper<byte[]> implements In
                     // we need to reread and store back, compressed....
                     CachedStreamOutput.Entry cachedEntry = CachedStreamOutput.popEntry();
                     try {
-                        XContentBuilder builder = XContentFactory.contentBuilder(formatContentType, cachedEntry.cachedBytes());
+                        XContentBuilder builder = XContentFactory.contentBuilder(formatContentType, cachedEntry.bytes());
                         builder.copyCurrentStructure(XContentFactory.xContent(contentType).createParser(data, dataOffset, dataLength));
                         builder.close();
                         data = cachedEntry.bytes().copiedByteArray();
diff --git a/src/main/java/org/elasticsearch/index/translog/fs/FsTranslog.java b/src/main/java/org/elasticsearch/index/translog/fs/FsTranslog.java
index dab85836481..0b83cc69a9e 100644
--- a/src/main/java/org/elasticsearch/index/translog/fs/FsTranslog.java
+++ b/src/main/java/org/elasticsearch/index/translog/fs/FsTranslog.java
@@ -323,7 +323,7 @@ public class FsTranslog extends AbstractIndexShardComponent implements Translog
         CachedStreamOutput.Entry cachedEntry = CachedStreamOutput.popEntry();
         rwl.readLock().lock();
         try {
-            BytesStreamOutput out = cachedEntry.cachedBytes();
+            BytesStreamOutput out = cachedEntry.bytes();
             out.writeInt(0); // marker for the size...
             TranslogStreams.writeTranslogOperation(out, operation);
             out.flush();
diff --git a/src/main/java/org/elasticsearch/node/internal/InternalNode.java b/src/main/java/org/elasticsearch/node/internal/InternalNode.java
index 7bc7aae745d..0ae610e8eca 100644
--- a/src/main/java/org/elasticsearch/node/internal/InternalNode.java
+++ b/src/main/java/org/elasticsearch/node/internal/InternalNode.java
@@ -35,6 +35,7 @@ import org.elasticsearch.common.StopWatch;
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.component.Lifecycle;
 import org.elasticsearch.common.component.LifecycleComponent;
+import org.elasticsearch.common.compress.CompressorFactory;
 import org.elasticsearch.common.inject.Injector;
 import org.elasticsearch.common.inject.Injectors;
 import org.elasticsearch.common.inject.ModulesBuilder;
@@ -117,6 +118,8 @@ public final class InternalNode implements Node {
         this.settings = pluginsService.updatedSettings();
         this.environment = tuple.v2();
 
+        CompressorFactory.configure(settings);
+
         NodeEnvironment nodeEnvironment = new NodeEnvironment(this.settings, this.environment);
 
         ModulesBuilder modules = new ModulesBuilder();
diff --git a/src/main/java/org/elasticsearch/rest/action/support/RestXContentBuilder.java b/src/main/java/org/elasticsearch/rest/action/support/RestXContentBuilder.java
index 330150beb35..ce01bb759f3 100644
--- a/src/main/java/org/elasticsearch/rest/action/support/RestXContentBuilder.java
+++ b/src/main/java/org/elasticsearch/rest/action/support/RestXContentBuilder.java
@@ -47,7 +47,7 @@ public class RestXContentBuilder {
             contentType = XContentType.JSON;
         }
         CachedStreamOutput.Entry cachedEntry = CachedStreamOutput.popEntry();
-        XContentBuilder builder = new XContentBuilder(XContentFactory.xContent(contentType), cachedEntry.cachedBytes(), cachedEntry);
+        XContentBuilder builder = new XContentBuilder(XContentFactory.xContent(contentType), cachedEntry.bytes(), cachedEntry);
         if (request.paramAsBoolean("pretty", false)) {
             builder.prettyPrint();
         }
diff --git a/src/main/java/org/elasticsearch/transport/local/LocalTransport.java b/src/main/java/org/elasticsearch/transport/local/LocalTransport.java
index e22b6bcd342..b959dd3f819 100644
--- a/src/main/java/org/elasticsearch/transport/local/LocalTransport.java
+++ b/src/main/java/org/elasticsearch/transport/local/LocalTransport.java
@@ -159,7 +159,7 @@ public class LocalTransport extends AbstractLifecycleComponent<Transport> implem
     public <T extends Streamable> void sendRequest(final DiscoveryNode node, final long requestId, final String action, final Streamable message, TransportRequestOptions options) throws IOException, TransportException {
         CachedStreamOutput.Entry cachedEntry = CachedStreamOutput.popEntry();
         try {
-            StreamOutput stream = cachedEntry.cachedHandles();
+            StreamOutput stream = cachedEntry.handles();
 
             stream.writeLong(requestId);
             byte status = 0;
diff --git a/src/main/java/org/elasticsearch/transport/local/LocalTransportChannel.java b/src/main/java/org/elasticsearch/transport/local/LocalTransportChannel.java
index 5b686b099ff..ca2891479ad 100644
--- a/src/main/java/org/elasticsearch/transport/local/LocalTransportChannel.java
+++ b/src/main/java/org/elasticsearch/transport/local/LocalTransportChannel.java
@@ -68,7 +68,7 @@ public class LocalTransportChannel implements TransportChannel {
     public void sendResponse(Streamable message, TransportResponseOptions options) throws IOException {
         CachedStreamOutput.Entry cachedEntry = CachedStreamOutput.popEntry();
         try {
-            StreamOutput stream = cachedEntry.cachedHandles();
+            StreamOutput stream = cachedEntry.handles();
             stream.writeLong(requestId);
             byte status = 0;
             status = TransportStreams.statusSetResponse(status);
@@ -93,14 +93,15 @@ public class LocalTransportChannel implements TransportChannel {
         try {
             BytesStreamOutput stream;
             try {
-                stream = cachedEntry.cachedBytes();
+                stream = cachedEntry.bytes();
                 writeResponseExceptionHeader(stream);
                 RemoteTransportException tx = new RemoteTransportException(targetTransport.nodeName(), targetTransport.boundAddress().boundAddress(), action, error);
                 ThrowableObjectOutputStream too = new ThrowableObjectOutputStream(stream);
                 too.writeObject(tx);
                 too.close();
             } catch (NotSerializableException e) {
-                stream = cachedEntry.cachedBytes();
+                cachedEntry.reset();
+                stream = cachedEntry.bytes();
                 writeResponseExceptionHeader(stream);
                 RemoteTransportException tx = new RemoteTransportException(targetTransport.nodeName(), targetTransport.boundAddress().boundAddress(), action, new NotSerializableTransportException(error));
                 ThrowableObjectOutputStream too = new ThrowableObjectOutputStream(stream);
diff --git a/src/main/java/org/elasticsearch/transport/netty/NettyTransportChannel.java b/src/main/java/org/elasticsearch/transport/netty/NettyTransportChannel.java
index 8bcb5b8ec6c..3799ec26848 100644
--- a/src/main/java/org/elasticsearch/transport/netty/NettyTransportChannel.java
+++ b/src/main/java/org/elasticsearch/transport/netty/NettyTransportChannel.java
@@ -85,14 +85,15 @@ public class NettyTransportChannel implements TransportChannel {
         CachedStreamOutput.Entry cachedEntry = CachedStreamOutput.popEntry();
         BytesStreamOutput stream;
         try {
-            stream = cachedEntry.cachedBytes();
+            stream = cachedEntry.bytes();
             writeResponseExceptionHeader(stream);
             RemoteTransportException tx = new RemoteTransportException(transport.nodeName(), transport.wrapAddress(channel.getLocalAddress()), action, error);
             ThrowableObjectOutputStream too = new ThrowableObjectOutputStream(stream);
             too.writeObject(tx);
             too.close();
         } catch (NotSerializableException e) {
-            stream = cachedEntry.cachedBytes();
+            cachedEntry.reset();
+            stream = cachedEntry.bytes();
             writeResponseExceptionHeader(stream);
             RemoteTransportException tx = new RemoteTransportException(transport.nodeName(), transport.wrapAddress(channel.getLocalAddress()), action, new NotSerializableTransportException(error));
             ThrowableObjectOutputStream too = new ThrowableObjectOutputStream(stream);
diff --git a/src/main/java/org/elasticsearch/transport/support/TransportStreams.java b/src/main/java/org/elasticsearch/transport/support/TransportStreams.java
index e3396420da0..bcff6c71759 100644
--- a/src/main/java/org/elasticsearch/transport/support/TransportStreams.java
+++ b/src/main/java/org/elasticsearch/transport/support/TransportStreams.java
@@ -106,13 +106,13 @@ public class TransportStreams {
 
         if (options.compress()) {
             status = TransportStreams.statusSetCompress(status);
-            StreamOutput stream = cachedEntry.cachedHandles(CompressorFactory.defaultCompressor());
             cachedEntry.bytes().write(HEADER_PLACEHOLDER);
+            StreamOutput stream = cachedEntry.handles(CompressorFactory.defaultCompressor());
             stream.writeUTF(action);
             message.writeTo(stream);
             stream.close();
         } else {
-            StreamOutput stream = cachedEntry.cachedHandles();
+            StreamOutput stream = cachedEntry.handles();
             cachedEntry.bytes().write(HEADER_PLACEHOLDER);
             stream.writeUTF(action);
             message.writeTo(stream);
@@ -127,12 +127,12 @@ public class TransportStreams {
 
         if (options.compress()) {
             status = TransportStreams.statusSetCompress(status);
-            StreamOutput stream = cachedEntry.cachedHandles(CompressorFactory.defaultCompressor());
             cachedEntry.bytes().write(HEADER_PLACEHOLDER);
+            StreamOutput stream = cachedEntry.handles(CompressorFactory.defaultCompressor());
             message.writeTo(stream);
             stream.close();
         } else {
-            StreamOutput stream = cachedEntry.cachedHandles();
+            StreamOutput stream = cachedEntry.handles();
             cachedEntry.bytes().write(HEADER_PLACEHOLDER);
             message.writeTo(stream);
             stream.close();
diff --git a/src/test/java/org/elasticsearch/benchmark/compress/LuceneCompressionBenchmark.java b/src/test/java/org/elasticsearch/benchmark/compress/LuceneCompressionBenchmark.java
index c2352140657..ee6d88b0c96 100644
--- a/src/test/java/org/elasticsearch/benchmark/compress/LuceneCompressionBenchmark.java
+++ b/src/test/java/org/elasticsearch/benchmark/compress/LuceneCompressionBenchmark.java
@@ -27,8 +27,8 @@ import org.apache.lucene.store.Directory;
 import org.apache.lucene.store.FSDirectory;
 import org.apache.lucene.store.NIOFSDirectory;
 import org.elasticsearch.common.compress.CompressedDirectory;
-import org.elasticsearch.common.compress.Compressor;
-import org.elasticsearch.common.compress.CompressorFactory;
+import org.elasticsearch.common.compress.lzf.LZFCompressor;
+import org.elasticsearch.common.compress.snappy.xerial.XerialSnappyCompressor;
 import org.elasticsearch.common.io.FileSystemUtils;
 import org.elasticsearch.common.lucene.Lucene;
 import org.elasticsearch.common.unit.ByteSizeValue;
@@ -45,8 +45,6 @@ public class LuceneCompressionBenchmark {
         final long MAX_SIZE = ByteSizeValue.parseBytesSizeValue("50mb").bytes();
         final boolean WITH_TV = true;
 
-        final Compressor compressor = CompressorFactory.defaultCompressor();
-
         File testFile = new File("target/test/compress/lucene");
         FileSystemUtils.deleteRecursively(testFile);
         testFile.mkdirs();
@@ -54,9 +52,11 @@ public class LuceneCompressionBenchmark {
         FSDirectory uncompressedDir = new NIOFSDirectory(new File(testFile, "uncompressed"));
         IndexWriter uncompressedWriter = new IndexWriter(uncompressedDir, new IndexWriterConfig(Lucene.VERSION, Lucene.STANDARD_ANALYZER));
 
-        Directory compressedDir = new CompressedDirectory(new NIOFSDirectory(new File(testFile, "compressed")), compressor, false, "fdt", "tvf");
+        Directory compressedLzfDir = new CompressedDirectory(new NIOFSDirectory(new File(testFile, "compressed_lzf")), new LZFCompressor(), false, "fdt", "tvf");
+        IndexWriter compressedLzfWriter = new IndexWriter(compressedLzfDir, new IndexWriterConfig(Lucene.VERSION, Lucene.STANDARD_ANALYZER));
 
-        IndexWriter compressedWriter = new IndexWriter(compressedDir, new IndexWriterConfig(Lucene.VERSION, Lucene.STANDARD_ANALYZER));
+        Directory compressedSnappyDir = new CompressedDirectory(new NIOFSDirectory(new File(testFile, "compressed_snappy")), new XerialSnappyCompressor(), false, "fdt", "tvf");
+        IndexWriter compressedSnappyWriter = new IndexWriter(compressedSnappyDir, new IndexWriterConfig(Lucene.VERSION, Lucene.STANDARD_ANALYZER));
 
         System.out.println("feeding data...");
         TestData testData = new TestData();
@@ -72,19 +72,24 @@ public class LuceneCompressionBenchmark {
                 doc.add(field);
             }
             uncompressedWriter.addDocument(doc);
-            compressedWriter.addDocument(doc);
+            compressedLzfWriter.addDocument(doc);
+            compressedSnappyWriter.addDocument(doc);
         }
         System.out.println("optimizing...");
         uncompressedWriter.forceMerge(1);
-        compressedWriter.forceMerge(1);
+        compressedLzfWriter.forceMerge(1);
+        compressedSnappyWriter.forceMerge(1);
         uncompressedWriter.waitForMerges();
-        compressedWriter.waitForMerges();
+        compressedLzfWriter.waitForMerges();
+        compressedSnappyWriter.waitForMerges();
 
         System.out.println("done");
-        uncompressedDir.close();
-        compressedWriter.close();
+        uncompressedWriter.close();
+        compressedLzfWriter.close();
+        compressedSnappyWriter.close();
 
-        compressedDir.close();
+        compressedLzfDir.close();
+        compressedSnappyDir.close();
         uncompressedDir.close();
     }
 
diff --git a/src/test/java/org/elasticsearch/test/integration/search/compress/SearchSourceCompressTests.java b/src/test/java/org/elasticsearch/test/integration/search/compress/SearchSourceCompressTests.java
index 63a14852fbd..cadeabfbdbd 100644
--- a/src/test/java/org/elasticsearch/test/integration/search/compress/SearchSourceCompressTests.java
+++ b/src/test/java/org/elasticsearch/test/integration/search/compress/SearchSourceCompressTests.java
@@ -22,6 +22,10 @@ package org.elasticsearch.test.integration.search.compress;
 import org.elasticsearch.action.get.GetResponse;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.client.Client;
+import org.elasticsearch.common.compress.CompressorFactory;
+import org.elasticsearch.common.compress.lzf.LZFCompressor;
+import org.elasticsearch.common.compress.snappy.xerial.XerialSnappy;
+import org.elasticsearch.common.compress.snappy.xerial.XerialSnappyCompressor;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.index.query.QueryBuilders;
@@ -60,18 +64,29 @@ public class SearchSourceCompressTests extends AbstractNodesTests {
     }
 
     @Test
-    public void testSourceFieldCompressed() throws IOException {
+    public void testSourceCompressionLZF() throws IOException {
+        CompressorFactory.setDefaultCompressor(new LZFCompressor());
         verifySource(true);
-    }
-
-    @Test
-    public void testSourceFieldPlainExplicit() throws IOException {
         verifySource(false);
+        verifySource(null);
     }
 
     @Test
-    public void testSourceFieldPlain() throws IOException {
-        verifySource(null);
+    public void testSourceCompressionXerialSnappy() throws IOException {
+        if (XerialSnappy.available) {
+            CompressorFactory.setDefaultCompressor(new XerialSnappyCompressor());
+            verifySource(true);
+            verifySource(false);
+            verifySource(null);
+        }
+    }
+
+    @Test
+    public void testAll() throws IOException {
+        testSourceCompressionLZF();
+        testSourceCompressionXerialSnappy();
+        testSourceCompressionLZF();
+        testSourceCompressionXerialSnappy();
     }
 
     private void verifySource(Boolean compress) throws IOException {
diff --git a/src/test/java/org/elasticsearch/test/unit/common/compress/CompressIndexInputOutputTests.java b/src/test/java/org/elasticsearch/test/unit/common/compress/CompressIndexInputOutputTests.java
index 0e58108bda4..6249994d54b 100644
--- a/src/test/java/org/elasticsearch/test/unit/common/compress/CompressIndexInputOutputTests.java
+++ b/src/test/java/org/elasticsearch/test/unit/common/compress/CompressIndexInputOutputTests.java
@@ -1,3 +1,22 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
 package org.elasticsearch.test.unit.common.compress;
 
 import jsr166y.ThreadLocalRandom;
@@ -13,10 +32,15 @@ import org.apache.lucene.store.IndexInput;
 import org.apache.lucene.store.IndexOutput;
 import org.apache.lucene.store.RAMDirectory;
 import org.elasticsearch.common.RandomStringGenerator;
-import org.elasticsearch.common.compress.*;
+import org.elasticsearch.common.compress.CompressedDirectory;
+import org.elasticsearch.common.compress.CompressedIndexInput;
+import org.elasticsearch.common.compress.CompressedIndexOutput;
+import org.elasticsearch.common.compress.Compressor;
+import org.elasticsearch.common.compress.lzf.LZFCompressor;
+import org.elasticsearch.common.compress.snappy.xerial.XerialSnappy;
+import org.elasticsearch.common.compress.snappy.xerial.XerialSnappyCompressor;
 import org.elasticsearch.common.lucene.Lucene;
 import org.elasticsearch.common.unit.SizeValue;
-import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
 
 import java.io.EOFException;
@@ -29,15 +53,40 @@ import static org.hamcrest.Matchers.equalTo;
 @Test
 public class CompressIndexInputOutputTests {
 
-    private Compressor compressor;
-
-    @BeforeClass
-    public void buildCompressor() {
-        this.compressor = CompressorFactory.defaultCompressor();
+    @Test
+    public void testXerialSnappy() throws Exception {
+        if (XerialSnappy.available) {
+            testCompressor(new XerialSnappyCompressor());
+        }
     }
 
     @Test
-    public void empty() throws Exception {
+    public void testLZF() throws Exception {
+        testCompressor(new LZFCompressor());
+    }
+
+    @Test
+    public void testSideAffects() throws Exception {
+        if (XerialSnappy.available) {
+            testCompressor(new XerialSnappyCompressor());
+        }
+        testCompressor(new LZFCompressor());
+        if (XerialSnappy.available) {
+            testCompressor(new XerialSnappyCompressor());
+        }
+        testCompressor(new LZFCompressor());
+    }
+
+    private void testCompressor(Compressor compressor) throws Exception {
+        empty(compressor);
+        simple(compressor);
+        seek1Compressed(compressor);
+        seek1UnCompressed(compressor);
+        copyBytes(compressor);
+        lucene(compressor);
+    }
+
+    private void empty(Compressor compressor) throws Exception {
         Directory dir = new RAMDirectory();
         IndexOutput out = compressor.indexOutput(dir.createOutput("test"));
         out.close();
@@ -59,8 +108,7 @@ public class CompressIndexInputOutputTests {
         }
     }
 
-    @Test
-    public void simple() throws Exception {
+    private void simple(Compressor compressor) throws Exception {
         Directory dir = new RAMDirectory();
         IndexOutput out = compressor.indexOutput(dir.createOutput("test"));
         long pos1 = out.getFilePointer();
@@ -99,17 +147,15 @@ public class CompressIndexInputOutputTests {
         in.close();
     }
 
-    @Test
-    public void seek1Compressed() throws Exception {
-        seek1(true);
+    private void seek1Compressed(Compressor compressor) throws Exception {
+        seek1(true, compressor);
     }
 
-    @Test
-    public void seek1UnCompressed() throws Exception {
-        seek1(false);
+    private void seek1UnCompressed(Compressor compressor) throws Exception {
+        seek1(false, compressor);
     }
 
-    private void seek1(boolean compressed) throws Exception {
+    private void seek1(boolean compressed, Compressor compressor) throws Exception {
         Directory dir = new RAMDirectory();
         IndexOutput out = compressed ? compressor.indexOutput(dir.createOutput("test")) : dir.createOutput("test");
         long pos1 = out.getFilePointer();
@@ -152,8 +198,7 @@ public class CompressIndexInputOutputTests {
         }
     }
 
-    @Test
-    public void copyBytes() throws Exception {
+    private void copyBytes(Compressor compressor) throws Exception {
         Directory dir = new RAMDirectory();
         IndexOutput out = compressor.indexOutput(dir.createOutput("test"));
         long pos1 = out.getFilePointer();
@@ -204,8 +249,7 @@ public class CompressIndexInputOutputTests {
         }
     }
 
-    @Test
-    public void lucene() throws Exception {
+    private void lucene(Compressor compressor) throws Exception {
         CompressedDirectory dir = new CompressedDirectory(new RAMDirectory(), compressor, false, "fdt");
         IndexWriter writer = new IndexWriter(dir, new IndexWriterConfig(Lucene.VERSION, Lucene.STANDARD_ANALYZER));
         writer.addDocument(createDoc(1, (int) SizeValue.parseSizeValue("100b").singles()));
diff --git a/src/test/java/org/elasticsearch/test/unit/common/compress/CompressedStringTests.java b/src/test/java/org/elasticsearch/test/unit/common/compress/CompressedStringTests.java
index 7cbfea4aaf0..43a96de8485 100644
--- a/src/test/java/org/elasticsearch/test/unit/common/compress/CompressedStringTests.java
+++ b/src/test/java/org/elasticsearch/test/unit/common/compress/CompressedStringTests.java
@@ -20,6 +20,8 @@
 package org.elasticsearch.test.unit.common.compress;
 
 import org.elasticsearch.common.compress.CompressedString;
+import org.elasticsearch.common.compress.CompressorFactory;
+import org.elasticsearch.common.settings.ImmutableSettings;
 import org.testng.annotations.Test;
 
 import java.io.IOException;
@@ -34,7 +36,17 @@ import static org.hamcrest.Matchers.not;
 public class CompressedStringTests {
 
     @Test
-    public void simpleTests() throws IOException {
+    public void simpleTestsSnappy() throws IOException {
+        simpleTests("snappy");
+    }
+
+    @Test
+    public void simpleTestsLZF() throws IOException {
+        simpleTests("lzf");
+    }
+
+    public void simpleTests(String compressor) throws IOException {
+        CompressorFactory.configure(ImmutableSettings.settingsBuilder().put("compress.default.type", compressor).build());
         String str = "this is a simple string";
         CompressedString cstr = new CompressedString(str);
         assertThat(cstr.string(), equalTo(str));
diff --git a/src/test/java/org/elasticsearch/test/unit/common/io/streams/BytesStreamsTests.java b/src/test/java/org/elasticsearch/test/unit/common/io/streams/BytesStreamsTests.java
index 36419eb67e8..37307f8f56b 100644
--- a/src/test/java/org/elasticsearch/test/unit/common/io/streams/BytesStreamsTests.java
+++ b/src/test/java/org/elasticsearch/test/unit/common/io/streams/BytesStreamsTests.java
@@ -36,7 +36,7 @@ public class BytesStreamsTests {
 
     @Test
     public void testSimpleStreams() throws Exception {
-        BytesStreamOutput out = CachedStreamOutput.popEntry().cachedBytes();
+        BytesStreamOutput out = CachedStreamOutput.popEntry().bytes();
         out.writeBoolean(false);
         out.writeByte((byte) 1);
         out.writeShort((short) -1);