Repository: kafka-rust/kafka-rust (language: rust)

Issue:
Allow builds without certain dependencies
It would be desirable to be able to build without dependencies to native libraries. In certain deployments it is not necessarily to link against `openssl` and/or the utilized compression libraries. We can exclude them and the corresponding functionality using features.



Discussion / hints:
this seems to be quite problematic as a lot of code paths need to be duplicated in order to conditionally hide publicly exposed types and error variants. further duplication arises around `KafkaConnection`. after all, the effort doesn't (yet) seem to be justified.

Might help situations like outlined in #90.


Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index aaabeb99b..16e175f88 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -10,19 +10,26 @@ readme = "README.md"
 keywords = ["kafka", "queue"]
 license = "MIT"
 
+
 [dependencies]
 byteorder = "0.4"
-flate2 = "0.2"
 crc = "1.2"
 log = "0.3"
-snappy = "0.3"
 ref_slice = "1.0"
 fnv = "1.0"
-openssl = "0.7"
+
+flate2 = { version = "0.2", optional = true }
+snappy = { version = "0.3", optional = true }
+openssl = { version = "0.7", optional = true }
+
 
 [dev-dependencies]
 getopts = "0.2"
 env_logger = "0.3"
 
+
 [features]
+default = ["snappy", "gzip", "security"]
+gzip = ["flate2"]
+security = ["openssl"]
 nightly = []
diff --git a/README.md b/README.md
index 190e0cdd5..f8c3ec9fd 100644
--- a/README.md
+++ b/README.md
@@ -27,6 +27,9 @@ If that library is not installed in the usual path, you can export the
 `LD_LIBRARY_PATH` and `LD_RUN_PATH` environment variables before
 issueing `cargo build`.
 
