title: Sending Data to Kafka Producer
body: <p>i am trying to read the 100k file and send it to kafka topic. Here is my Kafka Code Which sends data to Kafka-console-consumer. When i am sending data i am receiving the data like this </p>
  
  <pre><code>java.util.stream.ReferencePipeline$Head@e9e54c2
  </code></pre>
  
  <p>Here is the sample single record data what i am sending:</p>
  
  <pre><code>173|172686|548247079|837113012|0x548247079f|7|173|172686a|0|173|2059 22143|0|173|1|173|172686|||0|||7|0||7|||7|172686|allowAllServices|?20161231:22143|548247079||0|173||172686|5:2266490827:DCCInter;20160905152146;2784
  </code></pre>
  
  <p>Any suggestion to get the data which i had showned in above...Thanks</p>
  
  <p>Code:</p>
  
  <pre><code>import java.io.BufferedReader;
  import java.io.FileNotFoundException;
  import java.io.FileReader;
  import java.io.IOException;
  import java.nio.file.Files;
  import java.nio.file.Paths;
  import java.util.Properties;
  import java.util.Properties;
  import java.util.concurrent.ExecutionException;
  import java.util.stream.Stream;
  
  import kafka.javaapi.producer.Producer;
  import kafka.producer.KeyedMessage;
  import kafka.producer.ProducerConfig;
  
  @SuppressWarnings("unused")
  public class HundredKRecords { 
     private static String sCurrentLine;
     public static void main(String args[]) throws InterruptedException, ExecutionException{ 
         String fileName = "/Users/sreeeedupuganti/Downloads/octfwriter.txt";
  
         //read file into stream, try-with-resources
         try (Stream&lt;String&gt; stream = Files.lines(Paths.get(fileName))) {
             stream.forEach(System.out::println);
             kafka(stream.toString());
         } catch (IOException e) {
             e.printStackTrace();
         }
         }
  
     public static void kafka(String stream)  {
         Properties props = new Properties();
         props.put("metadata.broker.list", "localhost:9092");
         props.put("serializer.class", "kafka.serializer.StringEncoder");
         props.put("partitioner.class","kafka.producer.DefaultPartitioner");
         props.put("request.required.acks", "1");
         ProducerConfig config = new ProducerConfig(props);
         Producer&lt;String, String&gt; producer = new Producer&lt;String, String&gt;(config);
         producer.send(new KeyedMessage&lt;String, String&gt;("test",stream));
         producer.close();
     }
  }
  </code></pre>