fix problem with aliquoting when using mini-batching, closes 3377 (#3404

) * fix problem with aliquoting when using mini-batching, closes 3377 * fix tests * make licence linter happy
SeldonIO · Jul 15, 2021 · 3de8b4e · 3de8b4e
1 parent e0d8312
commit 3de8b4e
Show file tree

Hide file tree

Showing 4 changed files with 19 additions and 4 deletions.
diff --git a/python/licenses/license_info.no_versions.csv b/python/licenses/license_info.no_versions.csv
@@ -15,7 +15,7 @@
 "cachetools","MIT License"
 "certifi","Mozilla Public License 2.0 (MPL 2.0)"
 "cffi","MIT License"
-"chardet","GNU Library or Lesser General Public License (LGPL)"
+"charset-normalizer","MIT License"
 "click","BSD License"
 "cryptography","Apache Software License; BSD License"
 "flatbuffers","Apache Software License"

diff --git a/python/seldon_core/batch_processor.py b/python/seldon_core/batch_processor.py
@@ -150,6 +150,8 @@ def _start_input_file_worker(
             q_in.put(batch)
             batch = []
         enum_idx += 1
+    if batch:
+        q_in.put(batch)
 
 
 def _start_output_file_worker(

diff --git a/python/seldon_core/metrics.py b/python/seldon_core/metrics.py
@@ -101,7 +101,7 @@ def __del__(self):
         self._manager.shutdown()
 
     def update_reward(self, reward: float):
-        """"Update metrics key corresponding to feedback reward counter."""
+        """Update metrics key corresponding to feedback reward counter."""
         if not reward or legacy_mode:
             return
         self.update(

diff --git a/testing/scripts/test_batch_processor.py b/testing/scripts/test_batch_processor.py
@@ -18,6 +18,8 @@
     wait_for_status,
 )
 
+logging.basicConfig(level=logging.DEBUG)
+
 
 class TestBatchWorker(object):
     def test_batch_worker(self, namespace):
@@ -35,6 +37,8 @@ def test_batch_worker(self, namespace):
             for i in range(batch_size):
                 f.write("[[1,2,3,4]]\n")
 
+        logging.info("Sending first batch: mini-batch size=1")
+
         start_multithreaded_batch_worker(
             "sklearn",
             "istio",
@@ -54,13 +58,20 @@ def test_batch_worker(self, namespace):
             str(uuid.uuid1()),
         )
 
+        logging.info("Finished first batch. Checking.")
+
         with open(output_data_path, "r") as f:
+            count = 0
             for line in f:
+                count += 1
                 output = json.loads(line)
                 # Ensure all requests are successful
                 assert output.get("data", {}).get("ndarray", False)
+            assert count == batch_size
 
-        # Now test that a batch size of 10 works
+        logging.info("Sending first batch: mini-batch size=30")
+
+        # Now test that with a mini batch size of 30 works
         start_multithreaded_batch_worker(
             "sklearn",
             "istio",
@@ -71,7 +82,7 @@ def test_batch_worker(self, namespace):
             "ndarray",
             100,
             3,
-            10,
+            30,
             input_data_path,
             output_data_path,
             "predict",
@@ -80,6 +91,8 @@ def test_batch_worker(self, namespace):
             str(uuid.uuid1()),
         )
 
+        logging.info("Finished first batch. Checking.")
+
         with open(output_data_path, "r") as f:
             count = 0
             for line in f: