title: Apache Flink and Apache Pulsar
body: <p>I am using Flink to read data from Apache Pulsar.
  I have a partitioned topic in pulsar with 8 partitions.
  I produced 1000 messages in this topic, distributed across the 8 partitions.
  I have 8 cores in my laptop, so I have 8 sub-tasks (by default parallelism = # of cores).
  I opened the Flink-UI after executing the code from Eclipse, I found that some sub-tasks are not receiving any records (idle).
  I am expecting that all the 8 sub-tasks will be utilized (I am expecting that each sub-task will be mapped to one partition in my topic).</p>
  
  <p>After restarting the job, I found that some times 3 sub-takes are utilized and some times 4 tasks are utilized while the remaining sub-tasks kept idle.</p>
  
  <p>please your support to clarify this scenario.</p>
  
  <p>Also how can I know that there is a shuffle between sub-takes or not?</p>
  
  <p>My Code:</p>
  
  <pre><code>ConsumerConfigurationData&lt;String&gt; consumerConfigurationData = new ConsumerConfigurationData&lt;&gt;();
  
  Set&lt;String&gt; topicsSet = new HashSet&lt;&gt;();
  topicsSet.add("flink-08");
  
  consumerConfigurationData.setTopicNames(topicsSet);
  consumerConfigurationData.setSubscriptionName("my-sub0111");
  consumerConfigurationData.setSubscriptionType(SubscriptionType.Key_Shared);
  consumerConfigurationData.setConsumerName("consumer-01");
  consumerConfigurationData.setSubscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
  
  PulsarSourceBuilder&lt;String&gt; builder = PulsarSourceBuilder.builder(new SimpleStringSchema()).pulsarAllConsumerConf(consumerConfigurationData).serviceUrl("pulsar://localhost:6650");
  
  SourceFunction&lt;String&gt; src = builder.build();
  DataStream&lt;String&gt; stream = env.addSource(src);
  
  stream.print(" &gt;&gt;&gt; ");
  </code></pre>