HADOOP-3208. Fix WritableDeserializer to set the Configuration on

deserialized Writables. Contributed by Enis Soztutar.



git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/branches/branch-0.17@646642 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -539,6 +539,9 @@ Release 0.17.0 - Unreleased
 
     HADOOP-3220. Safemode message corrected. (shv)
 
+    HADOOP-3208. Fix WritableDeserializer to set the Configuration on
+    deserialized Writables. (Enis Soztutar via cdouglas)
+
 Release 0.16.3 - Unreleased
 
   BUG FIXES

src/java/org/apache/hadoop/io/serializer/WritableSerialization.java

@@ -24,6 +24,8 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.Configured;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.util.ReflectionUtils;
 
@@ -32,14 +34,17 @@ import org.apache.hadoop.util.ReflectionUtils;
  * {@link Writable#write(java.io.DataOutput)} and
  * {@link Writable#readFields(java.io.DataInput)}.
  */
-public class WritableSerialization implements Serialization<Writable> {
+public class WritableSerialization extends Configured 
+  implements Serialization<Writable> {
   
-  static class WritableDeserializer implements Deserializer<Writable> {
+  static class WritableDeserializer extends Configured 
+    implements Deserializer<Writable> {
 
     private Class<?> writableClass;
     private DataInputStream dataIn;
     
-    public WritableDeserializer(Class<?> c) {
+    public WritableDeserializer(Configuration conf, Class<?> c) {
+      setConf(conf);
       this.writableClass = c;
     }
     
@@ -54,7 +59,8 @@ public class WritableSerialization implements Serialization<Writable> {
     public Writable deserialize(Writable w) throws IOException {
       Writable writable;
       if (w == null) {
-        writable = (Writable) ReflectionUtils.newInstance(writableClass, null);
+        writable 
+          = (Writable) ReflectionUtils.newInstance(writableClass, getConf());
       } else {
         writable = w;
       }
@@ -95,7 +101,7 @@ public class WritableSerialization implements Serialization<Writable> {
   }
 
   public Deserializer<Writable> getDeserializer(Class<Writable> c) {
-    return new WritableDeserializer(c);
+    return new WritableDeserializer(getConf(), c);
   }
 
   public Serializer<Writable> getSerializer(Class<Writable> c) {

src/test/org/apache/hadoop/io/TestGenericWritable.java

@@ -22,11 +22,11 @@ import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
 
+import junit.framework.TestCase;
+
 import org.apache.hadoop.conf.Configurable;
 import org.apache.hadoop.conf.Configuration;
 
-import junit.framework.TestCase;
-
 /**
  * TestCase for {@link GenericWritable} class.
  * @see TestWritable#testWritable(Writable)
@@ -34,9 +34,10 @@ import junit.framework.TestCase;
 public class TestGenericWritable extends TestCase {
 
   private Configuration conf;
-  private static final String CONF_TEST_KEY = "test.generic.writable";
-  private static final String CONF_TEST_VALUE = "dummy";
+  public static final String CONF_TEST_KEY = "test.generic.writable";
+  public static final String CONF_TEST_VALUE = "dummy";
 
+  @Override
   protected void setUp() throws Exception {
     super.setUp();
     conf = new Configuration();
@@ -53,6 +54,7 @@ public class TestGenericWritable extends TestCase {
     public void write(DataOutput out) throws IOException {
       Text.writeString(out, foo);
     }
+    @Override
     public boolean equals(Object obj) {
       if (!(obj instanceof Foo))
         return false;
@@ -75,6 +77,7 @@ public class TestGenericWritable extends TestCase {
     public void setConf(Configuration conf) {
       this.conf = conf;
     }
+    @Override
     public boolean equals(Object obj) {
       if (!(obj instanceof Bar))
         return false;
@@ -84,12 +87,14 @@ public class TestGenericWritable extends TestCase {
 
   /** Dummy class for testing {@link GenericWritable} */
   public static class Baz extends Bar {
+    @Override
     public void readFields(DataInput in) throws IOException {
       super.readFields(in);
       //needs a configuration parameter
       assertEquals("Configuration is not set for the wrapped object", 
           CONF_TEST_VALUE, getConf().get(CONF_TEST_KEY)); 
     }
+    @Override
     public void write(DataOutput out) throws IOException {
       super.write(out);
     }
@@ -97,10 +102,12 @@ public class TestGenericWritable extends TestCase {
 
   /** Dummy class for testing {@link GenericWritable} */ 
   public static class FooGenericWritable extends GenericWritable {
+    @Override
     @SuppressWarnings("unchecked")
     protected Class<? extends Writable>[] getTypes() {
       return new Class[] {Foo.class, Bar.class, Baz.class};
     }
+    @Override
     public boolean equals(Object obj) {
       if(! (obj instanceof FooGenericWritable))
         return false;

src/test/org/apache/hadoop/io/serializer/TestWritableSerialization.java

@@ -0,0 +1,95 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF 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.apache.hadoop.io.serializer;
+
+import static org.apache.hadoop.io.TestGenericWritable.CONF_TEST_KEY;
+import static org.apache.hadoop.io.TestGenericWritable.CONF_TEST_VALUE;
+import junit.framework.TestCase;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.io.DataOutputBuffer;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.TestGenericWritable.Baz;
+import org.apache.hadoop.io.TestGenericWritable.FooGenericWritable;
+import org.apache.hadoop.util.GenericsUtil;
+
+public class TestWritableSerialization extends TestCase {
+
+  private static final Configuration conf = new Configuration();
+  
+  static {
+    conf.set("io.serializations"
+        , "org.apache.hadoop.io.serializer.WritableSerialization");
+  }
+  
+  public void testWritableSerialization() throws Exception {
+    Text before = new Text("test writable"); 
+    testSerialization(conf, before);
+  }
+  
+  
+  public void testWritableConfigurable() throws Exception {
+    
+    //set the configuration parameter
+    conf.set(CONF_TEST_KEY, CONF_TEST_VALUE);
+
+    //reuse TestGenericWritable inner classes to test 
+    //writables that also implement Configurable.
+    FooGenericWritable generic = new FooGenericWritable();
+    generic.setConf(conf);
+    Baz baz = new Baz();
+    generic.set(baz);
+    Baz result = testSerialization(conf, baz);
+    assertNotNull(result.getConf());
+  }
+  
+  /**
+   * A utility that tests serialization/deserialization. 
+   * @param <K> the class of the item
+   * @param conf configuration to use, "io.serializations" is read to 
+   * determine the serialization
+   * @param before item to (de)serialize
+   * @return deserialized item
+   */
+  public static<K> K testSerialization(Configuration conf, K before) 
+    throws Exception {
+    
+    SerializationFactory factory = new SerializationFactory(conf);
+    Serializer<K> serializer 
+      = factory.getSerializer(GenericsUtil.getClass(before));
+    Deserializer<K> deserializer 
+      = factory.getDeserializer(GenericsUtil.getClass(before));
+   
+    DataOutputBuffer out = new DataOutputBuffer();
+    serializer.open(out);
+    serializer.serialize(before);
+    serializer.close();
+    
+    DataInputBuffer in = new DataInputBuffer();
+    in.reset(out.getData(), out.getLength());
+    deserializer.open(in);
+    K after = deserializer.deserialize(null);
+    deserializer.close();
+    
+    assertEquals(before, after);
+    return after;
+  }
+  
+}