[
https://issues.apache.org/jira/browse/FLINK-23177?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17371834#comment-17371834
]
Ingo Bürk commented on FLINK-23177:
-----------------------------------
Thanks for opening this issue. This is indeed the expected behavior.
`StreamTableEnvironment#createTemporaryView(String, DataStream)` uses
`#fromDataStream(DataStream)`, and the JavaDocs states that watermarks are not
propagated by default:
{quote}By default, the stream record's timestamp and watermarks are not
propagated unless explicitly declared via \{@link #fromDataStream(DataStream,
Schema)}.
{quote}
You can solve this by calling `#fromDataStream` yourself, adding the watermark
information there and then using `#createTemporaryView(String, Table)`.
CC [~twalthr] as an FYI
> watermarks generated in dateStream can not flow into table
> ----------------------------------------------------------
>
> Key: FLINK-23177
> URL: https://issues.apache.org/jira/browse/FLINK-23177
> Project: Flink
> Issue Type: Bug
> Components: API / Core
> Affects Versions: 1.13.1
> Environment: flink: 1.13.1
> Reporter: Spongebob
> Priority: Major
>
> I have assigned watermark in dataStream and then use the
> `createTemporaryView` method to build a table that is source from the
> dataStream. Out of my expectation, the watermarks works normally in
> dataStream but the watermarks of the table stay at -9223372036854775808
> forever.
> {code:java}
> def main(args: Array[String]): Unit = {
> val streamEnv = ...
> streamEnv.enableCheckpointing(1000)
> streamEnv.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, 1000))
> val dataStream = ...
> val resultStream = dataStream.map(
> value => {
> val data = value.split(",")
> (data(0), data(1).toInt)
> }
> ).assignTimestampsAndWatermarks(WatermarkStrategy
> .forBoundedOutOfOrderness(Duration.ZERO)
> .withTimestampAssigner(new SerializableTimestampAssigner[(String, Int)]
> {
> override def extractTimestamp(element: (String, Int),
> recordTimestamp: Long): Long = element._2 * 1000
> }))
> .process(new MyProcessFunc)
> // resultStream.print("raw")
> // streamEnv.execute("")
> val streamTableEnv = buildStreamTableEnv(streamEnv,
> EnvironmentSettings.newInstance().inStreamingMode().useBlinkPlanner().build())
> streamTableEnv.createTemporaryView("catalog_test1", resultStream)
> ...
> }
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
