private final ValueWriter<Object> jdkWriter;
private final ValueWriter<Writable> hadoopWriter;

public CascadingValueWriter() {
    this(false);
}

public CascadingValueWriter(boolean writeUnknownTypes) {
    jdkWriter = new JdkValueWriter(writeUnknownTypes);
    hadoopWriter = new WritableValueWriter(writeUnknownTypes);
}

@SuppressWarnings("unchecked")
@Override
public boolean write(SinkCall<Object[], ?> sinkCall, Generator generator) {
    Tuple tuple = sinkCall.getOutgoingEntry().getTuple();
    List<String> names = (List<String>) sinkCall.getContext()[0];

    generator.writeBeginObject();
    for (int i = 0; i < tuple.size(); i++) {
        String name = (i < names.size() ? names.get(i) : "tuple" + i);
        generator.writeFieldName(name);
        if (!jdkWriter.write(tuple.getObject(i), generator)) {
            //System.out.println("UNABLE TO WRITE - " + tuple.getObject(i) + " of type class=" + tuple.getObject(i).getClass());
            Object obj = tuple.getObject(i);
            if (obj instanceof Writable && hadoopWriter.write((Writable)obj, generator)) {
                continue;
            }
            return false;
        }
    }
    generator.writeEndObject();
    return true;
}