/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.intel.hibench.streambench;

import com.intel.hibench.streambench.common.ConfigLoader;

import java.io.BufferedReader;
import java.io.IOException;
import java.util.Properties;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Syncable;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

/**
 * KafkaSender hold an kafka producer. It gets content from input parameter, generates records and
 * sends records to kafka.
 */
public class KafkaSender {
  String sourcePath;
  Configuration dfsConf;
  KafkaProducer producer;
  StringSerializer serializer = new StringSerializer();

  // offset of file input stream. Currently it's fixed, which means same records will be sent
  // out on very batch.
  long offset;

  // constructor
  public KafkaSender(String sourcePath, long startOffset, ConfigLoader configLoader) {
    String brokerList = configLoader.getProperty("hibench.streamingbench.brokerList");
    String dfsMaster = configLoader.getProperty("hibench.hdfs.master");

    // Details of KafkaProducerConfig could be find from:
    //   http://kafka.apache.org/documentation.html#producerconfigs
    Properties props = new Properties();
    props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
    props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
        "org.apache.kafka.common.serialization.StringSerializer");
    props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
        "org.apache.kafka.common.serialization.StringSerializer");
    props.setProperty(ProducerConfig.ACKS_CONFIG, "1");
    props.getProperty(ProducerConfig.CLIENT_ID_CONFIG, "hibench_data_generator");

    this.producer = new KafkaProducer(props);
    this.sourcePath = sourcePath;
    this.dfsConf = new Configuration();
    dfsConf.set("fs.default.name", dfsMaster);
    this.offset = startOffset;
  }

  // The callback function will be triggered when receive ack from kafka.
  // Print error message if exception exist.
  Callback callback = new Callback() {
    public void onCompletion(RecordMetadata metadata, Exception e) {
      if (e != null)
        e.printStackTrace();
    }
  };

  // send content to Kafka
  public long send (String topic, long totalRecords) {
    long startTime = System.currentTimeMillis();

    long sentRecords = 0L;
    long sentBytes = 0L;
    BufferedReader reader = SourceFileReader.getReader(dfsConf, sourcePath, offset);

    try {
      while (sentRecords < totalRecords) {
        String line = reader.readLine();
        if (line == null) {
          break; // no more data from source files
        }
        String currentTime = Long.toString(System.currentTimeMillis());
        ProducerRecord record = new ProducerRecord(topic, currentTime, line);
        producer.send(record, callback);

        // Key and Value will be serialized twice.
        // 1. in producer.send method
        // 2. explicitly serialize here to count byte size.
        byte[] keyByte = serializer.serialize(topic, currentTime);
        byte[] valueByte = serializer.serialize(topic, line);

        //update counter
        sentRecords ++;
        sentBytes = sentBytes + keyByte.length + valueByte.length;
      }
    } catch (IOException e) {
      System.err.println("Failed read records from Path: " + sourcePath);
      e.printStackTrace();
    }

    // print out useful info
    long endTime = System.currentTimeMillis();
    double timeCost = (double) (endTime - startTime) / 1000;
    double throughput = (double) (sentBytes / timeCost) / 1000000;
    System.out.println("sent " + sentRecords + " records to Kafka topic: " + topic);
    System.out.println("totally sent " + sentBytes + " bytes in " + timeCost + " seconds (throughout: " + throughput + " MB/s)");

    return sentRecords;
  }

  // close kafka producer
  public void close() {
    producer.close();
  }
}

@@ -1407,16 +1407,41 @@
 oducer;%0A
+  CachedData cachedData;%0A
   String
@@ -3107,24 +3107,118 @@
 lRecords) %7B%0A
+    this.cachedData = CachedData.getInstance(sourcePath, dfsConf, offset, (int)totalRecords);%0A
     long sta
@@ -3311,106 +3311,9 @@
 0L;%0A
-    BufferedReader reader = SourceFileReader.getReader(dfsConf, sourcePath, offset);%0A%0A    try %7B%0A  
+%0A
     
@@ -3345,26 +3345,24 @@
 lRecords) %7B%0A
-  
       String
@@ -3373,23 +3373,28 @@
 e = 
-reader.readLine
+cachedData.getRecord
 ();%0A
@@ -3395,26 +3395,24 @@
 rd();%0A      
-  
 if (line == 
@@ -3415,26 +3415,24 @@
  == null) %7B%0A
-  
         brea
@@ -3468,18 +3468,16 @@
 e files%0A
-  
       %7D%0A
@@ -3478,26 +3478,24 @@
     %7D%0A      
-  
 String curre
@@ -3546,18 +3546,16 @@
 lis());%0A
-  
       Pr
@@ -3618,26 +3618,24 @@
 ime, line);%0A
-  
       produc
@@ -3668,18 +3668,16 @@
 %0A%0A      
-  
 // Key a
@@ -3717,18 +3717,16 @@
 .%0A      
-  
 // 1. in
@@ -3747,18 +3747,16 @@
  method%0A
-  
       //
@@ -3807,26 +3807,24 @@
 size.%0A      
-  
 byte%5B%5D keyBy
@@ -3872,26 +3872,24 @@
 ime);%0A      
-  
 byte%5B%5D value
@@ -3927,26 +3927,24 @@
 ic, line);%0A%0A
-  
       //upda
@@ -3956,26 +3956,24 @@
 unter%0A      
-  
 sentRecords 
@@ -3975,15 +3975,12 @@
 ords
- 
 ++;%0A
-  
     
@@ -4044,147 +4044,8 @@
 th;%0A
-      %7D%0A    %7D catch (IOException e) %7B%0A      System.err.println(%22Failed read records from Path: %22 + sourcePath);%0A      e.printStackTrace();%0A
     

Update KafkaSender