title: Flink Kafka - how to make App run in Parallel?
body: <p>I am creating a app in Flink to</p>
  
  <ol>
  <li>Read Messages from a topic</li>
  <li>Do some simple process on it</li>
  <li>Write Result to a different topic</li>
  </ol>
  
  <p><strong>My code does work</strong>, however it <strong><em>does not run in parallel</em></strong><br>
  How do I do that?<br>
  It seems my code runs only on one thread/block?  </p>
  
  <p>On the Flink Web Dashboard:</p>
  
  <ul>
  <li>App goes to running status</li>
  <li>But, there is only one block shown in the overview subtasks</li>
  <li>And Bytes Received / Sent, Records Received / Sent is always zero ( no Update )</li>
  </ul>
  
  <p>Here is my code, please assist me in learning how to split my app to be able to run in parallel, and am I writing the app correctly?</p>
  
  <pre><code>public class SimpleApp {
  
      public static void main(String[] args) throws Exception {
  
          // create execution environment INPUT
          StreamExecutionEnvironment env_in  =    
                   StreamExecutionEnvironment.getExecutionEnvironment();
          // event time characteristic
          env_in.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  
          // production Ready (Does NOT Work if greater than 1)
          env_in.setParallelism(Integer.parseInt(args[0].toString()));
  
          // configure kafka consumer
          Properties properties = new Properties();
          properties.setProperty("zookeeper.connect", "localhost:2181");
          properties.setProperty("bootstrap.servers", "localhost:9092");
          properties.setProperty("auto.offset.reset", "earliest");
  
          // create a kafka consumer
          final DataStream&lt;String&gt; consumer = env_in
                  .addSource(new FlinkKafkaConsumer09&lt;&gt;("test", new   
                              SimpleStringSchema(), properties));
  
          // filter data
          SingleOutputStreamOperator&lt;String&gt; result = consumer.filter(new  
              FilterFunction&lt;String&gt;(){
              @Override
              public boolean filter(String s) throws Exception {
                  return s.substring(0, 2).contentEquals("PS");
              }
          });
  
          // Process Data
          // Transform String Records to JSON Objects
          SingleOutputStreamOperator&lt;JSONObject&gt; data = result.map(new 
                  MapFunction&lt;String, JSONObject&gt;()
          {
              @Override
              public JSONObject map(String value) throws Exception
              {
                  JSONObject jsnobj = new JSONObject();
  
                  if(value.substring(0, 2).contentEquals("PS"))
                  {
                      // 1. Raw Data
                      jsnobj.put("Raw_Data", value.substring(0, value.length()-6));
  
                      // 2. Comment
                      int first_index_comment = value.indexOf("$");
                      int last_index_comment  = value.lastIndexOf("$") + 1;
                      //   - set comment
                      String comment          =  
                      value.substring(first_index_comment, last_index_comment);
                      comment = comment.substring(0, comment.length()-6);
                      jsnobj.put("Comment", comment);
                  }
                  else {
                      jsnobj.put("INVALID", value);
                  }
  
                  return jsnobj;
              }
          });
  
          // Write JSON to Kafka Topic
          data.addSink(new FlinkKafkaProducer09&lt;JSONObject&gt;("localhost:9092",
                  "FilteredData",
                  new SimpleJsonSchema()));
  
          env_in.execute();
      }
  }
  </code></pre>
  
  <p>My code does work, but it seems to run only on a single thread 
  ( One block shown ) in web interface ( No passing of data, hence the bytes sent / received are not updated ).  </p>
  
  <p>How do I make it run in parallel ?</p>