title: Flink streaming - latency and throughput detection
body: <p>I am trying to run the Flink streaming job. I want to determine the throughput and latency for the streaming process. i have started the Kafka broker server and have incoming messages from kafka.How do i count messages per second (Throughput)?
  (Like rdd.count. Is there any similar method to get the count of incoming messages)</p>
  
  <p>(Complete scenerio : I have sent the message through  Producer as a Json Object. I am adding some information like name as string and also System.currentTimeMills in the Json object.
  During streaming , how do i obtain the sent json object through messageStream(DataStream)?)</p>
  
  <p>Thanks in advance.</p>
  
  <p>CODE :</p>
  
  <p>/**
   * Read Strings from Kafka and print them to standard out.
  */</p>
  
  <pre><code>public static void main(String[] args) throws Exception {
      System.setProperty("hadoop.home.dir", "c:/winutils/");
      // parse input argum    ents
      final ParameterTool parameterTool = ParameterTool.fromArgs(args);
  
      if(parameterTool.getNumberOfParameters() &lt; 4) {
          System.out.println("Missing parameters!\nUsage: Kafka --topic &lt;topic&gt; " +
                  "--bootstrap.servers &lt;kafka brokers&gt; --zookeeper.connect &lt;zk quorum&gt; --group.id &lt;some id&gt;");
          return;
      }
  
      StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
      env.getConfig().disableSysoutLogging();
      env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000));
      env.enableCheckpointing(5000); // create a checkpoint every 5 seconds
      env.getConfig().setGlobalJobParameters(parameterTool); // make parameters available in the web interface
  
      DataStream&lt;String&gt; messageStream = env
              .addSource(new FlinkKafkaConsumer010&lt;&gt;(
                      parameterTool.getRequired("topic"),
                      new SimpleStringSchema(),
                      parameterTool.getProperties()));
  
  
      messageStream.print();
  
      env.execute();
  }
  </code></pre>