Unable to consume from streams in Scala or Python

[thoefkens]

Describe the bug

We have tried using rabbitmq streams to produce and consume, both in Scala and python-based pods.
Producing is never an issue, we see messages arriving on test streams, yet they never get consumed, even though or consumers show heartbeats, are connecting to the streams - but messages simply do not get consumed.

Reproduction steps

1. Use image:
   registry: docker.io
   repository: bitnami/rabbitmq
   tag: 4.1.0-debian-12-r1
   debug: true
2. Deploy a 3-node rabbimq cluster with streams enabled
3. Deploy our producer and consumer code

import os
import json
import logging
from datetime import datetime
from flask import Flask, request, jsonify
from rstream import Producer, AMQPMessage
from rstream.exceptions import StreamDoesNotExist
import asyncio
import threading

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = Flask(__name__)

# RabbitMQ configuration
RABBITMQ_HOST = os.getenv('RABBITMQ_HOST', 'rabbitmq')
RABBITMQ_PORT = int(os.getenv('RABBITMQ_PORT', '5552'))
RABBITMQ_USERNAME = os.getenv('RABBITMQ_USERNAME', 'admin')
RABBITMQ_PASSWORD = os.getenv('RABBITMQ_PASSWORD', 'xxx')
STREAM_NAME = os.getenv('STREAM_NAME', 'SomeStream')

# Global producer instance
producer = None
stream_declared = False

async def setup_rabbitmq_stream():
    global producer, stream_declared
    try:
        # Create producer
        producer = Producer(
            host=RABBITMQ_HOST,
            port=RABBITMQ_PORT,
            username=RABBITMQ_USERNAME,
            password=RABBITMQ_PASSWORD
        )
        
        await producer.start()
        logger.info(f"Connected to RabbitMQ stream at {RABBITMQ_HOST}:{RABBITMQ_PORT}")
        
        # Check/create stream with retention configuration
        try:
            # Always recreate the stream to ensure proper configuration
            logger.info(f"Attempting to delete and recreate stream {STREAM_NAME} with proper configuration")
            
            # Try to delete existing stream first
            try:
                await producer.delete_stream(STREAM_NAME)
                logger.info(f"Deleted existing stream {STREAM_NAME}")
            except Exception as delete_error:
                logger.info(f"Stream {STREAM_NAME} might not exist or couldn't be deleted: {delete_error}")
            
            # Create stream with proper retention configuration
            logger.info(f"Creating stream {STREAM_NAME} with retention configuration")
            await producer.create_stream(
                stream=STREAM_NAME,
                arguments={
                    "x-max-length-bytes": 1_000_000_000,  # 1GB
                    "x-max-age": "7D",  # 7 days
                    "x-stream-max-segment-size-bytes": 100_000_000  # 100MB
                }
            )
            logger.info(f"Created stream {STREAM_NAME} with proper retention settings")
            
        except Exception as stream_error:
            logger.warning(f"Stream setup issue: {stream_error}")
            # Try to continue anyway in case stream exists and is usable
        
        stream_declared = True
        logger.info(f"Stream {STREAM_NAME} ready for use")
        
        return True
        
    except Exception as e:
        logger.error(f"Failed to setup RabbitMQ stream: {e}")
        return False

async def send_to_stream(payload):
    global producer, stream_declared
    try:
        if producer is None:
            logger.error("Producer not initialized")
            return False
        
        if not stream_declared:
            logger.error("Stream not declared/verified")
            return False
        
        # Create message
        message_body = json.dumps(payload).encode('utf-8')
        message = AMQPMessage(body=message_body)
        
        logger.info(f"Sending message to stream {STREAM_NAME}")
        
        # Send message using fire-and-forget approach
        await producer.send(stream=STREAM_NAME, message=message)
        
        logger.info(f"Message sent to stream {STREAM_NAME}")
        return True
                
    except Exception as e:
        logger.error(f"Error in send_to_stream: {e}")
        logger.error(f"Error type: {type(e).__name__}")
        return False

