[SPARK-4079] [CORE] Default to LZF if Snappy not available

By default, snappy is the compression codec used.
If Snappy is not available, Spark currently throws
a stack trace. Now Spark falls back to LZF
if Snappy is not available on the cluster and logs
a warning message.

The only exception is if the user has explicitly
set spark.io.compression.codec=snappy. In this
case, if snappy is not available, an
IllegalArgumentException is thrown.

Because of the way the Snappy library uses static
initialization, it was very difficult in a unit test to
simulate Snappy not being available. The only way I
could think of was to create multiple classloaders
which seemed excessive. As a result, most of this was tested
adhoc on a test cluster by modifying the system property:
org.xerial.snappy.use.systemlib=true which caused Snappy
to not load and thus triggering this logic.

Author: Kostas Sakellis

core/src/main/scala/org/apache/spark/io/CompressionCodec.scala

@@ -21,11 +21,12 @@ import java.io.{InputStream, OutputStream}
 
 import com.ning.compress.lzf.{LZFInputStream, LZFOutputStream}
 import net.jpountz.lz4.{LZ4BlockInputStream, LZ4BlockOutputStream}
-import org.xerial.snappy.{SnappyInputStream, SnappyOutputStream}
+import org.xerial.snappy.{Snappy, SnappyInputStream, SnappyOutputStream}
 
 import org.apache.spark.SparkConf
 import org.apache.spark.annotation.DeveloperApi
 import org.apache.spark.util.Utils
+import org.apache.spark.Logging
 
 /**
  * :: DeveloperApi ::
@@ -42,28 +43,52 @@ trait CompressionCodec {
   def compressedOutputStream(s: OutputStream): OutputStream
 
   def compressedInputStream(s: InputStream): InputStream
-}
 
+  def isAvailable() : Boolean = true
+}
 
-private[spark] object CompressionCodec {
+private[spark] object CompressionCodec extends Logging {
 
+  private val configKey = "spark.io.compression.codec"
   private val shortCompressionCodecNames = Map(
     "lz4" -> classOf[LZ4CompressionCodec].getName,
     "lzf" -> classOf[LZFCompressionCodec].getName,
     "snappy" -> classOf[SnappyCompressionCodec].getName)
 
   def createCodec(conf: SparkConf): CompressionCodec = {
-    createCodec(conf, conf.get("spark.io.compression.codec", DEFAULT_COMPRESSION_CODEC))
+    conf.getOption(configKey)
+        .map(createCodec(conf, _))
+        .orElse(createCodecFromName(conf, DEFAULT_COMPRESSION_CODEC))
+        .orElse({
+          logWarning("Default codec " + DEFAULT_COMPRESSION_CODEC +
+            " is unavailable. Faling back to " + FALLBACK_COMPRESSION_CODEC)
+          createCodecFromName(conf, FALLBACK_COMPRESSION_CODEC)
+        })
+        .getOrElse(throw new IllegalArgumentException("The codec [" +
+          FALLBACK_COMPRESSION_CODEC + "] is not available."))
   }
 
   def createCodec(conf: SparkConf, codecName: String): CompressionCodec = {
+    createCodecFromName(conf, codecName)
+      .getOrElse(throw new IllegalArgumentException("The specified codec [" +
+                  codecName + "] is not available."))
+  }
+
+  private def createCodecFromName(conf: SparkConf, codecName : String)
+      : Option[CompressionCodec] = {
     val codecClass = shortCompressionCodecNames.getOrElse(codecName.toLowerCase, codecName)
-    val ctor = Class.forName(codecClass, true, Utils.getContextOrSparkClassLoader)
-      .getConstructor(classOf[SparkConf])
-    ctor.newInstance(conf).asInstanceOf[CompressionCodec]
+    try {
+      val ctor = Class.forName(codecClass, true, Utils.getContextOrSparkClassLoader)
+        .getConstructor(classOf[SparkConf])
+      Some(ctor.newInstance(conf).asInstanceOf[CompressionCodec])
+        .filter(_.isAvailable())
+    } catch {
+      case e: ClassNotFoundException => None
+    }
   }
 
   val DEFAULT_COMPRESSION_CODEC = "snappy"
+  val FALLBACK_COMPRESSION_CODEC = "lzf"
   val ALL_COMPRESSION_CODECS = shortCompressionCodecNames.values.toSeq
 }
 
@@ -126,4 +151,13 @@ class SnappyCompressionCodec(conf: SparkConf) extends CompressionCodec {
   }
 
   override def compressedInputStream(s: InputStream): InputStream = new SnappyInputStream(s)
+
+  override def isAvailable() = {
+    try {
+      Snappy.getNativeLibraryVersion
+      true
+    } catch {
+      case e: Error => false
+    }
+  }
 }

core/src/test/scala/org/apache/spark/io/CompressionCodecSuite.scala

@@ -85,4 +85,10 @@ class CompressionCodecSuite extends FunSuite {
     assert(codec.getClass === classOf[SnappyCompressionCodec])
     testCodec(codec)
   }
+
+  test("bad compression codec") {
+    intercept[IllegalArgumentException] {
+      CompressionCodec.createCodec(conf, "foobar")
+    }
+  }
 }