diff --git a/hadoop-common-project/hadoop-common/CHANGES.txt b/hadoop-common-project/hadoop-common/CHANGES.txt
index 119d5ae55a..18e2f8fdd7 100644
--- a/hadoop-common-project/hadoop-common/CHANGES.txt
+++ b/hadoop-common-project/hadoop-common/CHANGES.txt
@@ -104,6 +104,10 @@ Trunk (Unreleased)
     HADOOP-8814. Replace string equals "" by String#isEmpty().
     (Brandon Li via suresh)
 
+    HADOOP-8588. SerializationFactory shouldn't throw a
+    NullPointerException if the serializations list is empty.
+    (Sho Shimauchi via harsh)
+
   BUG FIXES
 
     HADOOP-8177. MBeans shouldn't try to register when it fails to create MBeanName.
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/serializer/SerializationFactory.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/serializer/SerializationFactory.java
index 52a0a253bb..d6c6588e79 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/serializer/SerializationFactory.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/serializer/SerializationFactory.java
@@ -40,12 +40,12 @@
 @InterfaceAudience.LimitedPrivate({"HDFS", "MapReduce"})
 @InterfaceStability.Evolving
 public class SerializationFactory extends Configured {
-  
-  private static final Log LOG =
+
+  static final Log LOG =
     LogFactory.getLog(SerializationFactory.class.getName());
 
   private List<Serialization<?>> serializations = new ArrayList<Serialization<?>>();
-  
+
   /**
    * <p>
    * Serializations are found by reading the <code>io.serializations</code>
@@ -55,15 +55,21 @@ public class SerializationFactory extends Configured {
    */
   public SerializationFactory(Configuration conf) {
     super(conf);
-    for (String serializerName : conf.getStrings(
-      CommonConfigurationKeys.IO_SERIALIZATIONS_KEY,
-      new String[]{WritableSerialization.class.getName(),
-        AvroSpecificSerialization.class.getName(),
-        AvroReflectSerialization.class.getName()})) {
-      add(conf, serializerName);
+    if (conf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY).equals("")) {
+      LOG.warn("Serialization for various data types may not be available. Please configure "
+          + CommonConfigurationKeys.IO_SERIALIZATIONS_KEY
+          + " properly to have serialization support (it is currently not set).");
+    } else {
+      for (String serializerName : conf.getStrings(
+          CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, new String[] {
+              WritableSerialization.class.getName(),
+              AvroSpecificSerialization.class.getName(),
+              AvroReflectSerialization.class.getName() })) {
+        add(conf, serializerName);
+      }
     }
   }
-  
+
   @SuppressWarnings("unchecked")
   private void add(Configuration conf, String serializationName) {
     try {
@@ -101,5 +107,5 @@ public <T> Serialization<T> getSerialization(Class<T> c) {
     }
     return null;
   }
-  
+
 }
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/serializer/TestSerializationFactory.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/serializer/TestSerializationFactory.java
index 18c2637ec5..c5805be5b4 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/serializer/TestSerializationFactory.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/serializer/TestSerializationFactory.java
@@ -17,27 +17,62 @@
  */
 package org.apache.hadoop.io.serializer;
 
+import org.junit.BeforeClass;
 import org.junit.Test;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertNotNull;
 
+import org.apache.commons.logging.impl.Log4JLogger;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeys;
 import org.apache.hadoop.io.Writable;
+import org.apache.log4j.Level;
 
 public class TestSerializationFactory {
 
+  static {
+    ((Log4JLogger) SerializationFactory.LOG).getLogger().setLevel(Level.ALL);
+  }
+
+  static Configuration conf;
+  static SerializationFactory factory;
+
+  @BeforeClass
+  public static void setup() throws Exception {
+    conf = new Configuration();
+    factory = new SerializationFactory(conf);
+  }
+
   @Test
-  public void testSerializerAvailability() {
+  public void testSerializationKeyIsEmpty() {
     Configuration conf = new Configuration();
+    conf.set(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, "");
     SerializationFactory factory = new SerializationFactory(conf);
+  }
+
+  @Test
+  public void testSerializationKeyIsInvalid() {
+    Configuration conf = new Configuration();
+    conf.set(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, "INVALID_KEY_XXX");
+    SerializationFactory factory = new SerializationFactory(conf);
+  }
+
+  @Test
+  public void testGetSerializer() {
     // Test that a valid serializer class is returned when its present
-    assertNotNull("A valid class must be returned for default Writable Serde",
+    assertNotNull("A valid class must be returned for default Writable SerDe",
         factory.getSerializer(Writable.class));
-    assertNotNull("A valid class must be returned for default Writable serDe",
-        factory.getDeserializer(Writable.class));
     // Test that a null is returned when none can be found.
     assertNull("A null should be returned if there are no serializers found.",
         factory.getSerializer(TestSerializationFactory.class));
+  }
+
+  @Test
+  public void testGetDeserializer() {
+    // Test that a valid serializer class is returned when its present
+    assertNotNull("A valid class must be returned for default Writable SerDe",
+        factory.getDeserializer(Writable.class));
+    // Test that a null is returned when none can be found.
     assertNull("A null should be returned if there are no deserializers found",
         factory.getDeserializer(TestSerializationFactory.class));
   }