def run_async_task(coro):
    """Run async coroutine in the background thread"""
    try:
        loop = asyncio.get_event_loop()
    except RuntimeError:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
    
    return loop.run_until_complete(coro)

@app.route('/v1/input/plots', methods=['POST'])
def receive_plots():
    try:
        # Get JSON payload
        if not request.is_json:
            return jsonify({'error': 'Content-Type must be application/json'}), 400
        
        payload = request.get_json()
        if payload is None:
            return jsonify({'error': 'Invalid JSON payload'}), 400
        
        # Add timestamp to payload
        payload['timestamp'] = datetime.utcnow().isoformat() + 'Z'
        payload['source'] = 'plot-receiver'
        
        logger.info(f"Received plot data: {payload}")
        
        # Send to RabbitMQ stream asynchronously
        def send_async():
            success = run_async_task(send_to_stream(payload))
            timestamp = datetime.utcnow().isoformat() + 'Z'
            if success:
                print(f"[{timestamp}] forwarded TrackerDataEvent")
                logger.info(f"[{timestamp}] forwarded TrackerDataEvent")
            else:
                print(f"[{timestamp}] failed to forward TrackerDataEvent")
                logger.error(f"[{timestamp}] failed to forward TrackerDataEvent")
        
        # Start background thread for sending
        thread = threading.Thread(target=send_async)
        thread.start()
        
        return jsonify({'message': 'Plot data received and queued for processing'}), 200
        
    except Exception as e:
        logger.error(f"Error processing plot data: {e}")
        return jsonify({'error': 'Internal server error'}), 500

@app.route('/health', methods=['GET'])
def health_check():
    global producer, stream_declared
    status = {
        'status': 'healthy',
        'producer_connected': producer is not None,
        'stream_declared': stream_declared
    }
    return jsonify(status), 200

if __name__ == '__main__':
    # Setup RabbitMQ stream in background thread
    def setup_in_background():
        success = run_async_task(setup_rabbitmq_stream())
        if not success:
            logger.error("Failed to setup RabbitMQ stream")
    
    setup_thread = threading.Thread(target=setup_in_background, daemon=True)
    setup_thread.start()
    
    # Give some time for setup
    import time
    time.sleep(3)
    
    app.run(host='0.0.0.0', port=9000, debug=False) 

and the consumer:

import os
import json
import logging
import asyncio
import signal
from datetime import datetime
from rstream import Consumer, AMQPMessage, ConsumerOffsetSpecification, OffsetType
from rstream.exceptions import StreamDoesNotExist

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# RabbitMQ configuration
RABBITMQ_HOST = os.getenv('RABBITMQ_HOST', 'rabbitmq')
RABBITMQ_PORT = int(os.getenv('RABBITMQ_PORT', '5552'))
RABBITMQ_USERNAME = os.getenv('RABBITMQ_USERNAME', 'admin')
RABBITMQ_PASSWORD = os.getenv('RABBITMQ_PASSWORD', 'xxx')
STREAM_NAME = os.getenv('STREAM_NAME', 'SomeStream')
CONSUMER_NAME = os.getenv('CONSUMER_NAME', 'plot-consumer')

