Apache Kafka × Sentiqra

Consume text from your existing Kafka pipeline, call Sentiqra for classification, and publish structured results to a downstream topic. Your Kafka retention and replay configuration remains under your control.

Why This Integration Matters

Every high-volume Trust & Safety deployment is streaming. Every real-time market signal pipeline is streaming. Every SIEM ingestion is streaming. For buyers processing millions of messages per day, a documented Kafka integration removes weeks of engineering work.

Apache Kafka is the industry standard for high-throughput streaming data. By publishing a reference implementation with consumer group configuration, schema registry integration, and exactly-once semantics, Sentiqra becomes deployable in any streaming architecture without re-engineering.

The reference implementation handles rate-limit-aware parallel calls, backpressure, and dead-letter queues for failed detections - the three engineering challenges that every team building a Sentiqra streaming pipeline has to solve.

Reference Architecture

┌──────────────────────────────────────────────────────────────┐
│                Your Kafka Cluster                            │
│                                                               │
│  ┌──────────┐    ┌──────────────┐    ┌──────────┐           │
│  │ Input     │    │ Kafka Topic: │    │ Consumer │           │
│  │ Stream    │───▶│ raw-content  │───▶│ Group:   │           │
│  │ (Kafka /  │    │              │    │ sentiqra │           │
│  │  Kinesis /│    │ Partitions:  │    │ workers  │           │
│  │  Pulsar)  │    │ N parallel   │    │ (N pods) │           │
│  └──────────┘    └──────────────┘    └────┬─────┘           │
│                                            │                   │
│                                    ┌───────▼───────┐         │
│                                    │ Sentiqra API  │         │
│                                    │ (rate-limit   │         │
│                                    │  aware)       │         │
│                                    └───────┬───────┘         │
│                                            │                   │
│                                    ┌───────▼───────┐         │
│                                    │ Kafka Topic:  │         │
│                                    │ sentiqra-     │         │
│                                    │ enriched      │         │
│                                    └───────┬───────┘         │
│                                            │                   │
│                                    ┌───────▼───────┐         │
│                                    │ Downstream    │         │
│                                    │ (SIEM, DW,    │         │
│                                    │  enforcement) │         │
│                                    └───────────────┘         │
│                                                               │
│  Dead Letter Queue: sentiqra-dlq (failed detections)        │
└──────────────────────────────────────────────────────────────┘

Data flow: Content arrives on the raw-content Kafka topic → consumer group sentiqra-workers processes messages in parallel (N pods, rate-limit aware) → each message is sent to Sentiqra API → enriched results are published to sentiqra-enriched topic → downstream consumers (SIEM, data warehouse, enforcement pipeline) consume enriched events. Failed detections go to sentiqra-dlq for retry.

Working Code

Actual, tested, copy-pasteable code. Not pseudocode. Not aspirational.

from kafka import KafkaConsumer, KafkaProducer
import json, os, time, requests

consumer = KafkaConsumer(
    "raw-content",
    bootstrap_servers=["kafka:9092"],
    group_id="sentiqra-workers",
    auto_offset_reset="latest",
    enable_auto_commit=False,
    max_poll_records=100,
)

producer = KafkaProducer(
    bootstrap_servers=["kafka:9092"],
    value_serializer=lambda v: json.dumps(v).encode("utf-8"),
)

def analyze(text, language="en", retries=3):
    """Call Sentiqra with 429-aware backoff (Retry-After honoured)."""
    for attempt in range(retries):
        response = requests.post(
            "https://sentiqra.tech/v1/analyze",
            headers={
                "Authorization": f"Bearer {os.environ['SENTIQRA_API_KEY']}",
                "Content-Type": "application/json",
            },
            json={"text": text, "language": language, "analyzers": ["trust-safety"]},
            timeout=10,
        )
        if response.status_code == 429:
            time.sleep(float(response.headers.get("Retry-After", 1)) * (attempt + 1))
            continue
        response.raise_for_status()
        return response.json()
    response.raise_for_status()

for message in consumer:
    try:
        content = json.loads(message.value.decode("utf-8"))

        data = analyze(content["text"], content.get("language", "en"))
        top = data["signals"][0] if data["signals"] else None

        # Enrich the original message with Sentiqra results
        enriched = {
            **content,
            "sentiqra_top_signal": top["code"] if top else None,
            "sentiqra_score": top["score"] if top else 0.0,
            "sentiqra_config": data.get("verification", {}).get("configuration"),
        }

        # Publish to enriched topic
        producer.send("sentiqra-enriched", enriched)
        consumer.commit()

    except Exception as e:
        # Send to dead-letter queue for retry
        producer.send("sentiqra-dlq", {
            "original": content,
            "error": str(e),
            "timestamp": time.time(),
        })
        consumer.commit()

pip install kafka-python sentiqra

# For production, use confluent-kafka (higher performance):
pip install confluent-kafka sentiqra

Security & Compliance Posture

• ✓
  Exactly-Once Semantics: Kafka's commit-after-process pattern ensures every message is processed exactly once - no duplicate detections, no missed content.
• ✓
  DSA Article 6 - Risk Assessment: Streaming architecture enables real-time content moderation at scale, supporting platform risk assessment requirements for high-volume deployments.
• ✓
  Audit Trail: The sentiqra-enriched topic retains every detection with full attribution, retained per your Kafka retention policy (typically 7–30 days).
• ✓
  Dead Letter Queue: Failed detections are preserved in the DLQ for retry and analysis, ensuring no content is silently dropped.
• ✓
  Scalability: Consumer group parallelism scales horizontally - add more pods to handle higher throughput. Sentiqra's SDK handles rate limiting automatically.

Deployment Notes