scala - 重置与 Url/Port 的连接 - 如果在过去 N 分钟内未收到任何数据

Question

我有一个连接到 Url/Port 的 flink 应用程序，我看到它允许检查连接是否仍然打开的重启策略。

我的查询是...如果连接已打开，但在过去的“N”分钟内未收到任何数据，我想重置连接

目前使用基本的 flink 教程设置连接

    // set up the streaming execution environment
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val data_stream = env.socketTextStream(url, port, socket_stream_deliminator, socket_connection_retries)
                         .map(x => printInput(x))
                          .writeToSocket(url, port, new SimpleStringSchema())

    // execute program
    env.execute("Flink Streaming Scala API Skeleton")

是否有函数调用或一些包含的映射来调用以检查连接是否在过去的“N”分钟内发送了数据，如果没有重新连接。我该怎么做

标签： scalaapache-flinkflink-streaming