Mirror of https://github.com/roostorg/osprey
github.com/roostorg/osprey
fix: make kafka client ids unique by hostname (#6)
+8
-1
+1
docker-compose.yaml
+1
docker-compose.yaml
···
76
76
- OSPREY_STDOUT_OUTPUT_SINK=True
77
77
- OSPREY_KAFKA_BOOTSTRAP_SERVERS=["kafka:29092"]
78
78
- OSPREY_KAFKA_INPUT_STREAM_TOPIC=osprey.actions_input
79
+
# Client ID will default to the machine hostname if it isn't defined
79
80
- OSPREY_KAFKA_INPUT_STREAM_CLIENT_ID=localhost
80
81
- OSPREY_KAFKA_OUTPUT_SINK=True
81
82
- OSPREY_KAFKA_OUTPUT_TOPIC=osprey.execution_results
+7
-1
osprey_worker/src/osprey/worker/sinks/__init__.py
+7
-1
osprey_worker/src/osprey/worker/sinks/__init__.py
···
8
8
but not ideal as clients should be responsible for patching
9
9
"""
10
10
11
+
import platform
12
+
11
13
from osprey.worker.lib.patcher import patch_all
12
14
13
15
patch_all()
···
142
144
return StaticInputStream(random_actions)
143
145
elif input_stream_source == InputStreamSource.KAFKA:
144
146
config = CONFIG.instance()
145
-
client_id = config.get_str('OSPREY_KAFKA_INPUT_STREAM_CLIENT_ID', 'localhost')
147
+
client_id = config.get_str('OSPREY_KAFKA_INPUT_STREAM_CLIENT_ID', platform.node())
148
+
client_id_suffix = config.get_optional_str('OSPREY_KAFKA_INPUT_STREAM_CLIENT_ID_SUFFIX')
146
149
input_topic: str = config.get_str('OSPREY_KAFKA_INPUT_STREAM_TOPIC', 'osprey.actions_input')
147
150
input_bootstrap_servers: list[str] = config.get_str_list('OSPREY_KAFKA_BOOTSTRAP_SERVERS', ['localhost'])
148
151
group_id = config.get_optional_str('OSPREY_KAFKA_GROUP_ID')
152
+
153
+
if client_id_suffix:
154
+
client_id = f'{client_id}-{client_id_suffix}'
149
155
150
156
consumer: PatchedKafkaConsumer = PatchedKafkaConsumer(
151
157
input_topic,