title: flink kafka consumer groupId not working
body: <p>I am using kafka with flink. 
  In a simple program, I used flinks FlinkKafkaConsumer09, assigned the group id to it.</p>
  
  <p>According to Kafka's behavior, when I run 2 consumers on the same topic with same group.Id, it should work like a message queue. I think it's supposed to work like:
  If 2 messages sent to Kafka, each or one of the flink program would process the 2 messages totally twice(let's say 2 lines of output in total).</p>
  
  <p>But the actual result is that, each program would receive 2 pieces of the messages.</p>
  
  <p>I have tried to use consumer client that came with the kafka server download. It worked in the documented way(2 messages processed).<br>
  I tried to use 2 kafka consumers in the same Main function of a flink programe. 4 messages processed totally.<br>
  I also tried to run 2 instances of flink, and assigned each one of them the same program of kafka consumer. 4 messages.  </p>
  
  <p>Any ideas? 
  This is the output I expect:  </p>
  
  <pre><code>1&gt; Kafka and Flink2 says: element-65  
  2&gt; Kafka and Flink1 says: element-66 
  </code></pre>
  
  <p>Here's the wrong output i always get:  </p>
  
  <pre><code>1&gt; Kafka and Flink2 says: element-65  
  1&gt; Kafka and Flink1 says: element-65  
  2&gt; Kafka and Flink2 says: element-66  
  2&gt; Kafka and Flink1 says: element-66 
  </code></pre>
  
  <p>And here is the segment of code:</p>
  
  <pre><code>public static void main(String[] args) throws Exception {
  
      StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  
      ParameterTool parameterTool = ParameterTool.fromArgs(args);
  
      DataStream&lt;String&gt; messageStream = env.addSource(new FlinkKafkaConsumer09&lt;&gt;(parameterTool.getRequired("topic"), new SimpleStringSchema(), parameterTool.getProperties()));
  
      messageStream.rebalance().map(new MapFunction&lt;String, String&gt;() {
          private static final long serialVersionUID = -6867736771747690202L;
  
          @Override
          public String map(String value) throws Exception {
              return "Kafka and Flink1 says: " + value;
          }
      }).print();
  
  
      env.execute();
  }
  </code></pre>
  
  <p>I have tried to run it twice and also in the other way:
  create 2 datastreams and env.execute() for each one in the Main function.</p>