title: Apache flink 1.52 Rowtime timestamp is null
body: <p>I am doing some query with the following code:</p>
  
  <pre><code>    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
      DataStream&lt;Row&gt; ds = SourceHelp.builder().env(env).consumer010(MyKafka.builder().build().kafkaWithWaterMark2())
              .rowTypeInfo(MyRowType.builder().build().typeInfo())
              .build().source4();
      //,proctime.proctime,rowtime.rowtime
      String sql1 = "select a,b,max(rowtime)as rowtime from user_device group by a,b";
      DataStream&lt;Row&gt; ds2 = TableHelp.builder().tableEnv(tableEnv).tableName("user_device").fields("a,b,rowtime.rowtime")
              .rowTypeInfo(MyRowType.builder().build().typeInfo13())
              .sql(sql1).in(ds).build().result();
  
      ds2.print();
      // String sql2 = "select a,count(b) as b from user_device2 group by a";
      String sql2 = "select a,count(b) as b,HOP_END(rowtime,INTERVAL '5' SECOND,INTERVAL '30' SECOND) as c from user_device2 group by HOP(rowtime, INTERVAL '5' SECOND, INTERVAL '30' SECOND),a";
      DataStream&lt;Row&gt; ds3 = TableHelp.builder().tableEnv(tableEnv).tableName("user_device2").fields("a,b,rowtime.rowtime")
              .rowTypeInfo(MyRowType.builder().build().typeInfo14())
              .sql(sql2).in(ds2).build().result();
  
      ds3.print();
      env.execute("test");
  </code></pre>
  
  <p>note: For sql1, I use max function with rowtime, it is not working, and following Exception is thrown:</p>
  
  <blockquote>
    <p>Exception in thread "main"
    org.apache.flink.runtime.client.JobExecutionException:
    java.lang.RuntimeException: Rowtime timestamp is null. Please make
    sure that a proper TimestampAssigner is defined and the stream
    environment uses the EventTime time characteristic.   at
    org.apache.flink.runtime.minicluster.MiniCluster.executeJobBlocking(MiniCluster.java:625)
      at
    org.apache.flink.streaming.api.environment.LocalStreamEnvironment.execute(LocalStreamEnvironment.java:123)
      at
    com.aicaigroup.water.WaterTest.testRowtimeWithMoreSqls5(WaterTest.java:158)
      at com.aicaigroup.water.WaterTest.main(WaterTest.java:20) Caused by:
    java.lang.RuntimeException: Rowtime timestamp is null. Please make
    sure that a proper TimestampAssigner is defined and the stream
    environment uses the EventTime time characteristic.   at
    DataStreamSourceConversion$24.processElement(Unknown Source)  at
    org.apache.flink.table.runtime.CRowOutputProcessRunner.processElement(CRowOutputProcessRunner.scala:67)
      at
    org.apache.flink.streaming.api.operators.ProcessOperator.processElement(ProcessOperator.java:66)
      at
    org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:558)
      at
    org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:533)
      at
    org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:513)
      at
    org.apache.flink.streaming.runtime.tasks.OperatorChain$BroadcastingOutputCollector.collect(OperatorChain.java:628)
      at
    org.apache.flink.streaming.runtime.tasks.OperatorChain$BroadcastingOutputCollector.collect(OperatorChain.java:581)
      at
    org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:679)
      at
    org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:657)
      at
    org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:51)
      at com.aicaigroup.TableHelp$1.processElement(TableHelp.java:42)     at
    com.aicaigroup.TableHelp$1.processElement(TableHelp.java:39)  at
    org.apache.flink.streaming.api.operators.ProcessOperator.processElement(ProcessOperator.java:66)
      at
    org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:558)
      at
    org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:533)
      at
    org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:513)
      at
    org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:679)
      at
    org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:657)
      at
    org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:41)
      at
    org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:558)
      at
    org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:533)
      at
    org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:513)
      at
    org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:679)
      at
    org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:657)
      at
    org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:51)
      at
    org.apache.flink.table.runtime.aggregate.GroupAggProcessFunction.processElement(GroupAggProcessFunction.scala:151)
      at
    org.apache.flink.table.runtime.aggregate.GroupAggProcessFunction.processElement(GroupAggProcessFunction.scala:39)
      at
    org.apache.flink.streaming.api.operators.LegacyKeyedProcessOperator.processElement(LegacyKeyedProcessOperator.java:88)
      at
    org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:202)
      at
    org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:104)
      at
    org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:306)
      at org.apache.flink.runtime.taskmanager.Task.run(Task.java:703)     at
    java.lang.Thread.run(Thread.java:748) 2018-09-17 09:51:53.679 [Kafka
    0.10 Fetcher for Source: Custom Source -> Map -> from: (a, b, rowtime) -> select: (a, b, CAST(rowtime) AS rowtime) (2/8)] INFO  o.a.kafka.clients.consumer.internals.AbstractCoordinator  - Discovered
    coordinator 172.16.11.91:9092 (id: 2147483647 rack: null) for group
    test.</p>
  </blockquote>
  
  <p><strong>then I tried to update sql1 like this "select a,b,rowtime from user_device", and it works.
  So how to fix the error? First sql should use group by, and second sql should use rowtime by 
  timeWindow. 3QS</strong></p>