Articles / Data Systems
Kafka Anomalies in Python: Models, Events, and Offsets
Run a local Kafka and Isolation Forest demo in Python with three partitions, event validation, and offset commits after processing succeeds.
A producer sends two-feature observations to transactions. Three consumers in the same group evaluate them with Isolation Forest and publish detected anomalies to anomalies. This local demo shows what Kafka acknowledges and when processing is committed; its synthetic data does not represent real financial transactions.
The original repository preserves the demonstration that inspired the article. This page's example pins versions, uses Kafka in KRaft mode, validates messages and bounds execution. Its scripts include changes from the repository. The images below belong to that original demonstration.
Local environment and three-partition topics
Tests use Windows, Python 3.13.5 and the direct dependencies pinned below. The client was installed in an isolated directory alongside the existing environment's other dependencies; a fresh virtualenv installation and a transitive lockfile were not checked. Python commands use the Windows executable; on Linux or macOS use .venv/bin/python instead.
Save the following as compose.yaml. Kafka 4.3.1 runs a broker and KRaft controller in one container, without ZooKeeper. The published port binds to the local machine. This demo uses PLAINTEXT, one replica and limited resources; a production service needs its own authentication, availability and capacity configuration. Official Apache Kafka image.
# compose.yaml
services:
kafka:
image: apache/kafka:4.3.1
ports:
- "127.0.0.1:9092:9092"
environment:
KAFKA_HEAP_OPTS: "-Xms256m -Xmx256m"
KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
KAFKA_NODE_ID: 1
KAFKA_PROCESS_ROLES: "broker,controller"
KAFKA_LISTENERS: "PLAINTEXT://:9092,CONTROLLER://:9093"
KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://127.0.0.1:9092"
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT"
KAFKA_CONTROLLER_LISTENER_NAMES: "CONTROLLER"
KAFKA_CONTROLLER_QUORUM_VOTERS: "1@localhost:9093"
KAFKA_INTER_BROKER_LISTENER_NAME: "PLAINTEXT"
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
mem_limit: 768m
cpus: 1.0
python -m venv .venv
.\.venv\Scripts\python.exe -m pip install `
confluent-kafka==2.16.0 scikit-learn==1.9.0 numpy==2.4.6 joblib==1.5.3
New-Item -ItemType Directory -Force model, streaming
docker compose up -d
Wait for Kafka to finish starting, then explicitly create both topics. Each command must complete successfully; if the broker is not ready, inspect docker compose logs kafka before continuing. The --describe output must show three partitions. Automatic topic creation is disabled.
docker compose exec kafka /opt/kafka/bin/kafka-topics.sh `
--bootstrap-server localhost:9092 --create --topic transactions `
--partitions 3 --replication-factor 1
docker compose exec kafka /opt/kafka/bin/kafka-topics.sh `
--bootstrap-server localhost:9092 --create --topic anomalies `
--partitions 3 --replication-factor 1
docker compose exec kafka /opt/kafka/bin/kafka-topics.sh `
--bootstrap-server localhost:9092 --describe --topic transactions
A partition is an ordered sequence of messages, each with an offset within that partition. Kafka distributes a topic's partitions among consumers in the same group. Three consumers can work on these three partitions; adding more consumers leaves members without an assigned partition. Assignments may change during rebalancing: a thread's index does not bind it to a partition.
Train and save the model
Save this script as model/train.py. It generates 1,000 observations from two Gaussian clusters and saves the model next to the script, at the path the detector will use. contamination=0.01 determines a threshold from the training set; it does not measure real fraud prevalence or turn a score into a probability.
# model/train.py
from pathlib import Path
import numpy as np
from joblib import dump
from sklearn.ensemble import IsolationForest
rng = np.random.RandomState(42)
cluster = 0.3 * rng.randn(500, 2)
training = np.round(np.r_[cluster + 2, cluster - 2], 3)
model = IsolationForest(
n_estimators=50, max_samples=500, random_state=42,
contamination=0.01, n_jobs=1,
)
model.fit(training)
model_path = Path(__file__).resolve().parent / "isolation_forest.joblib"
dump(model, model_path)
print("Model saved:", model_path)
A joblib artifact can execute code when loaded. This demo loads only the model you just created in a trusted directory with compatible versions. Do not load files from an unknown source. Assessing quality requires representative data and independent evaluation of the threshold and detection errors.
Share confirmed publishing and event validation
Save streaming/kafka_io.py. The other two scripts import this module. Each send waits for its delivery callback and fails if Kafka does not confirm it within the stated deadline. Sending one event at a time makes acknowledgements visible without measuring or promising streaming throughput.
The consumer accepts a JSON object of at most 4 KiB with an ID and exactly one row of two finite numbers representable as float32. Invalid input raises an error and remains uncommitted; the demo stops so an operator can correct the source. A pipeline that continues after invalid data needs an explicit quarantine and retry policy.
# streaming/kafka_io.py
import json
import math
import numpy as np
from confluent_kafka import KafkaException, Producer
BOOTSTRAP = "127.0.0.1:9092"
TRANSACTIONS_TOPIC = "transactions"
ANOMALIES_TOPIC = "anomalies"
PARTITIONS = 3
def make_producer():
return Producer({
"bootstrap.servers": BOOTSTRAP,
"enable.idempotence": True,
"delivery.timeout.ms": 5000,
})
def publish(producer, topic, record, partition=-1):
delivery = {}
def acknowledged(error, message):
delivery["error"] = error
producer.produce(
topic, key=record["id"],
value=json.dumps(record, allow_nan=False).encode("utf-8"),
partition=partition, on_delivery=acknowledged,
)
if producer.flush(5) or "error" not in delivery:
raise TimeoutError("Event delivery was not confirmed.")
if delivery["error"] is not None:
raise KafkaException(delivery["error"])
def decode_event(payload):
if not isinstance(payload, bytes) or len(payload) > 4096:
raise ValueError("Expected a JSON event of at most 4096 bytes.")
try:
record = json.loads(payload.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError, RecursionError) as error:
raise ValueError("Invalid event JSON.") from error
if not isinstance(record, dict):
raise ValueError("Expected an event object.")
event_id = record.get("id")
if not isinstance(event_id, str) or not 1 <= len(event_id) <= 80:
raise ValueError("Expected a nonempty event id of at most 80 characters.")
data = record.get("data")
if not isinstance(data, list) or len(data) != 1:
raise ValueError("Expected one observation.")
row = data[0]
if not isinstance(row, list) or len(row) != 2:
raise ValueError("Expected two features.")
limit = float(np.finfo(np.float32).max)
for value in row:
if type(value) not in (int, float) or abs(value) > limit or not math.isfinite(value):
raise ValueError("Features must be finite numbers representable as float32.")
return record, np.asarray(data, dtype=np.float64)
Publish a finite sample
Save streaming/producer.py. It sends 60 events across the three partitions, with a distinct batch ID and a UTC timestamp. The uniform branch is selected with probability 0.05; some of those points may look normal to the model. Nor is every Gaussian point assumed to be classified as normal.
# streaming/producer.py
from datetime import datetime, timezone
from uuid import uuid4
import time
import numpy as np
from kafka_io import PARTITIONS, TRANSACTIONS_TOPIC, make_producer, publish
def produce_events(count=60):
rng = np.random.default_rng(42)
batch = uuid4().hex
with make_producer() as producer:
for index in range(count):
if rng.random() < 0.05:
observation = rng.uniform(-4, 4, size=2)
else:
observation = 0.3 * rng.normal(size=2) + rng.choice([-2, 2])
record = {
"id": f"{batch}-{index}",
"data": [np.round(observation, 3).tolist()],
"current_time": datetime.now(timezone.utc).isoformat(),
}
publish(producer, TRANSACTIONS_TOPIC, record, partition=index % PARTITIONS)
print("SENT", record["id"], flush=True)
time.sleep(0.1)
if __name__ == "__main__":
produce_events()
Detect and commit after processing
Save streaming/anomalies_detector.py. Each thread creates its own consumer and producer; the three consumers share a group. Automatic commits and automatic offset storage are disabled. For an anomaly, output delivery is confirmed before the input offset is committed. Normal events are committed after evaluation. Before subscribing, each consumer checks connectivity and the three partitions with a five-second timeout. This demo stops consumption on a global client error received during polling, even when Kafka could retry; delivery, input and commit errors also propagate. Contexts close clients and the stop event ends the other threads.
# streaming/anomalies_detector.py
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from threading import Event
import time
from joblib import load
from confluent_kafka import Consumer, KafkaException
from kafka_io import (
ANOMALIES_TOPIC, BOOTSTRAP, PARTITIONS, TRANSACTIONS_TOPIC,
decode_event, make_producer, publish,
)
model_path = Path(__file__).resolve().parents[1] / "model/isolation_forest.joblib"
def detect(stop, seconds=30, group_id="anomaly-detector-group"):
model = load(model_path)
errors = []
config = {
"bootstrap.servers": BOOTSTRAP,
"group.id": group_id,
"auto.offset.reset": "earliest",
"enable.auto.commit": False,
"enable.auto.offset.store": False,
"allow.auto.create.topics": False,
"socket.timeout.ms": 5000,
"error_cb": errors.append,
}
with Consumer(config) as consumer:
metadata = consumer.list_topics(topic=TRANSACTIONS_TOPIC, timeout=5)
topic = metadata.topics[TRANSACTIONS_TOPIC]
if topic.error is not None:
raise KafkaException(topic.error)
if len(topic.partitions) != PARTITIONS:
raise ValueError("Create the input topic with three partitions first")
with make_producer() as producer:
consumer.subscribe([TRANSACTIONS_TOPIC])
deadline = time.monotonic() + seconds
while not stop.is_set() and time.monotonic() < deadline:
message = consumer.poll(1)
if errors:
raise KafkaException(errors[0])
if message is None:
continue
if message.error():
raise KafkaException(message.error())
record, data = decode_event(message.value())
if model.predict(data)[0] == -1:
record["score"] = float(model.score_samples(data)[0])
publish(producer, ANOMALIES_TOPIC, record)
print("ANOMALY", record["id"], record["score"], flush=True)
committed = consumer.commit(message=message, asynchronous=False)
for partition in committed:
if partition.error is not None:
raise KafkaException(partition.error)
def run(seconds=30):
stop = Event()
with ThreadPoolExecutor(max_workers=PARTITIONS) as workers:
futures = [workers.submit(detect, stop, seconds) for _ in range(PARTITIONS)]
try:
for future in as_completed(futures):
future.result()
finally:
stop.set()
if __name__ == "__main__":
run()
A failure after publishing an anomaly but before committing can cause the input to reappear and produce another output copy. The ID is preserved so a downstream consumer can identify duplicates; this example implements neither persistent deduplication nor a transaction across the two topics. Producer idempotence does not remove duplicates caused by reprocessing an input. Python client offsets and delivery guarantees.
Run, inspect, and stop
After creating the four files, train first. In two terminals in the same folder, start the detector and immediately start the producer. Each consumer processes for 30 seconds after its connectivity check; startup and shutdown add time. If you start the producer late, run the detector again to process pending messages. The producer finishes after 60 events. The detector also finishes at its deadline or on error; Ctrl+C activates thread shutdown. The console prints SENT for confirmed deliveries and ANOMALY with each published detection's ID and score.
# First train the model from the project folder.
.\.venv\Scripts\python.exe model/train.py
# Terminal 1: three consumers; each processes for 30 seconds after startup.
.\.venv\Scripts\python.exe streaming/anomalies_detector.py
# Terminal 2: publish 60 events while the detector is running.
.\.venv\Scripts\python.exe streaming/producer.py
Inspect output messages and stop the broker when finished. The console consumer exits after its no-message timeout; that timeout does not prove anomalies were present. The demo measures neither end-to-end latency, production capacity nor quality on real data.
docker compose exec kafka /opt/kafka/bin/kafka-console-consumer.sh `
--bootstrap-server localhost:9092 --topic anomalies --from-beginning `
--timeout-ms 5000
docker compose down
For request-driven inference, continue with FastAPI and Docker. To examine changes in data and model results, see monitoring with Evidently.
Are you building an AI or data platform?
If you need to adapt this approach to your system, tell me about your data and the technical decision you need to resolve.