[STORM-3054] Add Topology level configuration socket timeout for DRPC Invocation Client #2651
Conversation
| } | ||
|
|
||
| private String getServer(String host, int port) { | ||
| return host + port; |
There was a problem hiding this comment.
Better to keep it as List since bad case could be happen (when host ends with number).
There was a problem hiding this comment.
Reverting to List.
| @@ -103,21 +85,32 @@ public void execute(Tuple input) { | |||
| LOG.error("Failed to return results to DRPC server", tex); | |||
| _collector.fail(input); | |||
| } | |||
| reconnectClient((DRPCInvocationsClient) client); | |||
| client = getDRPCClient(host, port); | |||
There was a problem hiding this comment.
Now it always reuse the existing client even TException is being raised from the client. I guess we need to handle the case.
There was a problem hiding this comment.
This is same as before, if we could not make connection. or reconnect failed in previous case. The whole reason to create new client from scratch in getDCPClient is to avoid using reconnect with security enabled. The stale connection fails to respond to security challenge.
There was a problem hiding this comment.
@kishorvpatil
I meant we don't invalidate broken client from _clients so it will always pick same client, instead of rebuilding new client.
|
@kishorvpatil any update on the comments from @HeartSaVioR ? |
|
@HeartSaVioR @revans2 sorry for delay in addressing the review comments. |
| }; | ||
| if (!_clients.containsKey(server)) { | ||
| try { | ||
| DRPCInvocationsClient oldClient = _clients.put(server, new DRPCInvocationsClient(_conf, host, port)); |
There was a problem hiding this comment.
Now it loses the cache functionality. Instead of this approach, can we invalidate cache when we find that the client is broken?
This patch fixes following this:
_clientsmap key inReturnResultsReturnResultsdebug log entries.