FIxing WritableBytesConverter to not overwrite bytes with wrong value (…

…elastic#2075)

mr/src/main/java/org/elasticsearch/hadoop/mr/SafeWritableConverter.java

@@ -31,15 +31,26 @@ public SafeWritableConverter() {
         Text.class.getName(); // force class to be loaded
     }
 
-    public void invoke(Object from, BytesArray to) {
+    /**
+     * If from is a Text or BytesWritable, the value is written to the to field, and true is returned. Otherwise does not write to
+     * the to field and returns false.
+     * @param from The object to copy bytes from
+     * @param to The BytesArray to copy bytes to
+     * @return true if from has been handled
+     */
+    public boolean invoke(Object from, BytesArray to) {
         // handle common cases
         if (from instanceof Text) {
             Text t = (Text) from;
             to.bytes(t.getBytes(), t.getLength());
+            return true;
         }
-        if (from instanceof BytesWritable) {
+        else if (from instanceof BytesWritable) {
             BytesWritable b = (BytesWritable) from;
             to.bytes(b.getBytes(), b.getLength());
+            return true;
+        } else {
+            return false;
         }
     }
 }

mr/src/main/java/org/elasticsearch/hadoop/mr/WritableBytesConverter.java

@@ -35,10 +35,14 @@ public class WritableBytesConverter extends JdkBytesConverter {
 
     @Override
     public void convert(Object from, BytesArray to) {
-
-        if (safeWritableConverter != null)
-            safeWritableConverter.invoke(from, to);
-
-        super.convert(from, to);
+        final boolean handled;
+        if (safeWritableConverter != null) {
+            handled = safeWritableConverter.invoke(from, to);
+        } else {
+            handled = false;
+        }
+        if (handled == false) {
+            super.convert(from, to);
+        }
     }
-}
+}

mr/src/test/java/org/elasticsearch/hadoop/mr/WritableBytesConverterTests.java

@@ -0,0 +1,74 @@
+package org.elasticsearch.hadoop.mr;
+
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.Text;
+import org.elasticsearch.hadoop.util.BytesArray;
+import org.junit.Test;
+
+import java.security.SecureRandom;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+
+public class WritableBytesConverterTests {
+
+    @Test
+    public void testConvertBytesWritable() throws Exception {
+        WritableBytesConverter converter = new WritableBytesConverter();
+        byte[] randomBytes = new byte[20];
+        SecureRandom.getInstanceStrong().nextBytes(randomBytes);
+        Object input = new BytesWritable(randomBytes);
+        BytesArray output = new BytesArray(10);
+        converter.convert(input, output);
+        assertEquals(randomBytes.length, output.length());
+        assertArrayEquals(randomBytes, output.bytes());
+    }
+
+    @Test
+    public void testConvertInteger() {
+        WritableBytesConverter converter = new WritableBytesConverter();
+        int inputInteger = ThreadLocalRandom.current().nextInt();
+        Object input = inputInteger;
+        BytesArray output = new BytesArray(10);
+        converter.convert(input, output);
+        // Integer is not directly supported, so we expect its string value to be used:
+        assertEquals(Integer.toString(inputInteger), output.toString());
+    }
+
+    @Test
+    public void testConvertText() {
+        WritableBytesConverter converter = new WritableBytesConverter();
+        Text input = new Text("This is some text");
+        BytesArray output = new BytesArray(10);
+        converter.convert(input, output);
+        assertEquals(input.getLength(), output.length());
+        assertArrayEquals(input.getBytes(), output.bytes());
+    }
+
+    @Test
+    public void testConvertByteArray() throws Exception {
+        WritableBytesConverter converter = new WritableBytesConverter();
+        byte[] input = new byte[20];
+        SecureRandom.getInstanceStrong().nextBytes(input);
+        BytesArray output = new BytesArray(10);
+        converter.convert(input, output);
+        assertEquals(input.length, output.length());
+        assertArrayEquals(input, output.bytes());
+    }
+
+    @Test
+    public void testConvertBytesArray() throws Exception {
+        WritableBytesConverter converter = new WritableBytesConverter();
+        byte[] randomBytes = new byte[20];
+        SecureRandom.getInstanceStrong().nextBytes(randomBytes);
+        Object input = new BytesArray(randomBytes);
+        BytesArray output = new BytesArray(10);
+        converter.convert(input, output);
+        assertEquals(randomBytes.length, output.length());
+        byte[] usedOutputBytes = new byte[output.length()];
+        // BytesArray::bytes can give you bytes beyond length
+        System.arraycopy(output.bytes(), 0, usedOutputBytes, 0, output.length());
+        assertArrayEquals(randomBytes, usedOutputBytes);
+    }
+}