Merge branch 'master' of https://github.com/scrapinghub/crawl-frontier

Author: sibiryakov

frontera/contrib/backends/redis_backend/__init__.py

@@ -258,7 +258,7 @@ def get(obj):
 
         [get(obj) for obj in objs]
 
-    def flush(self, force_clear):
+    def flush(self, force_clear=False):
         if len(self._cache) > self._cache_size_limit:
             force_clear = True
         [self._redis_pipeline.hmset(fprint, {FIELD_STATE: state}) for (fprint, state) in self._cache.items()]

frontera/contrib/messagebus/kafka/offsets_fetcher.py

@@ -178,7 +178,7 @@ def offsets(self, partitions, timestamp):
 
                 if future.exception.invalid_metadata:
                     refresh_future = self._client.cluster.request_update()
-                    self._client.poll(future=refresh_future, sleep=True)
+                    self._client.poll(future=refresh_future)
                     log.warning("Got exception %s and kept the loop", future.exception)
             if offsets:
                 return offsets

frontera/settings/default_settings.py

@@ -49,6 +49,7 @@
 REDIS_BACKEND_CODEC = 'frontera.contrib.backends.remote.codecs.msgpack'
 REDIS_HOST = 'localhost'
 REDIS_PORT = 6379
+REDIS_STATE_CACHE_SIZE_LIMIT = 0
 REQUEST_MODEL = 'frontera.core.models.Request'
 RESPONSE_MODEL = 'frontera.core.models.Response'
 

tests/contrib/backends/redis_backend/test_redis.py

@@ -501,6 +501,13 @@ def test_get_next_request_has_requests(self):
         requests = subject.get_next_requests(max_next_requests=10, partitions=['0', '1'])
         self.assertEqual(3, len(requests))
 
+    def test_close_manager(self):
+        settings = Settings(module='frontera.settings.default_settings')
+        settings.set('BACKEND', 'frontera.contrib.backends.redis_backend.RedisBackend')
+        manager = WorkerFrontierManager.from_settings(settings, strategy_worker=True)
+        self.assertEqual(RedisBackend, manager.backend.__class__)
+        manager.close()
+
 
 if __name__ == '__main__':
     main()