RichParallelSourceFunction中的水印

我正在实现一个SourceFunction，它从数据库中读取数据。如果停止或压碎（即保存点和检查点），数据只能处理一次，则应该能够恢复该作业。

到目前为止我所拥有的：

@SerialVersionUID(1L)
class JDBCSource(private val waitTimeMs: Long) extends
RichParallelSourceFunction[Event] with StoppableFunction with LazyLogging{

@transient var client: PostGreClient = _
@volatile var isRunning: Boolean = true
val DEFAULT_WAIT_TIME_MS = 1000

def this(clientConfig: Serializable) =
    this(clientConfig, DEFAULT_WAIT_TIME_MS)

override def stop(): Unit = {
    this.isRunning = false
}

override def open(parameters: Configuration): Unit = {
    super.open(parameters)
    client = new JDBCClient
}

override def run(ctx: SourceFunction.SourceContext[Event]): Unit = {

    while (isRunning){
       val statement = client.getConnection.createStatement()
       val resultSet = statement.executeQuery("SELECT name, timestamp FROM MYTABLE")

        while (resultSet.next()) {
            val event: String = resultSet.getString("name")
            val timestamp: Long = resultSet.getLong("timestamp")

            ctx.collectWithTimestamp(new Event(name, timestamp), timestamp)

        }
    }
}

override def cancel(): Unit = {
    isRunning = false
}

}
如何确保只获取尚未处理的数据库行？我假设ctx变量会有一些关于当前水印的信息，以便我可以将我的查询更改为：

select name, timestamp from myTable where timestamp > ctx.getCurrentWaterMark