Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 9 additions & 10 deletions tests/unit/test_concurrency/test_consumer_concurrency.py
Original file line number Diff line number Diff line change
Expand Up @@ -272,31 +272,30 @@ async def handler(msg, c, f):
class TestPollTimeout:

@pytest.mark.asyncio
async def test_poll_timeout_is_100ms(self):
"""Consumer receive timeout should be 100ms, not the original 2000ms.

A 2000ms poll timeout means every service adds up to 2s of idle
blocking between message bursts. With many sequential hops in a
query pipeline, this compounds into seconds of unnecessary latency.
100ms keeps responsiveness high without significant CPU overhead.
async def test_poll_timeout_is_2000ms(self):
"""Consumer receive timeout should be 2000ms.

receive() is a blocking call that returns immediately when a
message arrives — the timeout only governs how often the loop
checks the shutdown flag during idle periods. Lower values
(e.g. 100ms) generate excessive C++ client WARN logging with
no latency benefit.
"""
consumer = _make_consumer()

# Wire up a mock Pulsar consumer that records the receive kwargs
mock_pulsar_consumer = MagicMock()
received_kwargs = {}

def capture_receive(**kwargs):
received_kwargs.update(kwargs)
# Stop after one call
consumer.running = False
raise type('Timeout', (Exception,), {})("timeout")

mock_pulsar_consumer.receive = capture_receive

await consumer.consume_from_queue(mock_pulsar_consumer)

assert received_kwargs.get("timeout_millis") == 100
assert received_kwargs.get("timeout_millis") == 2000


# ---------------------------------------------------------------------------
Expand Down
2 changes: 1 addition & 1 deletion trustgraph-base/trustgraph/base/consumer.py
Original file line number Diff line number Diff line change
Expand Up @@ -188,7 +188,7 @@ async def consume_from_queue(self, consumer, executor=None):
try:
msg = await loop.run_in_executor(
executor,
lambda: consumer.receive(timeout_millis=100),
lambda: consumer.receive(timeout_millis=2000),
)
except Exception as e:
# Handle timeout from any backend
Expand Down
4 changes: 4 additions & 0 deletions trustgraph-base/trustgraph/base/pulsar_backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -139,6 +139,10 @@ def __init__(self, host: str, api_key: str = None, listener: str = None):
if api_key:
client_args['authentication'] = pulsar.AuthenticationToken(api_key)

client_args['logger'] = pulsar.ConsoleLogger(
_pulsar.LoggerLevel.Error
)

self.client = pulsar.Client(**client_args)
logger.info(f"Pulsar client connected to {host}")

Expand Down
Loading