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
|
Dead Letter Queue
(failed messages)Step 1: IAM Policy
resource "oci_identity_dynamic_group" "queue_consumers_dg" {
compartment_id = var.tenancy_ocid
name = "queue-consumer-instances"
description = "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 = [
"Allow group ${var.producer_group_name} to use queues in compartment id ${var.compartment_id}",
"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 Queue and Dead Letter Queue
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
max_in_flight_messages = 100
defined_tags = {
"Operations.Environment" = "production"
"Operations.Purpose" = "dead-letter-queue"
"Operations.ManagedBy" = "terraform"
}
}
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
max_in_flight_messages = 500
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
}
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 it from other consumers for 60 seconds. If the consumer processes and deletes it within that window, it is gone. If the consumer crashes, the message becomes visible again after 60 seconds. The dead_letter_queue_delivery_count = 3 means after three failed visibility timeout expirations, OCI Queue moves the message to the DLQ automatically.
Step 3: Producer Implementation
import oci
import json
import logging
import base64
from datetime import datetime, timezone
from typing import List
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
]
response = self.client.put_messages(
queue_id=self.queue_id,
put_messages_details=oci.queue.models.PutMessagesDetails(messages=entries)
)
return [msg.id for msg in response.data.messages]Step 4: Consumer with Visibility Timeout Renewal
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
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: {content.get('event_type')}")
return True
except Exception as e:
logger.error(f"Processing failed: {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
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
)
for message in response.data.messages:
self.process_message(message, handler)
except oci.exceptions.ServiceError as e:
logger.error(f"Queue service error: {e}")
time.sleep(5)
except KeyboardInterrupt:
self.running = FalseStep 5: Dead Letter Queue Monitoring
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 requiring 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."
}Step 6: Deploying Consumers on OKE
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"Key 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 must 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. 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 exceed your maximum processing time. If processing occasionally takes 45 seconds, set the timeout to at least 90 seconds. A message that expires its visibility timeout before deletion gets reprocessed, triggering your idempotency logic and wasting 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.
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