[FLINK-21814] Add host and port information in exception when CollectSink can't connect to the client

• Type: Improvement
• Status: Closed
• Priority: Minor
• Resolution: Fixed
• Affects Version/s: 1.12.2
• Fix Version/s: 1.13.0
• Component/s: API / DataStream, Table SQL / Ecosystem
• Labels: pull-request-available
• Assignee: Peihui He
• Reporter: Peihui He
• Created: 16/Mar/21 08:08
• Updated: 22/Mar/21 04:31
• Resolved: 22/Mar/21 04:31

Description

org.apache.flink.streaming.experimental.CollectSink initialization need host and port. When Network is unavailable, i will get the error:

java.io.IOException: Cannot connect to the client to send back the stream
    at org.apache.flink.streaming.experimental.CollectSink.open(CollectSink.java:86)
    at org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:34)
    at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:102)
    at org.apache.flink.streaming.api.operators.StreamSink.open(StreamSink.java:46)
    at org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:426)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$2(StreamTask.java:535)
    at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:93)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:525)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:565)
    at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:755)
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:570)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.net.ConnectException: Connection refused (Connection refused)
    at java.net.PlainSocketImpl.socketConnect(Native Method)
    at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:476)
    at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:218)
    at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:200)
    at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:394)
    at java.net.Socket.connect(Socket.java:606)
    at java.net.Socket.connect(Socket.java:555)
    at java.net.Socket.<init>(Socket.java:451)
    at java.net.Socket.<init>(Socket.java:261)
    at org.apache.flink.streaming.experimental.CollectSink.open(CollectSink.java:82)
    ... 11 more

 
However, i couldn't know the host and port.
 
Hope the error info could includ the host and port.

Issue Links

• links to: GitHub Pull Request #15297