Synchronous API calls between microservices create tight coupling. Service A calls Service B and waits. If Service B is slow, Service A is slow. If Service B is down, Service A fails. If traffic spikes, both services need to scale simultaneously or requests start timing out. This is a fragile architecture for anything beyond simple workloads.
Message queues decouple producers from consumers. Service A puts a message in a queue and moves on. Service B reads from the queue at its own pace. If Service B is temporarily unavailable, messages accumulate in the queue and get processed when it comes back. If traffic spikes, the queue absorbs the burst and consumers process it without the producer being affected.
OCI Queue is Oracle’s managed message queue service. It is HTTP-based, requires no client libraries beyond a standard HTTP client, supports at-least-once delivery, and integrates natively with OCI IAM, Monitoring, and Functions. This post covers deploying OCI Queue with Terraform, building producers and consumers in Python, configuring dead letter queues, and handling visibility timeouts correctly.
Architecture
Order Service (producer)
|
OCI Queue (orders-processing-queue)
|
-------------------------
| |
Notification Inventory
Consumer Consumer
| |
OCI Notifications OCI Autonomous DB
|
Dead Letter Queue
(failed messages - manual review)Step 1: IAM Policy
resource "oci_identity_dynamic_group" "queue_consumers_dg" {
compartment_id = var.tenancy_ocid
name = "queue-consumer-instances"
description = "Compute instances and Functions that consume from OCI Queue"
matching_rule = "Any {instance.compartment.id = '${var.compartment_id}', resource.type = 'fnfunc'}"
}
resource "oci_identity_policy" "queue_policy" {
compartment_id = var.compartment_id
name = "oci-queue-policy"
description = "Permissions for OCI Queue producers and consumers"
statements = [
# Producers can put messages
"Allow group ${var.producer_group_name} to use queues in compartment id ${var.compartment_id}",
# Consumer instances can get, delete, and update messages
"Allow dynamic-group queue-consumer-instances to use queues in compartment id ${var.compartment_id}",
"Allow dynamic-group queue-consumer-instances to read queues in compartment id ${var.compartment_id}"
]
}Step 2: Provision the Queue and Dead Letter Queue
# Dead Letter Queue - receives messages that fail processing repeatedly
resource "oci_queue_queue" "orders_dlq" {
compartment_id = var.compartment_id
display_name = "orders-processing-dlq"
visibility_in_seconds = 30
timeout_in_seconds = 30
retention_in_seconds = 604800 # 7 days
max_in_flight_messages = 100
defined_tags = {
"Operations.Environment" = "production"
"Operations.Purpose" = "dead-letter-queue"
"Operations.ManagedBy" = "terraform"
}
}
# Main processing queue
resource "oci_queue_queue" "orders_queue" {
compartment_id = var.compartment_id
display_name = "orders-processing-queue"
visibility_in_seconds = 60
timeout_in_seconds = 30
retention_in_seconds = 86400 # 24 hours
max_in_flight_messages = 500
# After 3 failed delivery attempts, move to DLQ
dead_letter_queue_delivery_count = 3
custom_encryption_key_id = var.vault_key_id
defined_tags = {
"Operations.Environment" = "production"
"Operations.Application" = "orders"
"Operations.ManagedBy" = "terraform"
}
}
output "queue_endpoint" {
value = oci_queue_queue.orders_queue.messages_endpoint
description = "HTTP endpoint for producing and consuming messages"
}
output "queue_id" {
value = oci_queue_queue.orders_queue.id
}The visibility_in_seconds = 60 setting is critical. When a consumer reads a message, OCI Queue hides that message from other consumers for 60 seconds. If the consumer processes the message successfully and deletes it within that window, it is gone. If the consumer crashes or fails to delete the message, OCI Queue makes it visible again after 60 seconds for another consumer to pick up. Set this to comfortably exceed your maximum expected processing time for a single message.
The dead_letter_queue_delivery_count = 3 means after three failed visibility timeout expirations for the same message, OCI Queue moves it to the DLQ automatically. This prevents poison pill messages from blocking your queue forever.
Step 3: Producer Implementation
import oci
import json
import logging
import base64
from datetime import datetime, timezone
from typing import List, Optional
logger = logging.getLogger(__name__)
class OrderQueueProducer:
def __init__(self, queue_id: str, messages_endpoint: str):
self.queue_id = queue_id
self.messages_endpoint = messages_endpoint
config = oci.config.from_file()
self.client = oci.queue.QueueClient(
config,
service_endpoint=messages_endpoint
)
def publish_order_event(self, order_id: str, event_type: str, payload: dict) -> str:
message_body = json.dumps({
"event_type": event_type,
"order_id": order_id,
"payload": payload,
"published_at": datetime.now(timezone.utc).isoformat(),
"version": "1.0"
})
put_details = oci.queue.models.PutMessagesDetails(
messages=[
oci.queue.models.PutMessagesDetailsEntry(
content=base64.b64encode(message_body.encode()).decode()
)
]
)
response = self.client.put_messages(
queue_id=self.queue_id,
put_messages_details=put_details
)
message_id = response.data.messages[0].id
logger.info(f"Published {event_type} for order {order_id}, message_id: {message_id}")
return message_id
def publish_batch(self, events: List[dict]) -> List[str]:
if len(events) > 20:
raise ValueError("OCI Queue supports maximum 20 messages per batch")
entries = [
oci.queue.models.PutMessagesDetailsEntry(
content=base64.b64encode(
json.dumps(event).encode()
).decode()
)
for event in events
]
put_details = oci.queue.models.PutMessagesDetails(messages=entries)
response = self.client.put_messages(
queue_id=self.queue_id,
put_messages_details=put_details
)
message_ids = [msg.id for msg in response.data.messages]
logger.info(f"Published batch of {len(message_ids)} messages")
return message_idsStep 4: Consumer Implementation with Visibility Timeout Renewal
The consumer reads messages, processes them, and deletes them. For long-running processing tasks, the consumer must renew the visibility timeout before it expires, otherwise OCI Queue makes the message visible again and another consumer picks it up, causing duplicate processing.
import oci
import json
import base64
import logging
import time
import threading
from typing import Callable, Optional
logger = logging.getLogger(__name__)
class OrderQueueConsumer:
def __init__(
self,
queue_id: str,
messages_endpoint: str,
visibility_in_seconds: int = 60
):
self.queue_id = queue_id
self.messages_endpoint = messages_endpoint
self.visibility_in_seconds = visibility_in_seconds
self.running = False
config = oci.config.from_file()
self.client = oci.queue.QueueClient(
config,
service_endpoint=messages_endpoint
)
def _renew_visibility(self, receipt: str, stop_event: threading.Event):
renewal_interval = self.visibility_in_seconds * 0.7
while not stop_event.wait(timeout=renewal_interval):
try:
self.client.update_message(
queue_id=self.queue_id,
message_receipt=receipt,
update_message_details=oci.queue.models.UpdateMessageDetails(
visibility_in_seconds=self.visibility_in_seconds
)
)
logger.debug(f"Renewed visibility for receipt {receipt[:20]}...")
except Exception as e:
logger.warning(f"Failed to renew visibility: {e}")
break
def process_message(self, message, handler: Callable) -> bool:
content = json.loads(base64.b64decode(message.content).decode())
receipt = message.receipt
stop_renewal = threading.Event()
renewal_thread = threading.Thread(
target=self._renew_visibility,
args=(receipt, stop_renewal),
daemon=True
)
renewal_thread.start()
try:
handler(content)
self.client.delete_message(
queue_id=self.queue_id,
message_receipt=receipt
)
logger.info(f"Processed and deleted message, event: {content.get('event_type')}")
return True
except Exception as e:
logger.error(f"Failed to process message: {e}. Message will become visible again.")
return False
finally:
stop_renewal.set()
renewal_thread.join(timeout=2)
def consume(self, handler: Callable, max_messages: int = 10):
self.running = True
logger.info(f"Starting consumer for queue {self.queue_id}")
while self.running:
try:
response = self.client.get_messages(
queue_id=self.queue_id,
limit=max_messages,
visibility_in_seconds=self.visibility_in_seconds,
timeout_in_seconds=20 # long polling
)
messages = response.data.messages
if not messages:
continue
logger.info(f"Received {len(messages)} messages")
for message in messages:
self.process_message(message, handler)
except oci.exceptions.ServiceError as e:
logger.error(f"OCI Queue service error: {e}")
time.sleep(5)
except KeyboardInterrupt:
logger.info("Consumer shutting down")
self.running = False
def stop(self):
self.running = FalseStep 5: The Message Handler
def handle_order_event(event: dict):
event_type = event.get("event_type")
order_id = event.get("order_id")
payload = event.get("payload", {})
logger.info(f"Processing {event_type} for order {order_id}")
if event_type == "ORDER_PLACED":
_handle_order_placed(order_id, payload)
elif event_type == "ORDER_PAID":
_handle_order_paid(order_id, payload)
elif event_type == "ORDER_CANCELLED":
_handle_order_cancelled(order_id, payload)
else:
logger.warning(f"Unknown event type: {event_type}. Skipping.")
def _handle_order_placed(order_id: str, payload: dict):
# Reserve inventory
items = payload.get("items", [])
for item in items:
reserve_inventory(item["sku"], item["quantity"])
# Send confirmation notification
send_notification(
customer_id=payload["customer_id"],
template="order_confirmation",
data={"order_id": order_id, "items": items}
)
def _handle_order_paid(order_id: str, payload: dict):
update_order_status(order_id, "PROCESSING")
trigger_fulfillment(order_id, payload["shipping_address"])
def _handle_order_cancelled(order_id: str, payload: dict):
release_inventory(order_id)
if payload.get("refund_required"):
initiate_refund(order_id, payload["payment_reference"])
if __name__ == "__main__":
import os
consumer = OrderQueueConsumer(
queue_id=os.environ["QUEUE_ID"],
messages_endpoint=os.environ["QUEUE_ENDPOINT"],
visibility_in_seconds=60
)
consumer.consume(handler=handle_order_event, max_messages=10)Step 6: Dead Letter Queue Monitor
Messages in the DLQ represent processing failures that need investigation. Set up a monitoring alarm so the team is notified when messages land in the DLQ, and build a separate process to inspect them.
resource "oci_monitoring_alarm" "dlq_messages_alarm" {
compartment_id = var.compartment_id
display_name = "orders-dlq-has-messages"
is_enabled = true
metric_compartment_id = var.compartment_id
namespace = "oci_queue"
query = "VisibleMessages[5m]{queueId = '${oci_queue_queue.orders_dlq.id}'}.max() > 0"
severity = "CRITICAL"
pending_duration = "PT5M"
destinations = [var.ops_notification_topic_id]
body = "Messages have arrived in the orders DLQ. These represent failed processing attempts that need manual investigation."
}
resource "oci_monitoring_alarm" "queue_depth_alarm" {
compartment_id = var.compartment_id
display_name = "orders-queue-depth-high"
is_enabled = true
metric_compartment_id = var.compartment_id
namespace = "oci_queue"
query = "VisibleMessages[5m]{queueId = '${oci_queue_queue.orders_queue.id}'}.max() > 1000"
severity = "WARNING"
pending_duration = "PT10M"
destinations = [var.ops_notification_topic_id]
body = "Orders queue depth exceeded 1000 messages. Consumers may be falling behind. Consider scaling up consumer instances."
}To inspect DLQ messages without deleting them, read with a long visibility timeout, log the content for investigation, and then choose whether to reprocess or discard each one:
def inspect_dlq(dlq_id: str, messages_endpoint: str):
config = oci.config.from_file()
client = oci.queue.QueueClient(config, service_endpoint=messages_endpoint)
response = client.get_messages(
queue_id=dlq_id,
limit=10,
visibility_in_seconds=300, # 5 minutes to inspect before it becomes visible again
timeout_in_seconds=5
)
for msg in response.data.messages:
content = json.loads(base64.b64decode(msg.content).decode())
logger.error(
f"DLQ message - "
f"event_type: {content.get('event_type')}, "
f"order_id: {content.get('order_id')}, "
f"receipt: {msg.receipt[:30]}..."
)
return response.data.messagesStep 7: Deploying Consumers on OKE
Run queue consumers as Kubernetes Deployments on OKE. Each pod runs one consumer process. Scale the deployment based on queue depth using the OCI Queue metrics exposed to the Kubernetes HPA through a custom metrics adapter.
apiVersion: apps/v1
kind: Deployment
metadata:
name: orders-queue-consumer
namespace: orders
spec:
replicas: 3
selector:
matchLabels:
app: orders-queue-consumer
template:
metadata:
labels:
app: orders-queue-consumer
spec:
serviceAccountName: queue-consumer-sa
containers:
- name: consumer
image: <region>.ocir.io/<namespace>/orders-consumer:latest
env:
- name: QUEUE_ID
valueFrom:
secretKeyRef:
name: queue-config
key: queue-id
- name: QUEUE_ENDPOINT
valueFrom:
secretKeyRef:
name: queue-config
key: messages-endpoint
resources:
requests:
cpu: "250m"
memory: "256Mi"
limits:
cpu: "500m"
memory: "512Mi"
livenessProbe:
exec:
command: ["python", "-c", "import os; os.kill(1, 0)"]
initialDelaySeconds: 30
periodSeconds: 30Key Operational Considerations
Idempotency is mandatory. OCI Queue guarantees at-least-once delivery, which means the same message can be delivered more than once. Your handler must be idempotent: processing the same order event twice should produce the same result as processing it once. Use the order_id and event_type combination as an idempotency key and check a processed-events store before executing any state-changing operation.
Message ordering is not guaranteed within a single queue. If your use case requires strict ordering of events for the same order, include a sequence number in the message payload and handle out-of-order delivery in the consumer logic.
The visibility timeout must be longer than your maximum processing time. If processing occasionally takes 45 seconds, set the visibility timeout to at least 90 seconds, not 60. A message that expires its visibility timeout before deletion gets reprocessed, which triggers your idempotency logic and wastes consumer capacity.
Long polling reduces empty receive calls. Setting timeout_in_seconds=20 in the get_messages call tells OCI Queue to wait up to 20 seconds for a message before returning an empty response. Without long polling, a quiet queue generates a constant stream of empty API calls consuming CPU and generating unnecessary network traffic.
Regards,
Osama
#OCI #OracleCloud #Microservices #MessageQueue #Terraform #CloudNative #InfrastructureAsCode #IaC #OracleCloudInfrastructure #PlatformEngineering #CloudArchitecture #DevOps #DistributedSystems #BackendEngineering #TechBlog #AsyncMessaging #CloudEngineering #Oracle #OCIQueue #EventDriven
Technology Geek 
Leave a comment