### Question:
ValueToWritableFunction implements PairFunction<Tuple2<K,M>,Writable,Writable> { @Override public Tuple2<Writable,Writable> call(Tuple2<K,M> keyMessage) { return new Tuple2<>(keyConverter.toWritable(keyMessage._1()), messageConverter.toWritable(keyMessage._2())); } ValueToWritableFunction(Class<K> keyClass,
                          Class<M> messageClass,
                          Class<? extends Writable> keyWritableClass,
                          Class<? extends Writable> messageWritableClass); @Override Tuple2<Writable,Writable> call(Tuple2<K,M> keyMessage);  }### Answer:
@Test public void testFunction() { ValueToWritableFunction<Integer,Long> function = new ValueToWritableFunction<>(Integer.class, Long.class, IntWritable.class, LongWritable.class); Tuple2<Integer,Long> in = new Tuple2<>(3, 4L); Tuple2<Writable,Writable> out = function.call(in); assertEquals(new IntWritable(3), out._1()); assertEquals(new LongWritable(4L), out._2()); }
@Test public void testNullKey() { ValueToWritableFunction<String,String> function = new ValueToWritableFunction<>(String.class, String.class, Text.class, Text.class); Tuple2<Writable,Writable> keyMessage = function.call(new Tuple2<>(null, "foo")); assertEquals(new Text(), keyMessage._1()); assertEquals(new Text("foo"), keyMessage._2()); keyMessage = function.call(new Tuple2<>("foo", null)); assertEquals(new Text("foo"), keyMessage._1()); assertEquals(new Text(), keyMessage._2()); }