review comments

Author: rerpha

README.md

@@ -7,7 +7,9 @@ Also allows replaying data in a topic.
 
 # Usage
 
-To run the latest version, use `uvx saluki <args>`.
+To run the latest version, install [uv](https://docs.astral.sh/uv/getting-started/installation/) and use `uvx saluki <args>`.
+
+alternatively you can `pip install saluki` and run it from a `venv`. 
 
 See `saluki --help` for all options. 
 
@@ -28,6 +30,25 @@ You can also filter out messages to specific schema(s) with the `-f` flag, like
 ## `sniff` - List all topics and their high, low watermarks and number of messages
 `saluki sniff mybroker:9092`
 
+Output looks as follows:
+
+```
+$ saluki sniff mybroker:9092
+
+INFO:saluki:Cluster ID: redpanda.0faa4595-7298-407e-9db7-7e2758d1af1f
+INFO:saluki:Brokers:
+INFO:saluki:    192.168.0.111:9092/1
+INFO:saluki:    192.168.0.112:9092/2
+INFO:saluki:    192.168.0.113:9092/0
+INFO:saluki:Topics:
+INFO:saluki:    MERLIN_events:
+INFO:saluki:            0 - low:262322729, high:302663378, num_messages:40340649
+INFO:saluki:    MERLIN_runInfo:
+INFO:saluki:            0 - low:335, high:2516, num_messages:2181
+INFO:saluki:    MERLIN_monitorHistograms:
+INFO:saluki:            0 - low:7515, high:7551, num_messages:36
+```
+
 ## `play` - Replay data from one topic to another
 
 ### Between offsets

src/saluki/consume.py

@@ -1,4 +1,5 @@
 import logging
+import uuid
 
 from confluent_kafka import Consumer, TopicPartition
 
@@ -31,7 +32,7 @@ def consume(
     c = Consumer(
         {
             "bootstrap.servers": broker,
-            "group.id": "saluki-consume",
+            "group.id": f"saluki-consume-{uuid.uuid4()}",
             "session.timeout.ms": 6000,
             "auto.offset.reset": "latest",
             "enable.auto.offset.store": False,

src/saluki/listen.py

@@ -1,4 +1,5 @@
 import logging
+import uuid
 
 from confluent_kafka import Consumer, TopicPartition
 
@@ -24,7 +25,7 @@ def listen(
     c = Consumer(
         {
             "bootstrap.servers": broker,
-            "group.id": "saluki-listen",
+            "group.id": f"saluki-listen-{uuid.uuid4()}",
             "auto.offset.reset": "latest",
             "enable.auto.commit": False,
         }

src/saluki/play.py

@@ -1,4 +1,5 @@
 import logging
+import uuid
 
 from confluent_kafka import Consumer, Producer, TopicPartition
 
@@ -28,7 +29,7 @@ def play(
     consumer = Consumer(
         {
             "bootstrap.servers": src_broker,
-            "group.id": "saluki-play",
+            "group.id": f"saluki-play-{uuid.uuid4()}",
         }
     )
     producer = Producer(
@@ -57,7 +58,8 @@ def play(
 
     try:
         msgs = consumer.consume(num_messages)
-        [producer.produce(dest_topic, message.value(), message.key()) for message in msgs]
+        for message in msgs:
+            producer.produce(dest_topic, message.value(), message.key())
         producer.flush()
     except Exception:
         logger.exception("Got exception while replaying:")

src/saluki/sniff.py

@@ -1,4 +1,5 @@
 import logging
+import uuid
 
 from confluent_kafka import Consumer, TopicPartition
 from confluent_kafka.admin import AdminClient
@@ -12,11 +13,12 @@ def sniff(broker: str) -> None:
     :param broker: The broker address including port number.
     """
     a = AdminClient({"bootstrap.servers": broker})
-    c = Consumer({"bootstrap.servers": broker, "group.id": "saluki-sniff"})
+    c = Consumer({"bootstrap.servers": broker, "group.id": f"saluki-sniff-{uuid.uuid4()}"})
     t = a.list_topics(timeout=5)
     logger.info(f"Cluster ID: {t.cluster_id}")
     logger.info("Brokers:")
-    [logger.info(f"\t{value}") for value in t.brokers.values()]
+    for value in t.brokers.values():
+        logger.info(f"\t{value}")
 
     logger.info("Topics:")