Kafka IP enrichment belongs at ingest: a consumer stamps each event with country, city, and ASN before it lands, so no query ever re-joins raw IPs against today's data. The part every example I found skips is the bill.
The usual version calls an IP geolocation API inside the per-message loop. Addresses repeat constantly on a busy topic, so you pay for every repeat, inherit the API's latency on every event, and turn the first slow response into consumer lag.
This post builds the other version in Python: batch a poll, dedupe the IPs, send the misses in one bulk request, produce, then commit. In the test run behind this post, 62 seeded events turned into one bulk request for 5 distinct public IPs, billed 5 credits. There's also a path that swaps the API for a local MMDB file once volume makes that cheaper.
TL;DR
- Run the enricher as its own consumer group between
events.rawandevents.enriched, with a dead-letter topic for records you can't parse. - Price IP enrichment by distinct IPs per cache window, not by events. Duplicates in a bulk request are billed; private addresses are not.
- Classify IPs locally first, dedupe each poll, and send cache misses in one bulk request of up to 50,000 addresses.
- Commit offsets only after the enriched batch is acknowledged. Ship lookup failures with a status field instead of stalling the partition.
- The API and a combined location + ASN MMDB return the same contract fields in different shapes. Normalize both into one event contract. The enricher below is about 270 lines of Python across nine blocks. It runs against plain Apache Kafka or Redpanda, keeps per-user ordering, and survives a dead API by shipping events marked
unavailableinstead of blocking. Every block ran against a real Kafka broker, and the API path ran against the live lookup endpoint.
Where the enricher sits
The enricher is a consumer group with one job. It reads raw events, adds geo, asn, and an enrichment block, and writes them to a second topic. Anything that isn't JSON goes to a third topic untouched.
producers ──> events.raw ──> [ ip-enricher group ] ──> events.enriched ──> sinks │ └──> events.dlq (unparseable records) I've made the case for log enrichment at write time instead of query time twice already, once for nginx access logs and once for Elasticsearch ingest pipelines. The short version: an IP's network owner and location drift, so an event should carry the answer that was true when it happened. Kafka adds the reason that matters here. Every downstream consumer, sink, and replay reads the enriched topic, so the lookup happens exactly once per event instead of once per reader.
Create the three topics. Match the raw topic's partition count on the enriched one, so a consumer instance's output lines up with its input:
kafka-topics.sh --bootstrap-server localhost:9092 --create --topic events.raw --partitions 3 kafka-topics.sh --bootstrap-server localhost:9092 --create --topic events.enriched --partitions 3 kafka-topics.sh --bootstrap-server localhost:9092 --create --topic events.dlq --partitions 1 The raw events in this post look like {"event_id": "e1", "user": "u1", "path": "/checkout", "client_ip": "91.128.103.196"}, keyed by user.
Kafka IP enrichment is priced by distinct IPs
An IP lookup API bills per address looked up, so events per second is the wrong number to plan with. The number that matters is how many distinct addresses you see per cache window.
A rough monthly bill, in credits, at one credit per address:
credits/month ≈ distinct IPs per TTL window × (30 ÷ TTL in days) × instances that see the same IP At a 24-hour TTL with one consumer instance, 5,000 distinct IPs a day is 150,000 credits a month, which is exactly the entry paid tier.
About 33,000 a day is roughly 1,000,000 credits. The last factor is the one people miss. Each consumer instance has its own cache, so an address whose events spread across partitions gets looked up by every instance that sees it.
Three details from the live bulk endpoint shape the code below. Duplicate addresses inside one request are billed again, so dedupe before you send. Private and bogon addresses cost nothing and come back as an error entry, so filter them locally anyway and skip the round trip. And the bulk endpoint needs a paid key: a free key gets a 401, and free responses also omit asn.type, which is the field that tells HOSTING from ISP.
| API bulk + TTL cache | Local MMDB file | |
|---|---|---|
| Cost | One credit per distinct IP per window | A fixed licence, nothing per lookup |
| Lookup | One HTTPS round trip per batch with misses | In-process, no network |
| Freshness | Whatever the API answers today | The build you downloaded, daily or weekly |
| Ops work | An API key | Shipping a large file to every node and reloading it |
| Pick it when | You're under about 1M credits a month | You're above that, or the stream tier has no internet egress |
I can't give you the exact crossover, because database licences are quote-based. What I'd do: start on the API, watch the credit counter for a week of real traffic, and ask for a database quote the month you pass a million.
The enriched event contract
This is what lands on events.enriched for one real event from the test run:
{ "event_id": "e1", "user": "u1", "path": "/checkout", "client_ip": "91.128.103.196", "geo": {"country_code": "SE", "city": "Stockholm", "lat": 59.4051, "lon": 17.9551}, "asn": {"number": 1257, "org": "Tele2 Sverige AB", "type": "ISP"}, "enrichment": {"status": "ok", "source": "api", "as_of": "2026-10-12T12:50:10+00:00"} } Coordinates are normalized to floats and the AS number to an integer. The API sends these values as strings, while the MMDB may return numeric values directly. enrichment.status is one of ok, not_found, private, invalid_ip, no_ip, or unavailable, so a downstream query can tell "no data" from "lookup failed". as_of is the lookup time on the API path and the database build date on the MMDB path. That field is what makes an ingest-time answer defensible six months later.
The lookup returns far more than the contract keeps. Here's the complete record the API returns for that address, for reference:
Full API record for 91.128.103.196{ "ip": "91.128.103.196", "location": { "continent_code": "EU", "continent_name": "Europe", "country_code2": "SE", "country_code3": "SWE", "country_name": "Sweden", "country_name_official": "Kingdom of Sweden", "country_capital": "Stockholm", "state_prov": "Stockholms län", "state_code": "SE-AB", "district": "Stockholm", "city": "Stockholm", "zipcode": "164 40", "latitude": "59.40510", "longitude": "17.95510", "is_eu": true, "country_flag": "https://ipgeolocation.io/static/flags/se_64.png", "geoname_id": "9972319", "country_emoji": "\ud83c\uddf8\ud83c\uddea" }, "country_metadata": { "calling_code": "+46", "tld": ".se", "languages": [ "sv-SE", "se", "sma", "fi-SE" ] }, "network": { "connection_type": "", "route": "91.128.0.0/14", "is_anycast": false, "is_cdn": false, "cdn_provider_name": "" }, "currency": { "code": "SEK", "name": "Swedish Krona", "symbol": "kr" }, "asn": { "as_number": "AS1257", "organization": "Tele2 Sverige AB", "country": "SE", "type": "ISP", "domain": "tele2.se", "date_allocated": "2002-09-19", "rir": "RIPE" }, "company": { "name": "Tele2 Sverige AB", "type": "ISP", "domain": "tele2.se" }, "time_zone": { "name": "Europe/Stockholm", "offset": 1, "offset_with_dst": 2, "current_time": "2026-10-12 14:39:15.199+0200", "current_time_unix": 1791808755.199, "current_tz_abbreviation": "CEST", "current_tz_full_name": "Central European Summer Time", "standard_tz_abbreviation": "CET", "standard_tz_full_name": "Central European Standard Time", "is_dst": true, "dst_savings": 1, "dst_exists": true, "dst_tz_abbreviation": "CEST", "dst_tz_full_name": "Central European Summer Time", "dst_start": { "utc_time": "2026-03-29 TIME 01:00", "duration": "+1.00H", "gap": true, "date_time_after": "2026-03-29 TIME 03:00", "date_time_before": "2026-03-29 TIME 02:00", "overlap": false }, "dst_end": { "utc_time": "2026-10-25 TIME 01:00", "duration": "-1.00H", "gap": false, "date_time_after": "2026-10-25 TIME 02:00", "date_time_before": "2026-10-25 TIME 03:00", "overlap": true } } } That record is about 1.8 KB. Asking for only the contract's fields with the fields parameter cuts it to about 240 bytes, which matters at 50,000 addresses per request: by my per-record measurements, roughly 92 MB of response versus 12 MB.
The API and a combined location + ASN MMDB carry the contract fields under different paths and types. This is the mapping the normalizer handles, and the reason a consumer written against one source silently produces empty fields on the other:
| Contract field | API (/v3/ipgeo-bulk) | MMDB file |
|---|---|---|
geo.country_code | location.country_code2 | location.country.code2 |
geo.city | location.city | location.city.name.en |
geo.lat / geo.lon | location.latitude (string) | location.coordinates.latitude (number) |
asn.number | asn.as_number, "AS1257" | asn.as_number (1257) |
asn.org | asn.organization | asn.organization |
asn.type | asn.type (paid plans only) | asn.type |
Building the consumer
You'll need pip install confluent-kafka requests cachetools maxminddb and Python 3.9 or newer. The blocks below go into one file, enricher.py, in order. Each one repeats the imports it uses, so you can read them in isolation. It was tested against Apache Kafka 4.2.2 in KRaft mode. The client is Confluent's Python client, which talks to Kafka and Redpanda alike.
Classify every IP before you spend anything
import ipaddress import logging import os log = logging.getLogger("ip-enricher") BOOTSTRAP = os.environ.get("KAFKA_BOOTSTRAP", "localhost:9092") RAW_TOPIC = os.environ.get("RAW_TOPIC", "events.raw") ENRICHED_TOPIC = os.environ.get("ENRICHED_TOPIC", "events.enriched") DLQ_TOPIC = os.environ.get("DLQ_TOPIC", "events.dlq") API_KEY = os.environ.get("IPGEO_API_KEY") # bulk lookups need a paid key; never hardcode it def classify_ip(raw): """Return (cache_key, status). status is None when the IP is worth looking up.""" if not isinstance(raw, str) or not raw.strip(): return None, "no_ip" try: ip = ipaddress.ip_address(raw.strip()) except ValueError: return None, "invalid_ip" # ::ffff:8.8.8.8 and 8.8.8.8 are the same host. Key them the same way. if ip.version == 6 and ip.ipv4_mapped: ip = ip.ipv4_mapped if not ip.is_global: # RFC 1918, CGNAT, loopback, link-local. Nothing to look up, nothing to pay for. return str(ip), "private" return str(ip), None is_global does the heavy lifting. It's False for RFC 1918 ranges, CGNAT's 100.64.0.0/10, loopback, link-local, and the documentation ranges, which is everything you'd otherwise pay a round trip to hear rejected. The IPv4-mapped unwrap matters more than it looks: a server listening on :: sees IPv4 clients as ::ffff:8.8.8.8, and without the unwrap the same client gets two cache entries and two credits.
One bulk request per batch
I'm using ipgeolocation.io for bulk IP lookups because its bulk lookup endpoint docs cover the shape this needs (up to 50,000 addresses per POST) and the same data ships as an MMDB file, so both paths in this post produce identical events. MaxMind and IP2Location also sell both an API and a downloadable database, and the cost model above applies to them unchanged.
The session, cache, and error types come first:
import requests from cachetools import TTLCache from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry BULK_URL = "https://api.ipgeolocation.io/v3/ipgeo-bulk" BULK_MAX = 50_000 # hard per-request limit; one more and the API returns 400 FIELDS = ("ip,location.country_code2,location.city,location.latitude," "location.longitude,asn.as_number,asn.organization,asn.type") # Location and network ownership drift slowly, so a day is a sane TTL. Each entry costs # about 1 KB in CPython, so 100,000 entries is roughly 100 MB. Size it to distinct IPs per day. cache = TTLCache(maxsize=100_000, ttl=24 * 3600) session = requests.Session() # Retry connection failures and explicit gateway errors, but never a read # failure after the POST was sent: the server may already have processed it. session.mount("https://", HTTPAdapter(max_retries=Retry( total=None, connect=2, read=0, status=2, other=0, backoff_factor=0.5, status_forcelist=(502, 503, 504), allowed_methods=frozenset({"POST"}), ))) class FatalLookupError(RuntimeError): """Bad key or a free-plan key. Stop the consumer rather than ship blank events forever.""" class LookupFailed(RuntimeError): """Transient failure. The caller fails open.""" A 100,000-entry cache is my starting point, not a rule. Size maxsize to the distinct addresses you see per TTL window, because an evicted entry is a re-bought one.
import ipaddress import logging import requests log = logging.getLogger("ip-enricher") def bulk_lookup(ips, api_key): """ips: distinct public IPs missing from the cache. Returns {ip: record or None}.""" found = {} for start in range(0, len(ips), BULK_MAX): chunk = ips[start:start + BULK_MAX] try: resp = session.post(BULK_URL, params={"apiKey": api_key, "fields": FIELDS}, json={"ips": chunk}, timeout=(2, 10)) except requests.RequestException as exc: # requests puts the full URL, apiKey included, in exception text. Don't log it. raise LookupFailed(type(exc).__name__) from None if resp.status_code == 401: raise FatalLookupError(resp.text[:200]) if resp.status_code != 200: raise LookupFailed(f"HTTP {resp.status_code}: {resp.text[:200]}") log.info( "bulk: %d ips, %s credits", len(chunk), resp.headers.get("X-Credits-Charged"), ) # Match valid results by canonical IP rather than depending on response order. # IPv6 may be returned in a different textual form from the request. for rec in resp.json(): if not isinstance(rec, dict) or "ip" not in rec: continue try: returned = ipaddress.ip_address(rec["ip"]) if returned.version == 6 and returned.ipv4_mapped: returned = returned.ipv4_mapped found[str(returned)] = rec except ValueError: continue # Rejected or missing records become not_found. for ip in chunk: found.setdefault(ip, None) return found Two response details are worth handling explicitly. A rejected entry contains only a message and no ip field, while a valid IPv6 address may be returned in a different textual form from the one you sent. The code therefore canonicalizes returned IPs with ipaddress and maps valid records by address instead of depending on array position.
Third, the except clause is there because requests puts the full URL into its exception text, query string and apiKey included. Log that message once on a connection error and your key is in your log pipeline. I confirmed it by pointing the client at a closed port: the ConnectionError text contained the key.
Normalize, then resolve with a fail-open path
def _float(value): try: return float(value) except (TypeError, ValueError): return None def _asn_int(value): # The API sends "AS1257"; the MMDB file sends "1257". Store one integer type. digits = str(value or "").upper().removeprefix("AS") return int(digits) if digits.isdigit() and int(digits) > 0 else None def _contract(country, city, lat, lon, asn): return { "geo": {"country_code": country or None, "city": city or None, "lat": _float(lat), "lon": _float(lon)}, "asn": {"number": _asn_int(asn.get("as_number")), "org": asn.get("organization") or None, "type": asn.get("type") or None}, } def from_api(rec): loc = rec.get("location") or {} return _contract(loc.get("country_code2"), loc.Source：https://dev.to/abdullah_afzal/enrich-kafka-events-with-ip-location-and-asn-at-ingest-3db8Related posts
I Had a Perfect Sprint Board. 18 AI Subagents Still Wrecked the Repo.
What Is Attribution? AI Music Needs a Trail Back to the People Who Made the Work
A Free Google Alerts Alternative: Monitoring Global News With the GDELT DOC API
Butter Scraped Over Too Much Bread
From Six Seven to Six Zeros to What Next?
