[FLINK-2536][streaming]add a re-connect for socket sink #1030
Conversation
|
Add tests for retry 10 times and 0. |
| dataOutputStream.write(msg); | ||
| success = true; | ||
|
|
||
| }catch(Exception ee){ |
There was a problem hiding this comment.
We usually add whitespaces between keywords.
There was a problem hiding this comment.
Do we really want to catch all exceptions here? Maybe IOException is enough?
There was a problem hiding this comment.
Yes, it depends on you.
There was a problem hiding this comment.
Well, it actually depends on the exceptions for which we want to restart. If I'm not mistaken, then new Socket() and dataOutputStream.write only throw IOExceptions. Thus, we should change it.
|
Hi @HuangWHWHW, I had some minor comments. Could we also add a test case where we test that the |
|
" Could we also add a test case where we test that the SocketClientSink can reconnect against a newly opened socket after it has been closed? This would be great." Good idea! |
|
@tillrohrmann BTW:Why does not the CI rerun?? |
| server.close(); | ||
|
|
||
| if (error.get() != null) { | ||
| Throwable t = error.get(); |
There was a problem hiding this comment.
I think it's good to set the error to null again. Otherwise all subsequent tests will fail with the same error message which is, however, completely unrelated.
|
Hi @HuangWHWHW, I had some comments concerning the test cases. |
|
Hi, |
| } catch(IOException ee) { | ||
| Log.error("Reconnect to socket server and send message failed. Caused by " + | ||
| ee.toString() + ". Retry time(s):" + retries); | ||
| synchronized (this) { |
There was a problem hiding this comment.
Locking on this is usually not recommended because outside code can influence your behaviour here. Thus, use a private lock field here.
|
@tillrohrmann |
|
@tillrohrmann |
|
@tillrohrmann |
| byte[] msg = schema.serialize(value); | ||
| try { | ||
| dataOutputStream.write(msg); | ||
| } catch (IOException e) { | ||
| throw new RuntimeException("Cannot send message " + value.toString() + | ||
| " to socket server at " + hostName + ":" + port, e); | ||
| LOG.error("Cannot send message " + value.toString() + |
There was a problem hiding this comment.
You can log the exception simpler like this:
LOG.error("Cannot send message " + value + " to socket server at " + hostName + ":" + port + ". Trying to reconnect.", e);
`´`
|
@StephanEwen |
|
@StephanEwen |
HuangWHWHW
force-pushed
the
FLINK-2536
branch
from
4fe83c4
to
2970b13
Compare
|
@StephanEwen |
|
Sorry, @HuangWHWHW, currently we're really busy. I'll try to review your PR once I find a free minute. |
|
I think this looks good, except for the comment with the final variable for the lock. One more comment: When concatenating strings, avoid constructs like |
|
@tillrohrmann |
|
No worries. We are simply overloaded right now. Many hard features under development, and many pull requests being opened. |
HuangWHWHW
force-pushed
the
FLINK-2536
branch
from
2970b13
to
93971ea
Compare
HuangWHWHW
force-pushed
the
FLINK-2536
branch
from
93971ea
to
2ba7184
Compare
|
I removed the toString() method and change the lock to |
|
Sorry for misunderstading. |
|
Hi, very sorry for bothering again. |
|
Looks good now, will merge this... |
|
I reworked this quite heavily during merging. There were a lot of issues that were against good Java style:
You can have a look at the code after my fixes, to see these issues in context. I would suggest to get a Java book (like "Effective Java", that is a good one) and take this as a guideline for future work. This pull request took more than 70 comments and still needed quite some rework (not for Flink-specific issues, but all of it general Java style/efficiency/correctness). I am afraid we cannot do that for every pull request, it would be completely overwhelming... |
|
@StephanEwen |
|
@HuangWHWHW No problem, we all learn all the time. |
add a re-connect in function invoke() when it throws exception.