class PlotConsumer:
    def __init__(self):
        self.consumer = None
        self.running = True
        self.message_count = 0
        
    async def on_message(self, cr: Consumer, message: AMQPMessage):
        try:
            self.message_count += 1
            logger.info(f"🎉 Message received! Count: {self.message_count}")
            
            # Log raw message details for debugging
            logger.info(f"Message type: {type(message)}")
            logger.info(f"Message body type: {type(message.body)}")
            logger.info(f"Message body length: {len(message.body) if message.body else 0}")
            
            # Decode message body
            if message.body:
                body = message.body.decode('utf-8')
                logger.info(f"Raw message body: {body}")
                
                payload = json.loads(body)
                
                # Print received message with timestamp
                timestamp = datetime.utcnow().isoformat() + 'Z'
                print(f"[{timestamp}] received TrackerDataEvent")
                logger.info(f"[{timestamp}] received TrackerDataEvent: {payload}")
            else:
                logger.warning("Message body is empty or None")
            
        except json.JSONDecodeError as json_error:
            logger.error(f"JSON decode error: {json_error}")
            logger.error(f"Raw body that failed to decode: {message.body}")
        except UnicodeDecodeError as unicode_error:
            logger.error(f"Unicode decode error: {unicode_error}")
            logger.error(f"Raw bytes: {message.body}")
        except Exception as e:
            logger.error(f"CRITICAL: Exception in message callback: {e}")
            logger.error(f"Exception type: {type(e).__name__}")
            logger.exception("Full traceback:")
    
    async def start_consuming(self):
        retry_count = 0
        max_retries = 5
        
        while self.running and retry_count < max_retries:
            try:
                logger.info(f"Creating consumer (attempt {retry_count + 1})")
                self.consumer = Consumer(
                    host=RABBITMQ_HOST,
                    port=RABBITMQ_PORT,
                    username=RABBITMQ_USERNAME,
                    password=RABBITMQ_PASSWORD
                )
                
                await self.consumer.start()
                logger.info(f"✅ Connected to RabbitMQ stream at {RABBITMQ_HOST}:{RABBITMQ_PORT}")
                
                try:
                    # Check if stream exists first
                    logger.info(f"Checking if stream {STREAM_NAME} exists...")
                    
                    # Subscribe to the stream starting from the first messages
                    logger.info(f"📡 Subscribing to stream {STREAM_NAME} from FIRST offset")
                    logger.info(f"📋 Callback function: {self.on_message}")
                    logger.info(f"📋 Callback is callable: {callable(self.on_message)}")
                    
                    await self.consumer.subscribe(
                        stream=STREAM_NAME,
                        callback=self.on_message,
                        decoder=None,
                        offset_specification=ConsumerOffsetSpecification(OffsetType.FIRST, None)
                    )
                    
                    logger.info(f"✅ Started consuming from stream: {STREAM_NAME}")
                    logger.info(f"👂 Waiting for messages... (message count: {self.message_count})")
                    logger.info(f"🔧 Consumer subscribed successfully - callback should be active")
                    
                    # Reset retry count on successful connection
                    retry_count = 0
                    
                    # Keep running until stopped
                    heartbeat_counter = 0
                    while self.running:
                        await asyncio.sleep(10)  # Check every 10 seconds
                        heartbeat_counter += 1
                        if heartbeat_counter % 6 == 0:  # Every minute
                            logger.info(f"💓 Consumer alive - Total messages received: {self.message_count}")
                        
                except StreamDoesNotExist:
                    logger.error(f"❌ Stream '{STREAM_NAME}' does not exist")
                    logger.info("ℹ️  Stream might need to be created by the producer first")
                    retry_count += 1
                    await asyncio.sleep(10)
                except Exception as subscribe_error:
                    logger.error(f"❌ Subscription error: {subscribe_error}")
                    logger.exception("Subscription error details:")
                    retry_count += 1
                    await asyncio.sleep(5)
                    
            except Exception as e:
                retry_count += 1
                logger.error(f"❌ Consumer error (attempt {retry_count}): {e}")
                logger.error(f"Error type: {type(e).__name__}")
                logger.exception("Full exception traceback:")
                
                if retry_count < max_retries:
                    wait_time = min(retry_count * 5, 30)  # Max 30 seconds
                    logger.info(f"⏳ Waiting {wait_time} seconds before retry...")
                    await asyncio.sleep(wait_time)
                else:
                    logger.error(f"❌ Max retries ({max_retries}) reached. Giving up.")
                    break
            finally:
                if self.consumer:
                    try:
                        await self.consumer.close()
                        logger.info("🔒 Consumer closed")
                    except:
                        pass
                    self.consumer = None
    
    async def stop(self):
        self.running = False
        if self.consumer:
            await self.consumer.close()

