title: Flink on Yarn, parallel source with Kafka
body: <p>I am trying to have parallelism with my Kafka source within my Flink job, but I failed so far.</p>
  
  <p>I set 4 partitions to my Kafka producer :</p>
  
  <pre><code>$ ./bin/kafka-topics.sh --describe --zookeeper X.X.X.X:2181 --topic mytopic
  Topic:mytopic   PartitionCount:4    ReplicationFactor:1 Configs:
      Topic: mytopic  Partition: 0    Leader: 0   Replicas: 0 Isr: 0
      Topic: mytopic  Partition: 1    Leader: 0   Replicas: 0 Isr: 0
      Topic: mytopic  Partition: 2    Leader: 0   Replicas: 0 Isr: 0
      Topic: mytopic  Partition: 3    Leader: 0   Replicas: 0 Isr: 0
  </code></pre>
  
  <p>My scala code is as follow :</p>
  
  <pre><code>    val env = StreamExecutionEnvironment.getExecutionEnvironment
          env.setParallelism(4)
          env.getConfig.setGlobalJobParameters(params)
  
      // **** Kafka CONNECTION ****
      val properties = new Properties();
      properties.setProperty("bootstrap.servers", params.get("server"));
      properties.setProperty("group.id", "test");
  
  
      // **** Get KAFKA source ****
     val stream: DataStream[String] = env.addSource(new FlinkKafkaConsumer010[String](params.get("topic"), new SimpleStringSchema(), properties))
  </code></pre>
  
  <p>I run my job on YARN :</p>
  
  <pre><code>$ ./bin/flink run -m yarn-cluster -yn 4 -yjm 8192 -ynm test -ys 1 -ytm 8192 myjar.jar --server X.X.X.X:9092 --topic mytopic
  </code></pre>
  
  <p>I tried a bunch of things, but my source is not parallelized :</p>
  
  <p><a href="https://i.stack.imgur.com/EvFZU.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/EvFZU.png" alt="Kafka source on Flink WEB UI"></a></p>
  
  <p>Having several Kafka partitions and at least as much slot / Task Managers should do it, right?</p>