fix: replace table with GlobalTable to allow data share between workers

faust_route_table/Makefile

@@ -14,13 +14,17 @@ clean:
 
 .PHONY: up
 up:
-	docker-compose up -d
+	docker-compose up
 
 .PHONY: topics
 topics:
 	docker exec -ti faust_route_table_kafka_1 kafka-topics --create --topic hellos-here --zookeeper zookeeper:2181 --partitions 8 --replication-factor 1
 	docker exec -ti faust_route_table_kafka_1 kafka-topics --create --topic points-here --zookeeper zookeeper:2181 --partitions 8 --replication-factor 1
 
+show-changelog:
+	docker exec -ti faust_route_table_kafka_1 kafka-console-consumer --bootstrap-server localhost:9092 --topic get-hello-hello_counts-changelog --from-beginning
+
+
 .PHONY: list-topics
 list-topics:
 	docker exec -ti faust_route_table_kafka_1 kafka-topics --list --zookeeper zookeeper:2181
@@ -45,8 +49,14 @@ scan:
 run/generator:
 	${VENV} ; faust -A send_hellos worker -l info
 
-run/worker:
-	${VENV} ; faust -A get_hellos worker -l info --web-port ${WEB_PORT} --web-host localhost
+run-worker-1:
+	${VENV} ; faust -A get_hellos worker -l info --web-port 6067 --web-host localhost
+
+run-worker-2:
+	${VENV} ; faust -A get_hellos worker -l info --web-port 6068 --web-host localhost
+
+query-worker-1:
+	for i in $(shell seq 0 10); do curl localhost:6067/count/?i=hellohello-$$i; echo; done
 
-query:
-	for i in $(shell seq 0 11); do curl localhost:6669/count/?i=hellohello-$$i; echo; done
+query-worker-2:
+	for i in $(shell seq 0 10); do curl localhost:6068/count/?i=hellohello-$$i; echo; done

faust_route_table/get_hellos.py

@@ -3,16 +3,18 @@
 app = faust.App("get-hello")
 
 hello_topic = app.topic("hellos-here")
-hello_counts = app.Table('hello_counts', default=int)
+hello_counts = app.GlobalTable('hello_counts', default=int, partitions=8)
 
 @app.agent(hello_topic)
 async def get_hello(t):
     async for e in t:
-        print(e)
-        print(app.conf)
-        hello_counts[e.encode()] += 1
+        hello_counts[e] += 1
+        result = ""
+
         for k,v in hello_counts.items():
-            print(f"{k}: {v}")
+            result += f"{k}: {v}. "
+
+        print(result)
 
 @app.page('/count/')
 @app.table_route(table=hello_counts, query_param='i')

faust_route_table/send_hellos.py

@@ -11,7 +11,7 @@
 hello_topic = app.topic("hellos-here")
 hello_counts_send = defaultdict(int)
 
-@app.timer(interval=1)
+@app.timer(interval=2)
 async def hello():
     key = str(random.randint(0, 10))
     val = f"hellohello-{key}"