Avro decoding gives java.io.EOFException
<p>I use Apache avro schema with Kafka 0.0.8V. I Use same schema at producer/consumer ends. There is <strong>NO ANY Changes</strong> in the schema. But i get some exception at the consumer, when i try to consume the messages.
Why i get this error?</p>

<p><strong>Producer</strong></p>

<pre><code>public void sendFile(String topic, GenericRecord payload, Schema schema) throws CoreException, IOException {
    BinaryEncoder encoder = null;
    ByteArrayOutputStream out = null;
    try {
        DatumWriter&lt;GenericRecord&gt; writer = new SpecificDatumWriter&lt;GenericRecord&gt;(schema);
        out = new ByteArrayOutputStream();
        encoder = EncoderFactory.get().binaryEncoder(out, null);
        writer.write(payload, encoder);
        encoder.flush();

        byte[] serializedBytes = out.toByteArray();

        KeyedMessage&lt;String, byte[]&gt; message = new KeyedMessage&lt;String, byte[]&gt;(topic, serializedBytes);

            producer.send(message);
        }
</code></pre>

<p><strong>Consumer</strong></p>

<pre><code>public void run() {
        try {
            ConsumerIterator&lt;byte[], byte[]&gt; itr = stream.iterator();
            while (itr.hasNext()) {

                byte[] data = itr.next().message();

                Schema schema = new Schema.Parser()
                        .parse(new File("/Users/xx/avro_schemas/file.avsc"));

                DatumReader&lt;GenericRecord&gt; reader = new GenericDatumReader&lt;GenericRecord&gt;(schema);
                Decoder decoder = DecoderFactory.get().binaryDecoder(data, null);

                GenericRecord payload = reader.read(null, decoder);
                System.out.println("Message received --: " + payload);
</code></pre>

<p>But I get  following exception when the reader try to read message from the decoder.;</p>

<pre><code>java.io.EOFException
    at org.apache.avro.io.BinaryDecoder.ensureBounds(BinaryDecoder.java:473)
    at org.apache.avro.io.BinaryDecoder.readInt(BinaryDecoder.java:128)
    at org.apache.avro.io.BinaryDecoder.readString(BinaryDecoder.java:259)
    at org.apache.avro.io.ResolvingDecoder.readString(ResolvingDecoder.java:201)
    at org.apache.avro.generic.GenericDatumReader.readString(GenericDatumReader.java:363)
    at org.apache.avro.generic.GenericDatumReader.readString(GenericDatumReader.java:355)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:157)
    at org.apache.avro.generic.GenericDatumReader.readField(GenericDatumReader.java:193)
    at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:183)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:151)
    at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:142)
    at com.xx.KafkaMessageListenerThread.run(KafkaMessageListenerThread.java:55)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
</code></pre>

<p><strong>Consumer properties</strong></p>

<pre><code>enable.auto.commit=true
auto.commit.interval.ms=101
session.timeout.ms=7000
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
zookeeper.connect=zookeeper.xx.com\:2181
heartbeat.interval.ms=1000
auto.offset.reset=smallest
serializer.class=kafka.serializer.DefaultEncoder
bootstrap.servers=kafka.xx.com\:9092
group.id=test
consumer.timeout.ms=-1
fetch.min.bytes=1
receive.buffer.bytes=262144
</code></pre>