async def main():
    logger.info("🚀 Starting Plot Consumer")
    logger.info(f"📡 Target: {RABBITMQ_HOST}:{RABBITMQ_PORT}")
    logger.info(f"📺 Stream: {STREAM_NAME}")
    logger.info(f"👤 Consumer: {CONSUMER_NAME}")
    
    consumer = PlotConsumer()
    
    # Setup signal handlers for graceful shutdown
    def signal_handler():
        logger.info("📨 Received shutdown signal")
        asyncio.create_task(consumer.stop())
    
    # Register signal handlers
    loop = asyncio.get_event_loop()
    for sig in [signal.SIGTERM, signal.SIGINT]:
        loop.add_signal_handler(sig, signal_handler)
    
    try:
        await consumer.start_consuming()
    except KeyboardInterrupt:
        logger.info("⚠️ Interrupted by user")
    except Exception as e:
        logger.error(f"💥 Unexpected error: {e}")
        logger.exception("Full main exception traceback:")
    finally:
        await consumer.stop()
        logger.info("👋 Plot Consumer stopped")

if __name__ == '__main__':
    asyncio.run(main()) 

The message received block is never reached.

Expected behavior

We would expect messages to be picked up from the stream. This did not work in Scala either...

Additional context

No response

[michaelklishin]

@thoefkens you haven't provided any evidence of a bug in RabbitMQ, so this is discussion material.

Your code uses a combination of asyncio, threads and signal handlers. Start with this Python tutorial that happens to use the same client library, rstream but is much smaller in scope.

Plus start with just one standalone node, not a cluster on Kubernetes or whatever. Get that to work first, then add complexity as you go, we won't troubleshoot other people's K8S deployments and guess what's in your logs.

[Gsantomaggio]

@thoefkens, following what @michaelklishin said, I would start with something easier.

Easy way to test:

 docker run -it --rm --name rabbitmq -p 5552:5552 -p 5672:5672 -p 15672:15672 \
    -e RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS='-rabbitmq_stream advertised_host localhost' \
    rabbitmq:4-management

then

docker exec rabbitmq rabbitmq-plugins enable rabbitmq_stream_management

 ➜ python3 -m venv venv
 ➜ source venv/bin/activate
(venv) ➜  pip install stream
(venv) ➜ python3 send.py
 [x] Hello, World! message sent
 [x] Press Enter to close the producer  ...

(venv) ➜ python3 receive.py
Press control + C to close
Got message: b'Hello, World!' from stream hello-python-stream

When this ^^^ works, I would add a cluster configuration.
Here, you can find a cluster example with TLS if you prefer.

Note: In case of a cluster configuration with a proxy / load-balancer, you need to use load balancer mode

[thoefkens]

Thank you - that works just fine - I am just surprised that similar setup on a 3 node cluster does not work. Is there any working sample on K8S by chance that you may know of?

[Gsantomaggio]

cluster does not work.

Sorry @thoefkens but it is not enough to understand the problem. You should add some detail, for example, client logs and/or server logs.

Have you tried the load-balancer configuration?

[thoefkens]

Hi well, yes I did.
I then also did a small test with this consumer:

import asyncio
import os
import json
import logging
from rstream import (
    AMQPMessage,
    Consumer,
    ConsumerOffsetSpecification,
    MessageContext,
    OffsetType,
)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

RABBITMQ_HOST = os.getenv('RABBITMQ_HOST', 'rabbitmq.infrastructure.svc.cluster.local')
RABBITMQ_PORT = int(os.getenv('RABBITMQ_PORT', '5552'))
RABBITMQ_USERNAME = os.getenv('RABBITMQ_USERNAME', 'admin')
RABBITMQ_PASSWORD = os.getenv('RABBITMQ_PASSWORD', 'YOUR_SECURE_PASSWORD')
STREAM_NAME = os.getenv('STREAM_NAME', 'hello-python-stream')

STREAM_RETENTION = 5000000000  # 5GB

