cluster shutdown should handle waiting requests quit and shutdown client properly. (#65)

Cleanup noisy near-cache logs

Author: asimarslan

hazelcast/connection.py

@@ -191,6 +191,8 @@ def start(self):
         Starts sending periodic HeartBeat operations.
         """
         def _heartbeat():
+            if not self._client.lifecycle.is_live:
+                return
             self._heartbeat()
             self._heartbeat_timer = self._client.reactor.add_timer(self._heartbeat_interval, _heartbeat)
 

hazelcast/exception.py

@@ -27,6 +27,13 @@ class AuthenticationError(HazelcastError):
     pass
 
 
+class HazelcastClientNotActiveException(ValueError):
+    """
+    A ValueError which is raised when Hazelcast client is not active during an invocation.
+    """
+    pass
+
+
 @retryable
 class HazelcastInstanceNotActiveError(HazelcastError):
     """

hazelcast/invocation.py

@@ -4,7 +4,7 @@
 from Queue import Queue
 import functools
 from hazelcast.exception import create_exception, HazelcastInstanceNotActiveError, is_retryable_error, TimeoutError, \
-    AuthenticationError, TargetDisconnectedError
+    AuthenticationError, TargetDisconnectedError, HazelcastClientNotActiveException
 from hazelcast.future import Future
 from hazelcast.lifecycle import LIFECYCLE_STATE_CONNECTED
 from hazelcast.protocol.client_message import LISTENER_FLAG
@@ -154,7 +154,6 @@ def invoke_smart(self, invocation, ignore_heartbeat=False):
         else:  # send to random address
             addr = self._client.load_balancer.next_address()
             self._send_to_address(invocation, addr)
-
         return invocation.future
 
     def invoke_non_smart(self, invocation, ignore_heartbeat=False):
@@ -251,9 +250,11 @@ def _handle_event(self, invocation, message):
             self.logger.warn("Error handling event %s", message, exc_info=True)
 
     def _handle_exception(self, invocation, error, traceback=None):
+        if not self._client.lifecycle.is_live:
+            invocation.set_exception(HazelcastClientNotActiveException(error.message), traceback)
+            return
         if self.logger.isEnabledFor(logging.DEBUG):
-            self.logger.debug("Got exception for request %s: %s: %s", invocation.request,
-                              type(error).__name__, error)
+            self.logger.debug("Got exception for request %s: %s: %s", invocation.request, type(error).__name__, error)
         if isinstance(error, (AuthenticationError, IOError, HazelcastInstanceNotActiveError)):
             if self._try_retry(invocation):
                 return
@@ -272,7 +273,6 @@ def _try_retry(self, invocation):
             return False
 
         invoke_func = functools.partial(self.invoke, invocation)
-        self.logger.debug("Rescheduling request %s to be retried in %s seconds", invocation.request,
-                          RETRY_WAIT_TIME_IN_SECONDS)
+        self.logger.debug("Rescheduling request %s to be retried in %s seconds", invocation.request, RETRY_WAIT_TIME_IN_SECONDS)
         self._client.reactor.add_timer(RETRY_WAIT_TIME_IN_SECONDS, invoke_func)
         return True

hazelcast/near_cache.py

@@ -159,13 +159,12 @@ def _do_eviction_if_required(self):
 
         if len(new_eviction_samples) == len(new_eviction_samples_cleaned):  # did any item expired or do we need to evict
             try:
-                self.logger.debug("Evicting key:{}".format(self._eviction_candidates[0].key))
                 self.__delitem__(self._eviction_candidates[0].key)
                 self._evicted_count += 1
                 del self._eviction_candidates[0]
             except KeyError:
                 # key may be evicted previously so just ignore it
-                self.logger.debug("Trying to evict but key:{} already expired.".format(self._eviction_candidates[0].key))
+                pass
 
     def _find_new_random_samples(self):
         records = self.values()  # has random order because of dict hash
@@ -206,12 +205,11 @@ def get_statistics(self):
 
     def _clean_expired_record(self, key):
         try:
-            self.logger.debug("Expiring key:{}".format(key))
             self.__delitem__(key)
             self._expired_count += 1
         except KeyError:
             # key may be evicted previously so just ignore it
-            self.logger.debug("Trying to expire but key:{} already expired.".format(key))
+            pass
 
     def __repr__(self):
         return "NearCache[len:{}, evicted:{}]".format(self.__len__(), self._evicted_count)

hazelcast/partition.py

@@ -51,7 +51,7 @@ def get_partition_owner(self, partition_id):
         """
         if partition_id not in self.partitions:
             self._do_refresh()
-        return self.partitions[partition_id]
+        return self.partitions.get(partition_id, None)
 
     def get_partition_id(self, key):
         """
@@ -62,6 +62,8 @@ def get_partition_id(self, key):
         """
         data = self._client.serialization_service.to_data(key)
         count = self.get_partition_count()
+        if count <= 0:
+            return 0
         return hash_to_index(data.get_partition_hash(), count)
 
     def get_partition_count(self):
@@ -76,7 +78,7 @@ def get_partition_count(self):
 
     def _get_partition_count_blocking(self):
         event = threading.Event()
-        while not self.partitions:
+        while not event.isSet():
             self._do_refresh(callback=lambda: event.set())
             event.wait(timeout=1)
 
@@ -86,6 +88,8 @@ def _do_refresh(self, callback=None):
         connection = self._client.connection_manager.get_connection(address)
         if connection is None:
             self.logger.debug("Could not update partition thread as owner connection is not available.")
+            if callback:
+                callback()
             return
         request = client_get_partitions_codec.encode_request()
 

hazelcast/reactor.py

@@ -44,7 +44,8 @@ def _loop(self):
                 self.logger.exception("Error in Reactor Thread")
                 # TODO: shutdown client
                 return
-        self.logger.debug("Reactor Thread exited.")
+        self.logger.debug("Reactor Thread exited. %s" % self._timers.qsize())
+        self._cleanup_all_timers()
 
     def _check_timers(self):
         now = time.time()
@@ -71,6 +72,9 @@ def add_timer(self, delay, callback):
         return self.add_timer_absolute(delay + time.time(), callback)
 
     def shutdown(self):
+        if not self._is_live:
+            return
+        self._is_live = False
         for connection in self._map.values():
             try:
                 connection.close(HazelcastError("Client is shutting down"))
@@ -80,7 +84,6 @@ def shutdown(self):
                 else:
                     raise
         self._map.clear()
-        self._is_live = False
         self._thread.join()
 
     def new_connection(self, address, connect_timeout, socket_options, connection_closed_callback, message_callback):
@@ -89,10 +92,19 @@ def new_connection(self, address, connect_timeout, socket_options, connection_cl
 
     def _cleanup_timer(self, timer):
         try:
+            self.logger.debug("Cancel timer %s" % timer)
             self._timers.queue.remove((timer.end, timer))
         except ValueError:
             pass
 
+    def _cleanup_all_timers(self):
+        while not self._timers.empty():
+            try:
+                _, timer = self._timers.get_nowait()
+                timer.timer_ended_cb()
+            except Empty:
+                return
+
 
 class AsyncoreConnection(Connection, asyncore.dispatcher):
     sent_protocol_bytes = False

run-tests.sh

@@ -6,7 +6,7 @@ else
     USER=""
 fi
 
-HAZELCAST_VERSION="3.8.1-SNAPSHOT"
+HAZELCAST_VERSION="3.8.2-SNAPSHOT"
 
 HAZELCAST_RC_VERSION="0.2-SNAPSHOT"
 SNAPSHOT_REPO="https://oss.sonatype.org/content/repositories/snapshots"
@@ -17,7 +17,7 @@ mvn dependency:get -DrepoUrl=${SNAPSHOT_REPO} -Dartifact=com.hazelcast:hazelcast
 
 pip install -r test-requirements.txt ${USER}
 
-nohup java -cp hazelcast-remote-controller-${HAZELCAST_RC_VERSION}.jar:hazelcast-${HAZELCAST_VERSION}.jar  com.hazelcast.remotecontroller.Main>rc_stdout.log 2>rc_stderr.log &
+java -cp hazelcast-remote-controller-${HAZELCAST_RC_VERSION}.jar:hazelcast-${HAZELCAST_VERSION}.jar  com.hazelcast.remotecontroller.Main>rc_stdout.log 2>rc_stderr.log &
 
 sleep 15
 

start-rc.sh

@@ -6,7 +6,7 @@ else
     USER=""
 fi
 
-HAZELCAST_VERSION="3.8.1-SNAPSHOT"
+HAZELCAST_VERSION="3.8.2-SNAPSHOT"
 
 HAZELCAST_RC_VERSION="0.2-SNAPSHOT"
 SNAPSHOT_REPO="https://oss.sonatype.org/content/repositories/snapshots"

tests/shutdown_test.py

@@ -0,0 +1,28 @@
+from hazelcast import ClientConfig
+from hazelcast.exception import HazelcastClientNotActiveException
+from tests.base import HazelcastTestCase
+from tests.util import configure_logging
+
+
+class ShutdownTest(HazelcastTestCase):
+    rc = None
+
+    def setUp(self):
+        configure_logging()
+        self.rc = self.create_rc()
+        self.cluster = self.create_cluster(self.rc)
+
+    def tearDown(self):
+        self.shutdown_all_clients()
+        self.rc.exit()
+
+    def test_shutdown_not_hang_on_member_closed(self):
+        config = ClientConfig()
+        member = self.cluster.start_member()
+        client = self.create_client(config)
+        my_map = client.get_map("test")
+        my_map.put("key", "value").result()
+        member.shutdown()
+        with self.assertRaises(HazelcastClientNotActiveException):
+            while True:
+                my_map.get("key").result()