+Alternatively, support for various features, including snappy, can be
+turned off at compile time.  See kafka-rust's `Cargo.toml` and
+[cargo's documentation](http://doc.crates.io/manifest.html#the-features-section).
 
 ### Supported Kafka version
 
diff --git a/examples/console-producer.rs b/examples/console-producer.rs
index 827f265e0..66d53597d 100644
--- a/examples/console-producer.rs
+++ b/examples/console-producer.rs
@@ -228,9 +228,11 @@ impl Config {
                     .unwrap_or_else(|| "NONE".to_owned());
                 match s.trim() {
                     "none" | "NONE" => Compression::NONE,
+                    #[cfg(feature = "gzip")]
                     "gzip" | "GZIP" => Compression::GZIP,
+                    #[cfg(feature = "snappy")]
                     "snappy" | "SNAPPY" => Compression::SNAPPY,
-                    _ => return Err(Error::Literal(format!("Unknown compression type: {}", s))),
+                    _ => return Err(Error::Literal(format!("Unsupported compression type: {}", s))),
                 }
             },
             batch_size: {
diff --git a/src/client/mod.rs b/src/client/mod.rs
index 96a772c95..d48af0564 100644
--- a/src/client/mod.rs
+++ b/src/client/mod.rs
@@ -9,6 +9,7 @@ use std::io::Cursor;
 use std::iter::Iterator;
 use std::mem;
 
+#[cfg(feature = "security")]
 use openssl::ssl::SslContext;
 
 // pub re-export
@@ -86,11 +87,26 @@ struct ClientConfig {
 struct ConnectionPool {
     conns: HashMap<String, KafkaConnection>,
     timeout: i32,
+    #[cfg(feature = "security")]
     security_config: Option<SecurityConfig>
 }
 
 impl ConnectionPool {
-    fn new(timeout: i32, security: Option<SecurityConfig>) -> ConnectionPool {
+    #[cfg(not(feature = "security"))]
+    fn new(timeout: i32) -> ConnectionPool {
+        ConnectionPool {
+            conns: HashMap::new(),
+            timeout: timeout,
+        }
+    }
+
+    #[cfg(feature = "security")]
+    fn new(timeout: i32) -> ConnectionPool {
+        Self::new_with_security(timeout, None)
+    }
+
+    #[cfg(feature = "security")]
+    fn new_with_security(timeout: i32, security: Option<SecurityConfig>) -> ConnectionPool {
         ConnectionPool {
             conns: HashMap::new(),
             timeout: timeout,
@@ -106,12 +122,20 @@ impl ConnectionPool {
             // method is no longer recursive).
             return Ok(unsafe { mem::transmute(conn) });
         }
-        self.conns.insert(host.to_owned(),
-                          try!(KafkaConnection::new(
-                              host, self.timeout,
-                              self.security_config.as_ref().map(|c| &c.0))));
+        let conn = try!(self.new_conn(host));
+        self.conns.insert(host.to_owned(), conn);
         Ok(self.conns.get_mut(host).unwrap())
     }
+
+    #[cfg(not(feature = "security"))]
+    fn new_conn(&self, host: &str) -> Result<KafkaConnection> {
+        KafkaConnection::new(host, self.timeout)
+    }
+
+    #[cfg(feature = "security")]
+    fn new_conn(&self, host: &str) -> Result<KafkaConnection> {
+        KafkaConnection::new(host, self.timeout, self.security_config.as_ref().map(|c| &c.0))
+    }
 }
 
 // --------------------------------------------------------------------
@@ -279,10 +303,13 @@ impl<'a> AsRef<FetchPartition<'a>> for FetchPartition<'a> {
 
 // --------------------------------------------------------------------
 
-/// This will be expanded in the future. See #51.
+/// Security relevant configuration options for KafkaClient.
+// This will be expanded in the future. See #51.
+#[cfg(feature = "security")]
 #[derive(Debug)]
 pub struct SecurityConfig(SslContext);
 
+#[cfg(feature = "security")]
 impl SecurityConfig {
     /// In the future this will also support a kerbos via #51.
     pub fn new(ssl: SslContext) -> SecurityConfig {
@@ -314,11 +341,11 @@ impl KafkaClient {
                 fetch_max_bytes_per_partition: DEFAULT_FETCH_MAX_BYTES_PER_PARTITION,
                 fetch_crc_validation: DEFAULT_FETCH_CRC_VALIDATION,
             },
-            conn_pool: ConnectionPool::new(DEFAULT_SO_TIMEOUT_SECS, None),
+            conn_pool: ConnectionPool::new(DEFAULT_SO_TIMEOUT_SECS),
             state: state::ClientState::new(),
         }
     }
-    
+
     /// Creates a new secure instance of KafkaClient. Before being able to
     /// successfully use the new client, you'll have to load metadata.
     ///
@@ -354,6 +381,7 @@ impl KafkaClient {
     /// [openssl](https://crates.io/crates/openssl)
     /// and [openssl_verify](https://crates.io/crates/openssl-verify),
     /// as well as [Kafka's documentation](https://kafka.apache.org/documentation.html#security_ssl).
+    #[cfg(feature = "security")]
     pub fn new_secure(hosts: Vec<String>, security: SecurityConfig) -> KafkaClient {
         KafkaClient {
             config: ClientConfig {
@@ -365,7 +393,7 @@ impl KafkaClient {
                 fetch_max_bytes_per_partition: DEFAULT_FETCH_MAX_BYTES_PER_PARTITION,
                 fetch_crc_validation: DEFAULT_FETCH_CRC_VALIDATION,
             },
-            conn_pool: ConnectionPool::new(DEFAULT_SO_TIMEOUT_SECS, Some(security)),
+            conn_pool: ConnectionPool::new_with_security(DEFAULT_SO_TIMEOUT_SECS, Some(security)),
             state: state::ClientState::new(),
         }
     }
@@ -383,9 +411,11 @@ impl KafkaClient {
     /// # Example
     ///
     /// ```no_run
-    /// let mut client = kafka::client::KafkaClient::new(vec!("localhost:9092".to_owned()));
+    /// use kafka::client::{Compression, KafkaClient};
+    ///
+    /// let mut client = KafkaClient::new(vec!("localhost:9092".to_owned()));
     /// client.load_metadata_all().unwrap();
-    /// client.set_compression(kafka::client::Compression::SNAPPY);
+    /// client.set_compression(Compression::NONE);
     /// ```
     #[inline]
     pub fn set_compression(&mut self, compression: Compression) {
diff --git a/src/compression/mod.rs b/src/compression/mod.rs
index 68bb91a67..1700728bf 100644
--- a/src/compression/mod.rs
+++ b/src/compression/mod.rs
@@ -1,14 +1,19 @@
-pub mod snappy;
+#[cfg(feature = "gzip")]
 pub mod gzip;
 
+#[cfg(feature = "snappy")]
+pub mod snappy;
+
 /// Compression types supported by kafka. The numeral values of this
 /// enumeration correspond to the compression encoding in the
 /// attributes of a Message in the protocol.
 #[derive(Debug, Copy, Clone)]
 pub enum Compression {
     NONE = 0,
+    #[cfg(feature = "gzip")]
     GZIP = 1,
-    SNAPPY = 2
+    #[cfg(feature = "snappy")]
+    SNAPPY = 2,
 }
 
 impl Default for Compression {
diff --git a/src/connection.rs b/src/connection.rs
index f564d9b37..fdd7c19fc 100644
--- a/src/connection.rs
+++ b/src/connection.rs
@@ -1,53 +1,77 @@
 use std::fmt;
-use std::io::{self, Read,Write};
+use std::io::{Read, Write};
 use std::net::TcpStream;
 use std::time::Duration;
 
+#[cfg(feature = "security")]
 use openssl::ssl::{SslContext, SslStream};
 
 use error::{Error, Result};
 
-enum KafkaStream {
-    Plain(TcpStream),
-    Ssl(SslStream<TcpStream>)
-}
+#[cfg(not(feature = "security"))]
+type KafkaStream = TcpStream;
 
-impl KafkaStream {
-    fn get_ref(&self) -> &TcpStream {
-        match self {
-            &KafkaStream::Plain(ref s) => s,
-            &KafkaStream::Ssl(ref s) => s.get_ref()
-        }
+#[cfg(feature = "security")]
+use self::openssled::KafkaStream;
+
+#[cfg(feature = "security")]
+mod openssled {
+    use std::io::{self, Read, Write};
+    use std::net::TcpStream;
+    use std::time::Duration;
+
+    use openssl::ssl::SslStream;
+
+    pub enum KafkaStream {
+        Plain(TcpStream),
+        Ssl(SslStream<TcpStream>)
     }
-}
 
-impl Read for KafkaStream {
-    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        match self {
-            &mut KafkaStream::Plain(ref mut s) => s.read(buf),
-            &mut KafkaStream::Ssl(ref mut s) => s.read(buf),
+    impl KafkaStream {
+        fn get_ref(&self) -> &TcpStream {
+            match self {
+                &KafkaStream::Plain(ref s) => s,
+                &KafkaStream::Ssl(ref s) => s.get_ref()
+            }
+        }
+
+        pub fn set_read_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
+            self.get_ref().set_read_timeout(dur)
+        }
+
+        pub fn set_write_timeout(&self, dur: Option<Duration>) -> io::Result<()> {
+            self.get_ref().set_write_timeout(dur)
         }
     }
-}
 
-impl Write for KafkaStream {
-    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        match self {
-            &mut KafkaStream::Plain(ref mut s) => s.write(buf),
-            &mut KafkaStream::Ssl(ref mut s) => s.write(buf),
+    impl Read for KafkaStream {
+        fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+            match self {
+                &mut KafkaStream::Plain(ref mut s) => s.read(buf),
+                &mut KafkaStream::Ssl(ref mut s) => s.read(buf),
+            }
         }
     }
-    fn flush(&mut self) -> io::Result<()> {
-        match self {
-            &mut KafkaStream::Plain(ref mut s) => s.flush(),
-            &mut KafkaStream::Ssl(ref mut s) => s.flush(),
+
+    impl Write for KafkaStream {
+        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+            match self {
+                &mut KafkaStream::Plain(ref mut s) => s.write(buf),
+                &mut KafkaStream::Ssl(ref mut s) => s.write(buf),
+            }
+        }
+        fn flush(&mut self) -> io::Result<()> {
+            match self {
+                &mut KafkaStream::Plain(ref mut s) => s.flush(),
+                &mut KafkaStream::Ssl(ref mut s) => s.flush(),
+            }
         }
     }
 }
 
 pub struct KafkaConnection {
     host: String,
-    stream: KafkaStream
+    stream: KafkaStream,
 }
 
 impl fmt::Debug for KafkaConnection {
@@ -73,17 +97,27 @@ impl KafkaConnection {
         }
     }
 
-    pub fn new(host: &str, timeout_secs: i32, security: Option<&SslContext>) -> Result<KafkaConnection> {
-        let plain_stream = try!(TcpStream::connect(host));
-        let stream = match security {
-            Some(ctx) => KafkaStream::Ssl(try!(SslStream::connect(ctx, plain_stream))),
-            None => KafkaStream::Plain(plain_stream),
-        };
+    fn from_stream(stream: KafkaStream, host: &str, timeout_secs: i32) -> KafkaConnection {
         if timeout_secs > 0 {
             let t = Some(Duration::from_secs(timeout_secs as u64));
-            stream.get_ref().set_read_timeout(t).expect("Set connection read-timeout");
-            stream.get_ref().set_write_timeout(t).expect("Set connection write-timeout");
+            stream.set_read_timeout(t).expect("Set connection read-timeout");
+            stream.set_write_timeout(t).expect("Set connection write-timeout");
         }
-        Ok(KafkaConnection{host: host.to_owned(), stream: stream})
+        KafkaConnection{host: host.to_owned(), stream: stream}
+    }
+
+    #[cfg(not(feature = "security"))]
+    pub fn new(host: &str, timeout_secs: i32) -> Result<KafkaConnection> {
+        Ok(KafkaConnection::from_stream(try!(TcpStream::connect(host)), host, timeout_secs))
+    }
+
+    #[cfg(feature = "security")]
+    pub fn new(host: &str, timeout_secs: i32, security: Option<&SslContext>) -> Result<KafkaConnection> {
+        let stream = try!(TcpStream::connect(host));
+        let stream = match security {
+            Some(ctx) => KafkaStream::Ssl(try!(SslStream::connect(ctx, stream))),
+            None => KafkaStream::Plain(stream),
+        };
+        Ok(KafkaConnection::from_stream(stream, host, timeout_secs))
     }
 }
diff --git a/src/consumer.rs b/src/consumer.rs
index 56778aae2..35c78828a 100644
--- a/src/consumer.rs
+++ b/src/consumer.rs
@@ -53,12 +53,17 @@ use client::{self, KafkaClient, FetchPartition, CommitOffset, FetchGroupOffset};
 use client::metadata::Topics;
 use error::{Error, KafkaCode, Result};
 use client::fetch;
-use client::SecurityConfig;
 
 // public re-exports
 pub use client::fetch::Message;
 pub use client::FetchOffset;
 
+#[cfg(feature = "security")]
+use client::SecurityConfig;
+
+#[cfg(not(feature = "security"))]
+type SecurityConfig = ();
+
 /// The default value for `Builder::with_retry_max_bytes_limit`.
 pub const DEFAULT_RETRY_MAX_BYTES_LIMIT: i32 = 0;
 
@@ -664,6 +669,7 @@ impl Builder {
 
     /// Specifies the security config to use.
     /// See `KafkaClient::new_secure` for more info.
+    #[cfg(feature = "security")]
     pub fn with_security(mut self, sec: SecurityConfig) -> Builder {
         self.security_config = Some(sec);
         self
@@ -740,17 +746,27 @@ impl Builder {
         self
     }
 
+    #[cfg(not(feature = "security"))]
+    fn new_kafka_client(hosts: Vec<String>, _: Option<SecurityConfig>) -> KafkaClient {
+        KafkaClient::new(hosts)
+    }
+
+    #[cfg(feature = "security")]
+    fn new_kafka_client(hosts: Vec<String>, security: Option<SecurityConfig>) -> KafkaClient {
+        if let Some(security) = security {
+            KafkaClient::new_secure(hosts, security)
+        } else {
+            KafkaClient::new(hosts)
+        }
+    }
+
     /// Finally creates/builds a new consumer based on the so far
     /// supplied settings.
     pub fn create(self) -> Result<Consumer> {
         let mut client = match self.client {
             Some(client) => client,
             None => {
-                let mut client = if let Some(security_config) = self.security_config {
-                    KafkaClient::new_secure(self.hosts, security_config)
-                } else {
-                    KafkaClient::new(self.hosts)
-                };
+                let mut client = Self::new_kafka_client(self.hosts, self.security_config);
                 try!(client.load_metadata_all());
                 client
             }
diff --git a/src/error.rs b/src/error.rs
index 241b8e39b..c336d068d 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -6,6 +6,8 @@ use std::io;
 use std::fmt;
 
 use byteorder;
+
+#[cfg(feature = "security")]
 use openssl::ssl;
 
 
@@ -24,13 +26,18 @@ pub enum Error {
     /// An error as reported by a remote Kafka server
     Kafka(KafkaCode),
     /// An error as reported by OpenSsl
+    #[cfg(feature = "security")]
     Ssl(ssl::error::SslError),
 
     /// Failure to correctly parse the server response due to the
     /// server speaking a newer protocol version (than the one this
     /// library supports)
     UnsupportedProtocol,
+    /// Failure to correctly parse the server response by this library
+    /// due to an unsupported compression format of the data
+    UnsupportedCompression,
     /// Failure to decode a snappy compressed response from Kafka
+    #[cfg(feature = "snappy")]
     InvalidInputSnappy,
     /// Failure to decode a response due to an insufficient number of bytes available
     UnexpectedEOF,
@@ -163,6 +170,7 @@ impl From<byteorder::Error> for Error {
     }
 }
 
+#[cfg(feature = "security")]
 impl From<ssl::error::SslError> for Error {
     fn from(err: ssl::error::SslError) -> Error { Error::Ssl(err) }
 }
@@ -172,6 +180,7 @@ impl Clone for Error {
         match self {
             &Error::Io(ref err) => Error::Io(io::Error::new(err.kind(), "Io Error")),
             &Error::Kafka(x) => Error::Kafka(x),
+            #[cfg(feature = "security")]
             &Error::Ssl(ref x) => match x {
                 &ssl::error::SslError::StreamError(ref e) => 
                     Error::Ssl(ssl::error::SslError::StreamError(
@@ -183,6 +192,8 @@ impl Clone for Error {
                     Error::Ssl(ssl::error::SslError::OpenSslErrors(v.clone())),
             },
             &Error::UnsupportedProtocol => Error::UnsupportedProtocol,
+            &Error::UnsupportedCompression => Error::UnsupportedCompression,
+            #[cfg(feature = "snappy")]
             &Error::InvalidInputSnappy => Error::InvalidInputSnappy,
             &Error::UnexpectedEOF => Error::UnexpectedEOF,
             &Error::CodecError => Error::CodecError,
@@ -197,8 +208,11 @@ impl error::Error for Error {
         match *self {
             Error::Io(ref err) => error::Error::description(err),
             Error::Kafka(_) => "Kafka Error",
+            #[cfg(feature = "security")]
             Error::Ssl(ref err) => error::Error::description(err),
             Error::UnsupportedProtocol => "Unsupported protocol version",
+            Error::UnsupportedCompression => "Unsupported compression format",
+            #[cfg(feature = "snappy")]
             Error::InvalidInputSnappy => "Snappy decode error",
             Error::UnexpectedEOF => "Unexpected EOF",
             Error::CodecError => "Encoding/Decoding error",
@@ -220,8 +234,11 @@ impl fmt::Display for Error {
         match *self {
             Error::Io(ref err) => err.fmt(f),
             Error::Kafka(ref c) => write!(f, "Kafka Error ({:?})", c),
+            #[cfg(feature = "security")]
             Error::Ssl(ref err) => err.fmt(f),
             Error::UnsupportedProtocol => write!(f, "Unsupported protocol version"),
+            Error::UnsupportedCompression => write!(f, "Unsupported compression format"),
+            #[cfg(feature = "snappy")]
             Error::InvalidInputSnappy => write!(f, "Snappy decode error"),
             Error::UnexpectedEOF => write!(f, "Unexpected EOF"),
             Error::CodecError => write!(f, "Encoding/Decoding Error"),
diff --git a/src/lib.rs b/src/lib.rs
index e6ff359e6..c2f13786c 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -10,16 +10,22 @@
 #![cfg_attr(feature = "nightly", feature(test))]
 
 extern crate byteorder;
-extern crate flate2;
 extern crate crc;
-extern crate snappy;
 extern crate ref_slice;
 extern crate fnv;
-extern crate openssl;
 
 #[macro_use]
 extern crate log;
 
+#[cfg(feature = "security")]
+extern crate openssl;
+
+#[cfg(feature = "gzip")]
+extern crate flate2;
+
+#[cfg(feature = "snappy")]
+extern crate snappy;
+
 #[cfg(all(test, feature = "nightly"))]
 extern crate test;
 
diff --git a/src/producer.rs b/src/producer.rs
index 3e4e18637..49a4812e8 100644
--- a/src/producer.rs
+++ b/src/producer.rs
@@ -69,8 +69,13 @@ use error::Result;
 use utils::TopicPartitionOffset;
 
 use ref_slice::ref_slice;
+
+#[cfg(feature = "security")]
 use client::SecurityConfig;
 
+#[cfg(not(feature = "security"))]
+type SecurityConfig = ();
+
 /// The default value for `Builder::with_ack_timeout`.
 pub const DEFAULT_ACK_TIMEOUT: i32 = 30 * 1000;
 
@@ -336,6 +341,7 @@ impl Builder {
 
     /// Specifies the security config to use.
     /// See `KafkaClient::new_secure` for more info.
+    #[cfg(feature = "security")]
     pub fn with_security(mut self, security: SecurityConfig) -> Self {
         self.security_config = Some(security);
         self
@@ -389,17 +395,28 @@ impl<P> Builder<P> {
         }
     }
 
+
+    #[cfg(not(feature = "security"))]
+    fn new_kafka_client(hosts: Vec<String>, _: Option<SecurityConfig>) -> KafkaClient {
+        KafkaClient::new(hosts)
+    }
+
+    #[cfg(feature = "security")]
+    fn new_kafka_client(hosts: Vec<String>, security: Option<SecurityConfig>) -> KafkaClient {
+        if let Some(security) = security {
+            KafkaClient::new_secure(hosts, security)
+        } else {
+            KafkaClient::new(hosts)
+        }
+    }
+
     /// Finally creates/builds a new producer based on the so far
     /// supplied settings.
     pub fn create(self) -> Result<Producer<P>> {
         let mut client = match self.client {
             Some(client) => client,
             None => {
-                let mut client = if let Some(security_config) = self.security_config {
-                    KafkaClient::new_secure(self.hosts, security_config)
-                } else {
-                    KafkaClient::new(self.hosts)
-                };
+                let mut client = Self::new_kafka_client(self.hosts, self.security_config);
                 try!(client.load_metadata_all());
                 client
             }
diff --git a/src/protocol/fetch.rs b/src/protocol/fetch.rs
index 80586281a..b92bde6ca 100644
--- a/src/protocol/fetch.rs
+++ b/src/protocol/fetch.rs
@@ -1,7 +1,7 @@
 //! A representation of fetched messages from Kafka.
 
 use std::borrow::Cow;
-use std::io::{Read, Write};
+use std::io::Write;
 use std::mem;
 use std::cell::Cell;
 use std::collections::HashMap;
@@ -11,7 +11,10 @@ use fnv::FnvHasher;
 
 use codecs::ToByte;
 use error::{Error, KafkaCode, Result};
-use compression::{gzip, Compression};
+use compression::Compression;
+#[cfg(feature = "gzip")]
+use compression::gzip;
+#[cfg(feature = "snappy")]
 use compression::snappy::SnappyReader;
 
 use super::{HeaderRequest, API_KEY_FETCH, API_VERSION};
@@ -346,6 +349,7 @@ pub struct Message<'a> {
 }
 
 impl<'a> MessageSet<'a> {
+    #[allow(dead_code)]
     fn from_vec(data: Vec<u8>, req_offset: i64, validate_crc: bool) -> Result<MessageSet<'a>> {
         // since we're going to keep the original
         // uncompressed vector around without
@@ -393,16 +397,19 @@ impl<'a> MessageSet<'a> {
                             }
                         }
                         // XXX handle recursive compression in future
+                        #[cfg(feature = "gzip")]
                         c if c == Compression::GZIP as i8 => {
                             let v = try!(gzip::uncompress(pmsg.value));
                             return Ok(try!(MessageSet::from_vec(v, req_offset, validate_crc)));
                         }
+                        #[cfg(feature = "snappy")]
                         c if c == Compression::SNAPPY as i8 => {
+                            use std::io::Read;
                             let mut v = Vec::new();
                             try!(try!(SnappyReader::new(pmsg.value)).read_to_end(&mut v));
                             return Ok(try!(MessageSet::from_vec(v, req_offset, validate_crc)));
                         }
-                        _ => panic!("Unknown compression type!"),
+                        _ => return Err(Error::UnsupportedCompression),
                     }
                 }
             };
@@ -473,8 +480,10 @@ mod tests {
         include_bytes!("../../test-data/fetch1.mytopic.1p.nocompression.kafka.0821");
     static FETCH1_FETCH_RESPONSE_SNAPPY_K0821: &'static [u8] =
         include_bytes!("../../test-data/fetch1.mytopic.1p.snappy.kafka.0821");
+    #[cfg(feature = "snappy")]
     static FETCH1_FETCH_RESPONSE_SNAPPY_K0822: &'static [u8] =
         include_bytes!("../../test-data/fetch1.mytopic.1p.snappy.kafka.0822");
+    #[cfg(feature = "gzip")]
     static FETCH1_FETCH_RESPONSE_GZIP_K0821: &'static [u8] =
         include_bytes!("../../test-data/fetch1.mytopic.1p.gzip.kafka.0821");
 
@@ -542,14 +551,26 @@ mod tests {
         test_decode_new_fetch_response(
             FETCH1_TXT, FETCH1_FETCH_RESPONSE_NOCOMPRESSION_K0821.to_owned(), Some(&req), false);
 
-        // ~ pretend we asked for messages as of offset-one (while the
-        // server delivered the zero-offset message as well)
+        // ~ pretend we asked for messages as of offset five (while
+        // the server delivered the zero-offset message as well)
         req = FetchRequest::new(0, "test", -1, -1);
         req.add("my-topic", 0, 5, -1);
         test_decode_new_fetch_response(
-            skip_lines(FETCH1_TXT, 5), FETCH1_FETCH_RESPONSE_SNAPPY_K0821.to_owned(), Some(&req), false);
+            skip_lines(FETCH1_TXT, 5), FETCH1_FETCH_RESPONSE_NOCOMPRESSION_K0821.to_owned(), Some(&req), false);
+    }
+
+    // verify we don't crash but cleanly fail and report we don't
+    // support the compression
+    #[cfg(not(feature = "snappy"))]
+    #[test]
+    fn test_unsupported_compression_snappy() {
+        let mut req = FetchRequest::new(0, "test", -1, -1);
+        req.add("my-topic", 0, 0, -1);
+        let r = Response::from_vec(FETCH1_FETCH_RESPONSE_SNAPPY_K0821.to_owned(), Some(&req), false);
+        assert!(match r { Err(Error::UnsupportedCompression) => true, _ => false });
     }
 
+    #[cfg(feature = "snappy")]
     #[test]
     fn test_from_slice_snappy_k0821() {
         let mut req = FetchRequest::new(0, "test", -1, -1);
@@ -557,14 +578,15 @@ mod tests {
         test_decode_new_fetch_response(
             FETCH1_TXT, FETCH1_FETCH_RESPONSE_SNAPPY_K0821.to_owned(), Some(&req), false);
 
-        // ~ pretend we asked for messages as of offset-one (while the
-        // server delivered the zero-offset message as well)
+        // ~ pretend we asked for messages as of offset three (while
+        // the server delivered the zero-offset message as well)
         req = FetchRequest::new(0, "test", -1, -1);
         req.add("my-topic", 0, 3, -1);
         test_decode_new_fetch_response(
             skip_lines(FETCH1_TXT, 3), FETCH1_FETCH_RESPONSE_SNAPPY_K0821.to_owned(), Some(&req), false);
     }
 
+    #[cfg(feature = "snappy")]
     #[test]
     fn test_from_slice_snappy_k0822() {
         let mut req = FetchRequest::new(0, "test", -1, -1);
@@ -573,6 +595,7 @@ mod tests {
             FETCH1_TXT, FETCH1_FETCH_RESPONSE_SNAPPY_K0822.to_owned(), Some(&req), false);
     }
 
+    #[cfg(feature = "gzip")]
     #[test]
     fn test_from_slice_gzip_k0821() {
         let mut req = FetchRequest::new(0, "test", -1, -1);
@@ -580,13 +603,13 @@ mod tests {
         test_decode_new_fetch_response(
             FETCH1_TXT, FETCH1_FETCH_RESPONSE_GZIP_K0821.to_owned(), Some(&req), false);
 
-        // ~ pretend we asked for messages as of offset-one (while the
+        // ~ pretend we asked for messages as of offset one (while the
         // server delivered the zero-offset message as well)
         req.add("my-topic", 0, 1, -1);
         test_decode_new_fetch_response(
             skip_lines(FETCH1_TXT, 1), FETCH1_FETCH_RESPONSE_GZIP_K0821.to_owned(), Some(&req), false);
 
-        // ~ pretend we asked for messages as of offset-one (while the
+        // ~ pretend we asked for messages as of offset ten (while the
         // server delivered the zero-offset message as well)
         req.add("my-topic", 0, 10, -1);
         test_decode_new_fetch_response(
@@ -635,16 +658,19 @@ mod tests {
             bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_NOCOMPRESSION_K0821.to_owned(), false)
         }
 
+        #[cfg(feature = "snappy")]
         #[bench]
         fn bench_decode_new_fetch_response_snappy_k0821(b: &mut Bencher) {
             bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_SNAPPY_K0821.to_owned(), false)
         }
 
+        #[cfg(feature = "snappy")]
         #[bench]
         fn bench_decode_new_fetch_response_snappy_k0822(b: &mut Bencher) {
             bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_SNAPPY_K0822.to_owned(), false)
         }
 
+        #[cfg(feature = "gzip")]
         #[bench]
         fn bench_decode_new_fetch_response_gzip_k0821(b: &mut Bencher) {
             bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_GZIP_K0821.to_owned(), false)
@@ -655,16 +681,19 @@ mod tests {
             bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_NOCOMPRESSION_K0821.to_owned(), true)
         }
 
+        #[cfg(feature = "snappy")]
         #[bench]
         fn bench_decode_new_fetch_response_snappy_k0821_validate_crc(b: &mut Bencher) {
             bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_SNAPPY_K0821.to_owned(), true)
         }
 
+        #[cfg(feature = "snappy")]
         #[bench]
         fn bench_decode_new_fetch_response_snappy_k0822_validate_crc(b: &mut Bencher) {
             bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_SNAPPY_K0822.to_owned(), true)
         }
 
+        #[cfg(feature = "gzip")]
         #[bench]
         fn bench_decode_new_fetch_response_gzip_k0821_validate_crc(b: &mut Bencher) {
             bench_decode_new_fetch_response(b, super::FETCH1_FETCH_RESPONSE_GZIP_K0821.to_owned(), true)
diff --git a/src/protocol/produce.rs b/src/protocol/produce.rs
index 1b4c1cf89..9a2eea739 100644
--- a/src/protocol/produce.rs
+++ b/src/protocol/produce.rs
@@ -1,7 +1,12 @@
 use std::io::{Read, Write};
 
 use codecs::{ToByte, FromByte};
-use compression::{Compression, gzip, snappy};
+use compression::Compression;
+#[cfg(feature = "gzip")]
+use compression::gzip;
+#[cfg(feature = "snappy")]
+use compression::snappy;
+
 use error::{Error, Result};
 use utils::TopicPartitionOffset;
 
@@ -138,34 +143,41 @@ impl<'a> PartitionProduceRequest<'a> {
         for msg in &self.messages {
             try!(msg._encode_to_buf(&mut buf, MESSAGE_MAGIC_BYTE, 0));
         }
-        if let Compression::NONE = compression {
-            buf.encode(out)
-        } else {
-            // ~ Compress the so far produced MessageSet (into a newly allocated buffer)
-            let cdata = match compression {
-                Compression::GZIP => try!(gzip::compress(&buf)),
-                Compression::SNAPPY => try!(snappy::compress(&buf)),
-                c => panic!("Unknown compression type: {:?}", c),
-            };
-            // ~ now wrap the compressed data into a new MessagesSet
-            // with exactly one Message
-            buf.clear();
-            let cmsg = MessageProduceRequest::from_value(&cdata);
-            try!(cmsg._encode_to_buf(&mut buf, MESSAGE_MAGIC_BYTE, compression as i8));
-            buf.encode(out)
+        match compression {
+            Compression::NONE => {
+                // ~ nothing to do
+            }
+            #[cfg(feature = "gzip")]
+            Compression::GZIP => {
+                let cdata = try!(gzip::compress(&buf));
+                try!(render_compressed(&mut buf, &cdata, compression));
+            }
+            #[cfg(feature = "snappy")]
+            Compression::SNAPPY => {
+                let cdata = try!(snappy::compress(&buf));
+                try!(render_compressed(&mut buf, &cdata, compression));
+            }
         }
+        buf.encode(out)
     }
 }
 
+// ~ A helper method to render `cdata` into `out` as a compressed message.
+// ~ `out` is first cleared and then populated with the rendered message.
+#[cfg(any(feature = "snappy", feature = "gzip"))]
+fn render_compressed(out: &mut Vec<u8>, cdata: &[u8], compression: Compression)
+                     -> Result<()>
+{
+    out.clear();
+    let cmsg = MessageProduceRequest::new(None, Some(cdata));
+    cmsg._encode_to_buf(out, MESSAGE_MAGIC_BYTE, compression as i8)
+}
+
 impl<'a> MessageProduceRequest<'a> {
     fn new<'b>(key: Option<&'b [u8]>, value: Option<&'b [u8]>) -> MessageProduceRequest<'b> {
         MessageProduceRequest { key: key, value: value }
     }
 
-    fn from_value<'b>(value: &'b [u8]) -> MessageProduceRequest<'b> {
-        MessageProduceRequest::new(None, Some(value))
-    }
-
     // render a single message as: Offset MessageSize Message
     //
     // Offset => int64 (always encoded as zero here)


Test patch: