Fix invocation retry logic (#413)

In the current code-base, we could retry invocations in the following
cases.

- In the `_do_invoke` function, when a target connection for the
invocation cannot be determined. This may happen in any other thread
than the reactor thread. But, in this case, a message is not sent
from the client (even, a message is not even put into one of the write
queue of a connection). Hence, the `sent_connection` of such invocations
is `None`.

- When an exceptional response is received from the server, which
always happen in the reactor thread.

- When a connection is closed, and we detect it in the clean resources
timer, which also always happens in the reactor thread.

Under this scenario, it was not possible for an invocation to
be retried multiple times, once for the exceptional response and one
for the connection close case, like the Java client.

But, there were a couple of problematic scenarios.

- When we schedule an invocation for retry, we were not resetting
the `sent_connection`. So, if the `sent_connection` is closed in
the meanwhile, it was possible to schedule
`invocation_retry_pause / clean_resources_timer_period` retries.
Note that, an invocation retry which will happen in the future
does not care about its last `sent_connection`.

- If we schedule an invocation for a retry and somehow, its result
is set in the meanwhile, while we are waiting for the
`invocation_retry_pause`, we will retry the invocation anyway. Now,
we check whether or not an invocation is done, before retrying it.

To tackle 1, we reset the `sent_connection` before scheduling the retry.

To tackle 2, we check the invocation's future before retrying.

Author: mdumandag

hazelcast/invocation.py

@@ -270,9 +270,14 @@ def _notify_error(self, invocation, error):
             self._complete_with_error(invocation, error)
             return
 
-        invoke_func = functools.partial(self._do_invoke, invocation)
+        invocation.sent_connection = None
+        invoke_func = functools.partial(self._retry_if_not_done, invocation)
         self._reactor.add_timer(self._invocation_retry_pause, invoke_func)
 
+    def _retry_if_not_done(self, invocation):
+        if not invocation.future.done():
+            self._do_invoke(invocation)
+
     def _should_retry(self, invocation, error):
         if invocation.connection and isinstance(error, (IOError, TargetDisconnectedError)):
             return False