async def receive():
    try:
        logger.info(f"Attempting to connect to {RABBITMQ_HOST}...")
        
        # Simplified connection - exactly like working receive.py
        async with Consumer(
            host=RABBITMQ_HOST,
            port=RABBITMQ_PORT,
            username=RABBITMQ_USERNAME,
            password=RABBITMQ_PASSWORD
        ) as consumer:
            # Create stream if it doesn't exist - exactly like receive.py
            logger.info(f"Creating/verifying stream: {STREAM_NAME}")
            await consumer.create_stream(
                STREAM_NAME, exists_ok=True, arguments={"max-length-bytes": STREAM_RETENTION}
            )

            # Message callback - simplified like working receive.py
            async def on_message(msg: AMQPMessage, message_context: MessageContext):
                try:
                    stream = message_context.consumer.get_stream(message_context.subscriber_name)
                    
                    # Print like working receive.py - simple and direct
                    print("Got message: {} from stream {}".format(msg, stream))
                    logger.info(f"*** MESSAGE RECEIVED *** from {RABBITMQ_HOST}: {msg}")
                    
                except Exception as e:
                    logger.error(f"Error processing message: {e}")

            logger.info(f"Successfully connected to {RABBITMQ_HOST}")
            logger.info("Starting consumer...")
            await consumer.start()
            
            logger.info(f"Subscribing to stream '{STREAM_NAME}' with FIRST offset...")
            await consumer.subscribe(
                stream=STREAM_NAME,
                callback=on_message,
                offset_specification=ConsumerOffsetSpecification(OffsetType.FIRST, None),
            )
            
            logger.info("Consumer subscribed successfully - waiting for messages...")
            logger.info("Press control + C to close")
            
            try:
                await consumer.run()
            except (KeyboardInterrupt, asyncio.CancelledError):
                print("Closing Consumer...")
                return
                
    except Exception as e:
        logger.error(f"Failed to connect to {RABBITMQ_HOST}: {e}")

if __name__ == '__main__':
    logger.info("Starting Plot Consumer - Simplified like working receive.py")
    logger.info(f"RabbitMQ Host: {RABBITMQ_HOST}")
    logger.info(f"RabbitMQ Port: {RABBITMQ_PORT}")
    logger.info(f"Stream Name: {STREAM_NAME}")
    logger.info(f"Username: {RABBITMQ_USERNAME}")
    logger.info("Using FIRST offset - will read all messages from the beginning of the stream")
    
    with asyncio.Runner() as runner:
        runner.run(receive()) 
        

And I sent some messages also manually onto the stream, I see them arriving, the consumer is connected but the messages are never handled.

[thoefkens]

These are the consumer logs:

INFO:__main__:Using FIRST offset - will read all messages from the beginning of the stream
INFO:__main__:Attempting to connect to rabbitmq.infrastructure.svc.cluster.local...
INFO:__main__:Creating/verifying stream: hello-python-stream
INFO:__main__:Successfully connected to rabbitmq.infrastructure.svc.cluster.local
INFO:__main__:Starting consumer...
INFO:__main__:Subscribing to stream 'hello-python-stream' with FIRST offset...
INFO:__main__:Consumer subscribed successfully - waiting for messages...

[Gsantomaggio]

can't see load_balancer_mode btw

[michaelklishin]

@thoefkens are you aware of this blog post on the nuances of connections using the RabbitMQ Stream Protocol?

It's essential to know if you intend to deploy your apps with a load balancer.

[Gsantomaggio]

@thoefkens I tested your setup

Cluster from https://github.com/rabbitmq/rabbitmq-stream-go-client/tree/main/compose:

git clone [email protected]:rabbitmq/rabbitmq-stream-go-client.git .
make rabbitmq-ha-proxy 

Three cluster nodes:

Run your script ( with a lb mode and changed the port to 5553)

import asyncio
import os
import json
import logging
from rstream import (
    AMQPMessage,
    Consumer,
    ConsumerOffsetSpecification,
    MessageContext,
    OffsetType,
)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

RABBITMQ_HOST = os.getenv('RABBITMQ_HOST', 'localhost')
RABBITMQ_PORT = int(os.getenv('RABBITMQ_PORT', '5553'))
RABBITMQ_USERNAME = os.getenv('RABBITMQ_USERNAME', 'guest')
RABBITMQ_PASSWORD = os.getenv('RABBITMQ_PASSWORD', 'guest')
STREAM_NAME = os.getenv('STREAM_NAME', 'hello-python-stream')

