title: Migrating from FlinkKafkaConsumer to KafkaSource, no windows executed
body: <p>I am a kafka and flink beginner.
  I have implemented <code>FlinkKafkaConsumer</code> to consume messages from a kafka-topic. The only custom setting other than &quot;group&quot; and &quot;topic&quot; is <code>(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, &quot;earliest&quot;)</code> to enable re-reading the same messages several times. It works out of the box for consuming and logic.
  Now <code>FlinkKafkaConsumer</code> is deprecated, and i wanted to change to the successor <code>KafkaSource</code>.</p>
  <p>Initializing <code>KafkaSource</code> with the same parameters as i do <code>FlinkKafkaConsumer</code> produces a read of the topic as expected, i can verify this by printing the stream. De-serialization and timestamps seem to work fine. However execution of windows are not done, and as such no results are produced.</p>
  <p>I assume some default setting(s) in <code>KafkaSource</code> are different from that of <code>FlinkKafkaConsumer</code>, but i have no idea what they might be.</p>
  <p>KafkaSource - <strong>Not working</strong></p>
  <pre><code> KafkaSource&lt;TestData&gt; source = KafkaSource.&lt;TestData&gt;builder()
                  .setBootstrapServers(propertiesForKafka.getProperty(&quot;bootstrap.servers&quot;))
                  .setTopics(TOPIC)
                  .setDeserializer(new CustomDeserializer())
                  .setGroupId(GROUP_ID)
                  .setStartingOffsets(OffsetsInitializer.earliest())
                  .build();
  
          DataStream&lt;TestData&gt; testDataStreamSource = env.fromSource(
                  source,
                  WatermarkStrategy.
                          &lt;TestData&gt;noWatermarks(),
                  &quot;Kafka Source&quot;
          );
  </code></pre>
  <p>Kafka consumer - <strong>Working</strong> (Properties contains <code>group.id</code>,<code>bootstrap.servers</code> and <code>zookeeper.connect</code>)</p>
  <pre><code>    propertiesForKafka.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, &quot;earliest&quot;);
      FlinkKafkaConsumer&lt;TestData&gt; flinkKafkaConsumer = new  FlinkKafkaConsumer(TOPIC, new CustomDeserializer(), propertiesForKafka);
      DataStreamSource&lt;TestData&gt; testDataStreamSource = env.addSource(flinkKafkaConsumer)
  </code></pre>
  <p>Both streams use the same pipeline that looks like this</p>
  <pre><code>testDataStreamSource
             .assignTimestampsAndWatermarks(WatermarkStrategy.
                          &lt;TestData&gt;forMonotonousTimestamps().
                          withTimestampAssigner((event, timestamp) -&gt; event.getTimestamp()))
              .keyBy(TestData::getKey)
              .window(SlidingEventTimeWindows.of(Time.hours(3), Time.hours(1)))
              .process(new ProcessWindowFunction&lt;TestData, TestDataOutput, String, TimeWindow&gt;() {
                      @Override
                      public void process(
              ....
              });
  </code></pre>
  <p>Things tried</p>
  <ul>
  <li>I've tried to experiment with setting committing of offsets, but it
  has not improved the situation.</li>
  <li>Setting timestamps already in the source.</li>
  </ul>