package com.test.groups;
 
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
 
public class ConsumerTest implements Runnable {
    private KafkaStream m_stream;
    private int m_threadNumber;
 
    public ConsumerTest(KafkaStream a_stream, int a_threadNumber) {
        m_threadNumber = a_threadNumber;
        m_stream = a_stream;
    }
 
    public void run() {
        ConsumerIterator<byte[], byte[]> it = m_stream.iterator();
        while (it.hasNext())
            System.out.println("Thread " + m_threadNumber + ": " + new String(it.next().message()));
        System.out.println("Shutting down Thread: " + m_threadNumber);
    }
}

@@ -211,16 +211,80 @@
 Number;%0A
+    private LogWriter logw;%0A    private OutputStreamWriter osw;%0A
  %0A    pu
@@ -341,16 +341,32 @@
 adNumber
+, LogWriter logw
 ) %7B%0A    
@@ -427,24 +427,82 @@
 = a_stream;%0A
+        this.logw = logw;%0A        osw = logw.getWriter();%0A
     %7D%0A %0A    
@@ -617,16 +617,16 @@
 Next())%0A
-
         
@@ -624,24 +624,81 @@
 %0A           
+osw.write(new String(it.next().message()));%0A           //
  System.out.

Update ConsumerTest.java