STREAM_RETENTION = 5000000000  # 5GB


async def receive():
    try:
        logger.info(f"Attempting to connect to {RABBITMQ_HOST}...")

        # Simplified connection - exactly like working receive.py
        async with Consumer(
                host=RABBITMQ_HOST,
                port=RABBITMQ_PORT,
                username=RABBITMQ_USERNAME,
                password=RABBITMQ_PASSWORD,
                load_balancer_mode=True,
        ) as consumer:
            # Create stream if it doesn't exist - exactly like receive.py
            logger.info(f"Creating/verifying stream: {STREAM_NAME}")
            await consumer.create_stream(
                STREAM_NAME, exists_ok=True, arguments={"max-length-bytes": STREAM_RETENTION}
            )

            # Message callback - simplified like working receive.py
            async def on_message(msg: AMQPMessage, message_context: MessageContext):
                try:
                    stream = message_context.consumer.get_stream(message_context.subscriber_name)

                    # Print like working receive.py - simple and direct
                    print("Got message: {} from stream: {}".format(msg, stream))

                except Exception as e:
                    logger.error(f"Error processing message: {e}")

            logger.info(f"Successfully connected to {RABBITMQ_HOST}")
            logger.info("Starting consumer...")
            await consumer.start()

            logger.info(f"Subscribing to stream '{STREAM_NAME}' with FIRST offset...")
            await consumer.subscribe(
                stream=STREAM_NAME,
                callback=on_message,
                offset_specification=ConsumerOffsetSpecification(OffsetType.FIRST, None),
            )

            logger.info("Consumer subscribed successfully - waiting for messages...")
            logger.info("Press control + C to close")

            try:
                await consumer.run()
            except (KeyboardInterrupt, asyncio.CancelledError):
                print("Closing Consumer...")
                return

    except Exception as e:
        logger.error(f"Failed to connect to {RABBITMQ_HOST}: {e}")


if __name__ == '__main__':
    logger.info("Starting Plot Consumer - Simplified like working receive.py")
    logger.info(f"RabbitMQ Host: {RABBITMQ_HOST}")
    logger.info(f"RabbitMQ Port: {RABBITMQ_PORT}")
    logger.info(f"Stream Name: {STREAM_NAME}")
    logger.info(f"Username: {RABBITMQ_USERNAME}")
    logger.info("Using FIRST offset - will read all messages from the beginning of the stream")

    with asyncio.Runner() as runner:
        runner.run(receive())

then the sender:

import asyncio
import time

from rstream import Producer, ConfirmationStatus, AMQPMessage

STREAM_NAME = "hello-python-stream"
# 5GB
STREAM_RETENTION = 5000000000

async def _on_publish_confirm_client(confirmation: ConfirmationStatus) -> None:
    if confirmation.is_confirmed:
        print("message id: {} is confirmed".format(confirmation.message_id))
    else:
        print(
            "message id: {} not confirmed. Response code {}".format(
                confirmation.message_id, confirmation.response_code
            )
        )

async def send():
    async with Producer(
        host="localhost",
        username="guest",
        password="guest",
        port=5553,
        load_balancer_mode=True,
    ) as producer:
      
        for i in range(1000):
            amqp_message = AMQPMessage(
                body=bytes("My AMQP message: {}".format(i), "utf-8"),
            )
            # send is asynchronous
            await producer.send(stream=STREAM_NAME, message=amqp_message, on_publish_confirm=_on_publish_confirm_client)
            await asyncio.sleep(1)
            print("sent message: {}".format(amqp_message.body))

        print(" [x] Hello, World! message sent")

        input(" [x] Press Enter to close the producer  ...")

asyncio.run(send())

It works

I did not try in k8s. This test should be enough.
Let me know!

[thoefkens]

thank you so much - I will test this and get back to you