title: Flink Kafka consumer StreamExecutionEnvironment only?
body: <p>I have pulling scenario,</p>
  
  <p><strong>HTTP -> Kafka -> Flink -> some output</strong> </p>
  
  <p>If im not wrong i can use kafka consumer on stream only  ?</p>
  
  <p>Therefor i need to "block"   the stream in order to sum/count the data im receiving from the HTTP call .</p>
  
  <p>The easiest way to "block" is to add <strong>window</strong>/.</p>
  
  <p>What is the best approach  for this pulling scenario .</p>
  
  <p><strong>UPDATE</strong></p>
  
  <p>I want to prevent from the collector to sum each value </p>
  
  <pre><code>SingleOutputStreamOperator&lt;Tuple2&lt;String, Integer&gt;&gt; t = 
          in.flatMap(new FlatMapFunction&lt;String, Tuple2&lt;String, Integer&gt;&gt;() {
              @Override
              public void flatMap(String s, Collector&lt;Tuple2&lt;String, Integer&gt;&gt; 
              collector) throws Exception {
                  ObjectMapper mapper = new ObjectMapper();
                  JsonNode node = mapper.readTree(s);
                  node.elements().forEachRemaining(v -&gt; {
                      collector.collect(new Tuple2&lt;&gt;(v.textValue(), 1));
                  });
  
              }
          }).keyBy(0).sum(1);
  </code></pre>