Merge pull request #26 from aiven/mte-fix-schema-registry-heartbeat

fix schema registry heartbeat

#26

Makefile

@@ -48,11 +48,19 @@ $(KAFKA_PATH): $(KAFKA_TAR)
 .PHONY: fetch-kafka
 fetch-kafka: $(KAFKA_PATH)
 
-.PHONY: kafka
-kafka: fetch-kafka
+.PHONY: start-kafka
+start-kafka: fetch-kafka
 	$(KAFKA_PATH)/bin/zookeeper-server-start.sh $(KAFKA_PATH)/config/zookeeper.properties &
 	$(KAFKA_PATH)/bin/kafka-server-start.sh $(KAFKA_PATH)/config/server.properties &
 
+.PHONY: stop-kafka
+stop-kafka:
+	$(KAFKA_PATH)/bin/kafka-server-stop.sh || true
+	$(KAFKA_PATH)/bin/zookeeper-server-stop.sh || true
+
+.PHONY: kafka
+kafka: start-kafka
+
 .PHONY: pylint
 pylint: $(GENERATED)
 	python3 -m pylint --rcfile .pylintrc $(ALL_PYTHON_DIRS)

karapace/master_coordinator.py

@@ -159,17 +159,20 @@ def close(self):
         self.running = False
 
     def run(self):
+        _hb_interval = 3.0
         while self.running:
             try:
                 if not self.kafka_client:
                     if self.init_kafka_client() is False:
                         continue
                 if not self.sc:
                     self.init_schema_coordinator()
+                    _hb_interval = self.sc.config["heartbeat_interval_ms"] / 1000
 
                 self.sc.ensure_active_group()
+                self.sc.poll_heartbeat()
                 self.log.debug("We're master: %r: master_uri: %r", self.sc.master, self.sc.master_url)
-                time.sleep(5.0)
+                time.sleep(min(_hb_interval, self.sc.time_to_next_heartbeat()))
             except:  # pylint: disable=bare-except
                 self.log.exception("Exception in master_coordinator")
                 time.sleep(1.0)