Every production system generates events: orders placed, payments processed, errors thrown, user actions taken. The question is not whether events exist but what you do with them. If they go nowhere, you have no real-time visibility and no way to drive downstream processes from upstream state changes. If they go into a queue, you get decoupling but limited replay and fan-out. If they go into a streaming platform, you get all of that plus ordered, replayable, partitioned event streams that multiple consumers can read independently at their own pace.
OCI Streaming is Oracle’s managed Apache Kafka-compatible streaming service. It runs inside your tenancy, integrates with OCI IAM, connects natively to OCI Functions and Service Connector Hub, and exposes a Kafka-compatible endpoint so existing Kafka producers and consumers work without code changes. This post covers deploying OCI Streaming with Terraform, writing producers and consumers in Python, connecting streams to downstream OCI services, and monitoring throughput and lag.
Architecture
Orders API (producer)
|
OCI Streaming (orders-events stream, 3 partitions)
|
-----------------------------------
| | |
Notification Analytics Audit Log
Consumer Consumer Consumer
(Function) (Data Flow) (Object Storage via SCH)Three independent consumer groups read the same stream simultaneously. Each maintains its own offset. The notification consumer processes events in real time. The analytics consumer feeds a batch processing job. The audit consumer archives everything to Object Storage via Service Connector Hub. None of them interfere with each other.
Step 1: IAM Policy
resource "oci_identity_dynamic_group" "streaming_producers_dg" {
compartment_id = var.tenancy_ocid
name = "streaming-producer-instances"
description = "Instances and Functions that produce to OCI Streaming"
matching_rule = "Any {instance.compartment.id = '${var.compartment_id}', resource.type = 'fnfunc'}"
}
resource "oci_identity_dynamic_group" "streaming_consumers_dg" {
compartment_id = var.tenancy_ocid
name = "streaming-consumer-instances"
description = "Instances and Functions that consume from OCI Streaming"
matching_rule = "Any {instance.compartment.id = '${var.compartment_id}', resource.type = 'fnfunc'}"
}
resource "oci_identity_policy" "streaming_policy" {
compartment_id = var.compartment_id
name = "oci-streaming-policy"
description = "Permissions for OCI Streaming producers and consumers"
statements = [
"Allow dynamic-group streaming-producer-instances to use stream-push in compartment id ${var.compartment_id}",
"Allow dynamic-group streaming-consumer-instances to use stream-pull in compartment id ${var.compartment_id}",
"Allow dynamic-group streaming-consumer-instances to use stream-family in compartment id ${var.compartment_id}",
"Allow service sch to use stream-push in compartment id ${var.compartment_id}",
"Allow service sch to use stream-pull in compartment id ${var.compartment_id}"
]
}Step 2: Stream Pool and Stream
resource "oci_streaming_stream_pool" "orders_pool" {
compartment_id = var.compartment_id
name = "orders-stream-pool"
kafka_settings {
auto_create_topics_enable = false
log_retention_hours = 24
num_partitions = 3
}
custom_encryption_key {
kms_key_id = var.vault_key_id
}
private_endpoint_settings {
subnet_id = var.private_subnet_id
private_endpoint_ip = "10.0.2.50"
nsg_ids = [var.streaming_nsg_id]
}
defined_tags = {
"Operations.Environment" = "production"
"Operations.ManagedBy" = "terraform"
}
}
resource "oci_streaming_stream" "orders_events" {
compartment_id = var.compartment_id
name = "orders-events"
partitions = 3
retention_in_hours = 24
stream_pool_id = oci_streaming_stream_pool.orders_pool.id
defined_tags = {
"Operations.Environment" = "production"
"Operations.Application" = "orders"
}
}
output "stream_id" {
value = oci_streaming_stream.orders_events.id
}
output "stream_endpoint" {
value = oci_streaming_stream_pool.orders_pool.endpoint_fqdn
}The private_endpoint_settings block places the stream pool inside your VCN. Producers and consumers connect to the private endpoint IP rather than the public OCI Streaming endpoint. No streaming traffic leaves your network.
Three partitions allow up to three parallel consumer instances to process messages concurrently within the same consumer group. A partition is the unit of parallelism in Kafka and OCI Streaming. One consumer per partition is the maximum effective parallelism.
Step 3: Producer Implementation
import oci
import json
import base64
import logging
import hashlib
from datetime import datetime, timezone
from typing import List
logger = logging.getLogger(__name__)
class OrderEventProducer:
def __init__(self, stream_id: str, stream_endpoint: str):
self.stream_id = stream_id
self.stream_endpoint = stream_endpoint
config = oci.config.from_file()
self.client = oci.streaming.StreamClient(
config,
service_endpoint=f"https://{stream_endpoint}"
)
def publish(self, order_id: str, event_type: str, payload: dict) -> str:
message = json.dumps({
"event_type": event_type,
"order_id": order_id,
"payload": payload,
"published_at": datetime.now(timezone.utc).isoformat(),
"version": "1.0"
})
# Use order_id as the partition key so all events
# for the same order land on the same partition in order
partition_key = hashlib.md5(order_id.encode()).hexdigest()[:8]
response = self.client.put_messages(
stream_id=self.stream_id,
put_messages_details=oci.streaming.models.PutMessagesDetails(
messages=[
oci.streaming.models.PutMessagesDetailsEntry(
key=base64.b64encode(partition_key.encode()).decode(),
value=base64.b64encode(message.encode()).decode()
)
]
)
)
if response.data.failures > 0:
logger.error(f"Failed to publish event: {response.data.entries[0].error_message}")
raise RuntimeError(f"Stream publish failed: {response.data.entries[0].error_message}")
offset = response.data.entries[0].offset
logger.info(f"Published {event_type} for order {order_id} at offset {offset}")
return str(offset)
def publish_batch(self, events: List[dict]) -> List[str]:
entries = [
oci.streaming.models.PutMessagesDetailsEntry(
key=base64.b64encode(
hashlib.md5(e["order_id"].encode()).hexdigest()[:8].encode()
).decode(),
value=base64.b64encode(json.dumps(e).encode()).decode()
)
for e in events
]
response = self.client.put_messages(
stream_id=self.stream_id,
put_messages_details=oci.streaming.models.PutMessagesDetails(messages=entries)
)
offsets = [str(entry.offset) for entry in response.data.entries]
logger.info(f"Published batch of {len(offsets)} events")
return offsetsThe partition key derived from order_id ensures all events for the same order are written to the same partition. This guarantees ordering: a consumer reading partition 0 sees all events for the orders that hash to partition 0 in the sequence they were written. Without a consistent partition key, events for the same order can land on different partitions and arrive at the consumer out of order.
Step 4: Consumer Implementation with Offset Management
import oci
import json
import base64
import logging
import time
from typing import Callable, List
logger = logging.getLogger(__name__)
class OrderEventConsumer:
def __init__(
self,
stream_id: str,
stream_endpoint: str,
group_name: str,
instance_name: str
):
self.stream_id = stream_id
self.stream_endpoint = stream_endpoint
self.group_name = group_name
self.instance_name = instance_name
self.running = False
config = oci.config.from_file()
self.client = oci.streaming.StreamClient(
config,
service_endpoint=f"https://{stream_endpoint}"
)
def _get_or_create_cursor(self) -> str:
response = self.client.create_group_cursor(
stream_id=self.stream_id,
create_group_cursor_details=oci.streaming.models.CreateGroupCursorDetails(
group_name=self.group_name,
instance_name=self.instance_name,
type=oci.streaming.models.CreateGroupCursorDetails.TYPE_TRIM_HORIZON,
commit_on_get=False
)
)
return response.data.value
def _commit_offsets(self, cursor: str, messages: list):
offsets = [
oci.streaming.models.CursorDetails(
partition=msg.partition,
offset=msg.offset
)
for msg in messages
]
self.client.consumer_commit(
stream_id=self.stream_id,
cursor=cursor,
cursor_details=oci.streaming.models.CursorDetails()
)
def consume(self, handler: Callable, batch_size: int = 100):
self.running = True
cursor = self._get_or_create_cursor()
logger.info(f"Consumer {self.group_name}/{self.instance_name} started")
while self.running:
try:
response = self.client.get_messages(
stream_id=self.stream_id,
cursor=cursor,
limit=batch_size
)
messages = response.data
cursor = response.headers["opc-next-cursor"]
if not messages:
time.sleep(1)
continue
processed = []
for msg in messages:
try:
content = json.loads(
base64.b64decode(msg.value).decode()
)
handler(content)
processed.append(msg)
except Exception as e:
logger.error(f"Failed to process message at offset {msg.offset}: {e}")
if processed:
self.client.consumer_heartbeat(
stream_id=self.stream_id,
cursor=cursor
)
logger.debug(f"Processed {len(processed)}/{len(messages)} messages")
except oci.exceptions.ServiceError as e:
if e.status == 404:
logger.warning("Cursor expired, recreating")
cursor = self._get_or_create_cursor()
else:
logger.error(f"Stream service error: {e}")
time.sleep(5)
except KeyboardInterrupt:
self.running = FalseThe consumer uses TYPE_TRIM_HORIZON on first connection, which starts reading from the oldest available message. On reconnect, the group cursor picks up from where it left off. The consumer_heartbeat call keeps the cursor alive during long processing batches. If the cursor goes without a heartbeat for too long, OCI Streaming expires it and the consumer must recreate it, which triggers the 404 handling in the except block.
Step 5: Service Connector Hub for Archival
Service Connector Hub connects OCI Streaming to Object Storage without writing any consumer code. Every message published to the stream is automatically archived to a bucket, partitioned by date and hour.
resource "oci_sch_service_connector" "stream_to_storage" {
compartment_id = var.compartment_id
display_name = "orders-events-archive"
description = "Archive all orders events from Streaming to Object Storage"
source {
kind = "streaming"
stream_id = oci_streaming_stream.orders_events.id
cursor {
kind = "TRIM_HORIZON"
}
}
target {
kind = "objectStorage"
bucket_name = var.archive_bucket_name
namespace = var.object_storage_namespace
object_name_prefix = "orders-events"
batch_size_in_kbs = 1024
batch_time_in_sec = 60
}
defined_tags = {
"Operations.Environment" = "production"
"Operations.ManagedBy" = "terraform"
}
}
resource "oci_sch_service_connector" "stream_to_function" {
compartment_id = var.compartment_id
display_name = "orders-events-processor"
description = "Route orders events to notification Function"
source {
kind = "streaming"
stream_id = oci_streaming_stream.orders_events.id
cursor {
kind = "LATEST"
}
}
target {
kind = "functions"
function_id = var.notification_function_id
batch_size_in_num = 100
batch_time_in_sec = 10
}
}Step 6: Monitoring Stream Lag and Throughput
resource "oci_monitoring_alarm" "stream_put_failures" {
compartment_id = var.compartment_id
display_name = "orders-stream-put-failures"
is_enabled = true
metric_compartment_id = var.compartment_id
namespace = "oci_streaming"
query = "PutMessageFailures[5m]{streamId = '${oci_streaming_stream.orders_events.id}'}.sum() > 10"
severity = "CRITICAL"
pending_duration = "PT2M"
destinations = [var.ops_notification_topic_id]
body = "Orders stream is experiencing publish failures. Producers may be losing events. Check application logs immediately."
}
resource "oci_monitoring_alarm" "stream_consumer_lag" {
compartment_id = var.compartment_id
display_name = "orders-stream-consumer-lag-high"
is_enabled = true
metric_compartment_id = var.compartment_id
namespace = "oci_streaming"
query = "GetMessagesCount[5m]{streamId = '${oci_streaming_stream.orders_events.id}'}.mean() < 1"
severity = "WARNING"
pending_duration = "PT10M"
destinations = [var.ops_notification_topic_id]
body = "Orders stream consumer appears idle for 10 minutes. Consumer group may have stopped processing."
}Using the Kafka-Compatible Endpoint
If you already have Kafka producers or consumers in your codebase, point them at the OCI Streaming Kafka-compatible endpoint without changing any application code. The only changes are the bootstrap server, the SASL credentials, and the topic name.
from kafka import KafkaProducer, KafkaConsumer
import json
# OCI Streaming Kafka-compatible connection config
KAFKA_CONFIG = {
"bootstrap_servers": f"{STREAM_POOL_ENDPOINT}:9092",
"security_protocol": "SASL_SSL",
"sasl_mechanism": "PLAIN",
"sasl_plain_username": f"{TENANCY_NAME}/{USERNAME}/{STREAM_POOL_OCID}",
"sasl_plain_password": AUTH_TOKEN,
"api_version": (0, 10, 1),
}
# Producer - identical to any Kafka producer
producer = KafkaProducer(
**KAFKA_CONFIG,
value_serializer=lambda v: json.dumps(v).encode("utf-8")
)
producer.send(
topic="orders-events",
key=b"ORD-001",
value={"event_type": "ORDER_PLACED", "order_id": "ORD-001"}
)
producer.flush()
# Consumer - identical to any Kafka consumer
consumer = KafkaConsumer(
"orders-events",
**KAFKA_CONFIG,
group_id="notification-consumer-group",
auto_offset_reset="earliest",
value_deserializer=lambda v: json.loads(v.decode("utf-8"))
)
for message in consumer:
print(f"Partition: {message.partition}, Offset: {message.offset}")
print(f"Event: {message.value}")The SASL username format for OCI Streaming Kafka endpoint is tenancy-name/username/stream-pool-ocid. The password is an OCI Auth Token, not your console password. Generate one under your user profile in the OCI Console under Auth Tokens.
Operational Notes
Partition count is fixed at creation time. You cannot increase partitions on an existing stream without creating a new stream and migrating consumers. Size your partition count based on your expected peak throughput and maximum desired consumer parallelism, then add a buffer. Three partitions work for most workloads under 3000 messages per second. Scale to 10 or more for high-throughput pipelines.
Retention is 24 hours by default and configurable up to 7 days. Set retention based on your recovery window. If a consumer goes down for a weekend, 24 hours retention means it loses events from Saturday evening onward. 7 days gives you the full weekend to recover without data loss.
Always use a partition key derived from a business entity ID, not a random value. Random keys distribute load evenly but destroy ordering guarantees. Business entity keys preserve ordering per entity while still distributing across partitions when the entity population is large enough.
Regards,
Osama
#OCI #OracleCloud #Streaming #EventDriven #Terraform #CloudNative #IaC #OracleCloudInfrastructure #PlatformEngineering #CloudArchitecture #DevOps #RealTime #BackendEngineering #TechBlog #Oracle #Kafka #MessageStreaming #ServiceConnectorHub #CloudEngineering #DistributedSystems
Technology Geek 
Leave a comment