Problems using custom output format

[Rakhi Khatwani]

Hi,
      I was tryin to run a map reduce program which reads a txt file filled
wid some keywords,
my map task takes each of these keywords, does some processing and returns a
complex object url which contains media, status, link and title (each being
a string).
my reduce class simply has one line
while(value.hasnext()){
output.collect(key, value.next())
}
i tried the following configuration:
 conf.setInputFormat(TextInputFormat.class);
  conf.setOutputFormat(TextOutputFormat.class);
  conf.setMapOutputKeyClass(Text.class);
  conf.setMapOutputValueClass(PObjectWritable.class);
  conf.setOutputKeyClass(Text.class);
  conf.setOutputValueClass(PObjectWritable.class);

Following is my PObjectWritable object:
************************************************************************************************************************************************************************

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.Writable;

public class PObjectWritable implements Writable{

 private String url;
 private String media;
 private String title;
 private String status;


 public String getUrl() {
  return url;
 }

 public void setUrl(String url) {
  this.url = url;
 }

 public String getMedia() {
  return media;
 }

 public void setMedia(String media) {
  this.media = media;
 }

 public String getTitle() {
  return title;
 }

 public void setTitle(String title) {
  this.title = title;
 }

 public String getStatus() {
  return status;
 }

 public void setStatus(String status) {
  this.status = status;
 }

 @Override
 public void readFields(DataInput in) throws IOException {
  // TODO Auto-generated method stub
  if(in.readBoolean()){
  url = in.readUTF();
  }else{
  url = null;
  }

  if(in.readBoolean()){
  media = in.readUTF();
  }else{
  media = null;
  }

  if(in.readBoolean()){
  title = in.readUTF();
  }else{
  title = null;
  }

  if(in.readBoolean()){
  status = in.readUTF();
  }else{
  status = null;
  }
 }

 @Override
 public void write(DataOutput out) throws IOException {
  // TODO Auto-generated method stub
  if(url!=null){
  out.writeUTF(url);
  }

  if(media!=null){
  out.writeUTF(media);
  }

  if(title!=null){
  out.writeUTF(title);
  }

  if(status!=null){
  out.writeUTF(status);
  }
 }

 @Override
 public String toString() {
  // TODO Auto-generated method stub
  return url + ", " + media + ", " + title + ", " + status;
 }

}

**************************************************************************************************************************************************************************
Following is the exception at reduce:
java.lang.RuntimeException: problem advancing post rec#0
 at org.apache.hadoop.mapred.Task$ValuesIterator.next(Task.java:831)
 at
org.apache.hadoop.mapred.ReduceTask$ReduceValuesIterator.moveToNext(ReduceTask.java:237)
 at
org.apache.hadoop.mapred.ReduceTask$ReduceValuesIterator.next(ReduceTask.java:233)
 at
org.apache.hadoop.mapred.lib.IdentityReducer.reduce(IdentityReducer.java:39)
 at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:430)
 at org.apache.hadoop.mapred.Child.main(Child.java:155)
Caused by: java.io.EOFException
 at java.io.DataInputStream.readFully(DataInputStream.java:180)
 at java.io.DataInputStream.readUTF(DataInputStream.java:592)
 at java.io.DataInputStream.readUTF(DataInputStream.java:547)
 at com.mapreduce.PObjectWritable.readFields(PObjectWritable.java:59)
 at
org.apache.hadoop.io.serializer.WritableSerialization$WritableDeserializer.deserialize(WritableSerialization.java:67)
 at
org.apache.hadoop.io.serializer.WritableSerialization$WritableDeserializer.deserialize(WritableSerialization.java:40)
 at
org.apache.hadoop.mapred.Task$ValuesIterator.readNextValue(Task.java:888)
 at org.apache.hadoop.mapred.Task$ValuesIterator.next(Task.java:828)
 ... 5 more
wht could cause the exception??
Regards
Raakhi