Integration  

{
"first_name": "Peace Smith",
"customer_id": "6808568926651744",
"email": "peace@peacemaker.com",
"phone": "+1234567890",
"last_name": "Moe",
"updated_at": "1759752209000",
"user_attributes": {
"city": "San Francisco",
"subscription_tier": "Premium",
"total_orders": 5
},
"event_attributes": {
"event_name": "Purchase Completed",
"product_id": "PROD-12345",
"amount": 99.99,
"currency": "USD"
}
}

pip install confluent-kafka requests python-dotenv
python --version

# MoEngage Configuration (from Step 1 above)
MOENGAGE_PARTNER_NAME=your_partner_name
MOENGAGE_DATA_CENTER=02
MOENGAGE_CONFIG_NAME=15fc62d8-efbd-42c7-aad5-ad723a83ae80
MOENGAGE_APP_ID=7IYSTOK1CLO9A1XDO...
MOENGAGE_API_SECRET=pIfghD6guNHTvwgZz...

Kafka Configuration (your Kafka cluster details)

KAFKA_BOOTSTRAP_SERVERS=broker1:9092,broker2:9092
KAFKA_TOPICS=user-events,transactions
KAFKA_GROUP_ID=moengage-consumer-group
KAFKA_SECURITY_PROTOCOL=PLAINTEXT

For authenticated Kafka (if using SASL)

KAFKA_SECURITY_PROTOCOL=SASL_SSL
KAFKA_SASL_MECHANISM=PLAIN
KAFKA_API_KEY=your_kafka_api_key
KAFKA_API_SECRET=your_kafka_api_secret

Application Configuration

LOG_LEVEL=INFO

#!/usr/bin/env python3
"""
Kafka to MoEngage Consumer
Reads events from Kafka and sends to MoEngage using standard JSON format
"""
import os
import json
import requests
import time
import base64
from dotenv import load_dotenv
from confluent_kafka import Consumer, KafkaError

load_dotenv()

Configuration

MOENGAGE_PARTNER_NAME = os.getenv('MOENGAGE_PARTNER_NAME')
MOENGAGE_DATA_CENTER = os.getenv('MOENGAGE_DATA_CENTER')
MOENGAGE_CONFIG_NAME = os.getenv('MOENGAGE_CONFIG_NAME')
MOENGAGE_APP_ID = os.getenv('MOENGAGE_APP_ID')
MOENGAGE_API_SECRET = os.getenv('MOENGAGE_API_SECRET')

KAFKA_BOOTSTRAP_SERVERS = os.getenv('KAFKA_BOOTSTRAP_SERVERS')
KAFKA_TOPICS = os.getenv('KAFKA_TOPICS', 'events').split(',')
KAFKA_GROUP_ID = os.getenv('KAFKA_GROUP_ID')
KAFKA_SECURITY_PROTOCOL = os.getenv('KAFKA_SECURITY_PROTOCOL', 'PLAINTEXT')
KAFKA_SASL_MECHANISM = os.getenv('KAFKA_SASL_MECHANISM', 'PLAIN')
KAFKA_API_KEY = os.getenv('KAFKA_API_KEY')
KAFKA_API_SECRET = os.getenv('KAFKA_API_SECRET')

LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')

MoEngage endpoint with configName

MOENGAGE_ENDPOINT = (
f'https://api-{MOENGAGE_DATA_CENTER}.moengage.com/v1/partner/'
f'{MOENGAGE_PARTNER_NAME}/events/?configName={MOENGAGE_CONFIG_NAME}'
)

def log(level, msg):
"""Simple logging"""
if level in ['ERROR', 'WARNING', 'INFO', 'DEBUG']:
print(f"[{level}] {msg}")

def create_basic_auth():
"""Create Basic Auth header"""
credentials = f"{MOENGAGE_APP_ID}:{MOENGAGE_API_SECRET}"
encoded = base64.b64encode(credentials.encode()).decode()
return f"Basic {encoded}"

def send_to_moengage(kafka_message):
"""Send Kafka message to MoEngage with retries"""
headers = {
'Content-Type': 'application/json',
'Authorization': create_basic_auth()
}

for attempt in range(3):
    try:
        response = requests.post(
            MOENGAGE_ENDPOINT,
            json=kafka_message,
            headers=headers,
            timeout=10
        )
        
        if response.status_code in [200, 201, 202, 204, 207]:
            log('INFO', f"Event sent successfully (status {response.status_code})")
            return True
        elif response.status_code = 500:
            log('WARNING', f"Server error {response.status_code}, retrying...")
            time.sleep(2 ** attempt)
            continue
        else:
            log('ERROR', f"API error {response.status_code}: {response.text}")
            return False
    
    except Exception as e:
        log('ERROR', f"Send failed (attempt {attempt+1}): {e}")
        if attempt <2: return="return" false="False" def="def" main="main" consumer="Consumer" loop="loop" conf="{" earliest="earliest" if="if" kafka_security_protocol="KAFKA_SECURITY_PROTOCOL" info="INFO" f="f" moengage="MoEngage" while="while" msg="msg" is="is" continue="continue" error="ERROR" kafka="Kafka" decode="Decode" message="message" kafka_message="msg.value().decode(" utf-8="utf-8" payload="payload" validate="Validate" required="required" fields="fields" customer_id="customer_id" not="not" in="in" missing="Missing" event_attributes="event_attributes" or="or" event_name="event_name" send="Send" to="to" debug="DEBUG" offset="Offset" except="except" invalid="Invalid" json="JSON" skipping="skipping" exception="Exception" as="as" processing="Processing" shutdown="Shutdown" requested="requested" name="=" environment="environment" variables="variables">

python3 kafka_consumer.py

[INFO] Consumer started. Topics: ['user-events']
[INFO] MoEngage endpoint: https://api-02.moengage.com/v1/partner/your-company/events/?configName=15fc62d8...
[INFO] Event sent successfully (status 200)
[DEBUG] Offset committed: 12345