Misc updates

bubuku/broker.py

@@ -3,7 +3,7 @@
 from time import time, sleep
 
 from bubuku.config import KafkaProperties
-from bubuku.id_generator import BrokerIdGenerator
+from bubuku.id_extractor import BrokerIdExtractor
 from bubuku.process import KafkaProcess
 from bubuku.zookeeper import BukuExhibitor
 
@@ -76,7 +76,7 @@ def on_timeout_fail(self):
 
 class BrokerManager(object):
     def __init__(self, process: KafkaProcess, exhibitor: BukuExhibitor,
-                 id_manager: BrokerIdGenerator, kafka_properties: KafkaProperties, timeout: StartupTimeout):
+                 id_manager: BrokerIdExtractor, kafka_properties: KafkaProperties, timeout: StartupTimeout):
         self.id_manager = id_manager
         self.exhibitor = exhibitor
         self.kafka_properties = kafka_properties

bubuku/env_provider.py

@@ -8,7 +8,7 @@
 import requests
 
 from bubuku.config import Config, KafkaProperties
-from bubuku.id_generator import BrokerIdGenerator
+from bubuku.id_extractor import BrokerIdExtractor
 from bubuku.zookeeper import BukuExhibitor, AddressListProvider
 from bubuku.zookeeper.exhibitor import ExhibitorAddressProvider
 
@@ -86,7 +86,7 @@ def get_address_provider(self):
             return ExhibitorAddressProvider(partial(self._load_instance_ips, self.config.zk_stack_name))
 
     def create_broker_id_manager(self, zk: BukuExhibitor, kafka_props: KafkaProperties):
-        return BrokerIdGenerator(zk, kafka_props)
+        return BrokerIdExtractor(zk, kafka_props)
 
 
 class _LocalAddressProvider(AddressListProvider):
@@ -117,4 +117,4 @@ def get_rack(self):
         return None
 
     def create_broker_id_manager(self, zk: BukuExhibitor, kafka_props: KafkaProperties):
-        return BrokerIdGenerator(zk, kafka_props)
+        return BrokerIdExtractor(zk, kafka_props)

bubuku/id_generator.py → bubuku/id_extractor.py

@@ -2,14 +2,22 @@
 import logging
 import os
 import re
+from typing import Optional, List
 
 from bubuku.config import KafkaProperties
 from bubuku.zookeeper import BukuExhibitor
 
 _LOG = logging.getLogger('bubuku.id_generator')
 
 
-class BrokerIdGenerator(object):
+def _search_broker_id(lines: List[str]) -> Optional[str]:
+    for line in lines:
+        match = re.search('^broker\\.id=(\\d+)$', line.strip())
+        if match:
+            return match.group(1)
+
+
+class BrokerIdExtractor(object):
     def __init__(self, zk: BukuExhibitor, kafka_properties: KafkaProperties):
         super().__init__()
         self.zk = zk
@@ -24,17 +32,11 @@ def get_broker_id(self):
         while not os.path.isfile(meta_path):
             return None
         with open(meta_path) as f:
-            lines = f.readlines()
-            for line in lines:
-                match = re.search('broker\.id=(\d+)', line)
-                if match:
-                    self.broker_id = match.group(1)
-                    return self.broker_id
-        return None
+            self.broker_id = _search_broker_id(f.readlines())
+        return self.broker_id
 
     def is_registered(self):
         broker_id = self.get_broker_id()
         if broker_id:
             return self.zk.is_broker_registered(broker_id)
         return False
-

delivery.yaml

@@ -4,7 +4,7 @@ pipeline:
     type: script
     overlay: guild-python/legacy
     env:
-        PYENV_VERSION: 3.6.10
+        PYENV_VERSION: 3.8.10
     commands:
       - desc: Install Docker
         cmd: |

tests/test_broker_id_generator.py

@@ -0,0 +1,20 @@
+import unittest
+
+from bubuku.id_extractor import _search_broker_id
+
+
+class TestBrokerIdExtractor(unittest.TestCase):
+    def test_match_valid(self):
+        assert '123534' == _search_broker_id(['broker.id=123534'])
+        assert '123534' == _search_broker_id(['\tbroker.id=123534'])
+        assert '123534' == _search_broker_id(['\tbroker.id=123534\n'])
+        assert '123534' == _search_broker_id(['broker.id=123534 \n\r'])
+        assert '123534' == _search_broker_id(['\tbroker.id=123534 \r'])
+        assert '123534' == _search_broker_id(['xbroker.id=1', 'broker.id=123534'])
+        assert '123534' == _search_broker_id(['broker.id=123534', 'boker.id=123534'])
+
+    def test_match_invalid(self):
+        assert _search_broker_id([]) is None
+        assert _search_broker_id(['broker_id=123534']) is None
+        assert _search_broker_id(['xbroker.id=1', 'broker.id=12f3534']) is None
+        assert _search_broker_id(['bruker.id=123534', 'boker.id=123534']) is None

tests/test_check_time_period.py

@@ -3,7 +3,7 @@
 
 
 def test_check_time_period():
-    test_check = TestCheck()
+    test_check = _TestCheck()
 
     assert test_check.check_if_time() is not None  # first time it should always run
     assert test_check.check_if_time() is None  # time has not come yet
@@ -14,7 +14,7 @@ def test_check_time_period():
     assert 0.0 < test_check.time_till_check() < 1  # there's still some time before the check can be run again
 
 
-class TestCheck(Check):
+class _TestCheck(Check):
     def __init__(self):
         super().__init__(check_interval_s=0.5)
 

tests/test_daemon.py

@@ -7,7 +7,7 @@
 from test_config import build_test_properties
 
 
-class TestController(object):
+class _TestController(object):
     def __init__(self):
         self.checks = []
 
@@ -18,7 +18,7 @@ def test_load_restart_on_exhibitor():
     exhibitor = object()
     broker = object()
 
-    controller = TestController()
+    controller = _TestController()
 
     apply_features(-1, {'restart_on_exhibitor': {}}, controller, exhibitor, broker, None, None)
 
@@ -33,7 +33,7 @@ def test_rebalance_on_start():
     exhibitor = object()
     broker = object()
 
-    controller = TestController()
+    controller = _TestController()
 
     apply_features(-1, {'rebalance_on_start': {}}, controller, exhibitor, broker, None, None)
 
@@ -49,7 +49,7 @@ def test_rebalance_on_broker_list_change():
     exhibitor = object()
     broker = object()
 
-    controller = TestController()
+    controller = _TestController()
 
     apply_features(-1, {'rebalance_on_brokers_change': {}}, controller, exhibitor, broker, None, None)
 
@@ -67,7 +67,7 @@ def test_graceful_terminate():
 
     broker = object()
 
-    controller = TestController()
+    controller = _TestController()
 
     apply_features(-1, {'graceful_terminate': {}}, controller, None, broker, None, None)