Enrich Kafka Events With IP Location and ASN at Ingest

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.raw and events.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 unavailable instead 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.get("city"),
                     loc.get("latitude"), loc.get("longitude"), rec.get("asn") or {})


def from_mmdb(rec):
    # Same data, different paths: nested names, a coordinates group, unprefixed AS numbers.
    loc = rec.get("location") or {}
    coords = loc.get("coordinates") or {}
    city = ((loc.get("city") or {}).get("name") or {}).get("en")
    return _contract((loc.get("country") or {}).get("code2"), city,
                     coords.get("latitude"), coords.get("longitude"), rec.get("asn") or {})

_contract is the single place the event shape is defined. Both sources feed it, so a schema change happens once. The > 0 check on AS numbers is deliberate: some records come back as AS0 with empty organization fields, and storing 0 as if it were a real network causes confusing joins later.

from datetime import datetime, timezone


def resolve_api(keys, api_key):
    """keys: set of public IPs. Returns {ip: (status, data, as_of)}."""
    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
    out = {ip: cache[ip] for ip in keys if ip in cache}
    misses = [ip for ip in keys if ip not in out]
    if not misses:
        return out
    try:
        for ip, rec in bulk_lookup(misses, api_key).items():
            out[ip] = cache[ip] = (("ok", from_api(rec), now) if rec
                                   else ("not_found", None, now))
    except LookupFailed as exc:
        # Fail open: the events still ship, marked unavailable. Failures are never cached.
        log.warning("lookup failed for %d ips, shipping un-enriched: %s", len(misses), exc)
        out.update({ip: ("unavailable", None, now) for ip in misses})
    return out

This is the whole caching policy. Hits skip the network. Misses go out in one call. A transient failure marks the affected addresses unavailable for this batch only, and failures are never cached, so the next batch tries again.

Failing open is the right default for analytics and observability topics, where a gap is better than a stall. If this topic feeds a fraud decision, you'd rather pause than ship blanks, and that's a different design.

Process a batch, then commit

import json


def process_batch(msgs, producer, on_delivery, resolve, source):
    """Parse, dedupe, resolve once per distinct IP, then produce one enriched event per message."""
    parsed = []
    for msg in msgs:
        try:
            event = json.loads(msg.value())
            if not isinstance(event, dict):
                raise ValueError("not a JSON object")
        except (TypeError, ValueError):
            # Unparseable records go to the DLQ untouched, so nothing is silently dropped.
            producer.produce(DLQ_TOPIC, key=msg.key(), value=msg.value(),
                             headers={"error": "bad_json"}, on_delivery=on_delivery)
            continue
        key, status = classify_ip(event.get("client_ip"))
        parsed.append((msg, event, key, status))

    # 500 messages from a busy topic are often a few dozen distinct IPs.
    results = resolve({key for _, _, key, status in parsed if status is None})

    for msg, event, key, status in parsed:
        data, as_of = None, None
        if status is None:
            status, data, as_of = results[key]
        event.update(data or {"geo": None, "asn": None})
        event["enrichment"] = {"status": status, "source": source if as_of else None,
                               "as_of": as_of}
        # Keep the original key: per-user ordering survives enrichment.
        producer.produce(ENRICHED_TOPIC, key=msg.key(), value=json.dumps(event),
                         on_delivery=on_delivery)

The comment about distinct IPs is the point of the whole post. The test seed was 12 client_ip values repeated five times: repeats, an IPv4-mapped duplicate, two private addresses, a garbage string, and one missing field, plus two records that weren't JSON objects. After classify_ip and the set comprehension, the lookup saw five addresses.

import signal

from confluent_kafka import Consumer, KafkaException, Producer


def run(resolve, source):
    consumer = Consumer({"bootstrap.servers": BOOTSTRAP, "group.id": "ip-enricher",
                         "enable.auto.commit": False, "auto.offset.reset": "earliest"})
    producer = Producer({"bootstrap.servers": BOOTSTRAP,
                         "enable.idempotence": True, "linger.ms": 20})
    failed = []
    stopping = []
    signal.signal(signal.SIGTERM, lambda *_: stopping.append(True))
    signal.signal(signal.SIGINT, lambda *_: stopping.append(True))

    def on_delivery(err, _msg):
        if err is not None:
            failed.append(err)

    consumer.subscribe([RAW_TOPIC])
    try:
        while not stopping:
            msgs = consumer.consume(num_messages=500, timeout=1.0)
            data_msgs = []
            for msg in msgs:
                if msg.error():
                    if msg.error().fatal():
                        raise KafkaException(msg.error())
                    log.warning("consumer error: %s", msg.error())
                    continue
                data_msgs.append(msg)
            if not data_msgs:
                continue
            process_batch(data_msgs, producer, on_delivery, resolve, source)
            # Commit only after every enriched record is acknowledged. A crash before this
            # line replays the batch: at-least-once, so downstream must tolerate duplicates.
            if producer.flush(30) > 0 or failed:
                raise KafkaException(f"{len(failed)} deliveries failed; not committing")
            consumer.commit(asynchronous=False)
    finally:
        consumer.close()

enable.auto.commit is off, and the commit sits after flush(). If the process dies anywhere above that line, the group replays the batch from the last committed offset. That's at-least-once delivery. The idempotent producer stops retry duplicates within a session, but a restart can still re-emit a batch, so give downstream something to dedupe on. event_id is that something here.

Wire it up

import functools
import logging
import os
import sys

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    mmdb_path = os.environ.get("IPGEO_MMDB_PATH")
    try:
        if mmdb_path:
            run(MmdbLookup(mmdb_path).resolve, source="mmdb")
        elif API_KEY:
            run(functools.partial(resolve_api, api_key=API_KEY), source="api")
        else:
            sys.exit("Set IPGEO_MMDB_PATH or IPGEO_API_KEY")
    except FatalLookupError as exc:
        # Exit non-zero with offsets uncommitted. Fix the key, restart, nothing is lost.
        sys.exit(f"lookup configuration error: {exc}")

This block goes last in enricher.py, after the MmdbLookup class from the next section. Run it with IPGEO_API_KEY set for the API path, or IPGEO_MMDB_PATH for the file path covered next. A bad or free-plan key raises FatalLookupError, and the process exits non-zero before committing anything. That's the one failure that should stop the stream, because failing open on a misconfigured key ships blank events forever. In testing, a bad key exited with status 1 and left the batch uncommitted. On the next start, the group picked those events up from the last committed offset, with nothing lost.

Swapping in a local MMDB

Once the credit counter says it's time, the same consumer can read a local MMDB instead. The implementation below expects one combined file containing both location and ASN data. A standalone city database does not contain asn.*, and a standalone ASN database does not contain location.*. If your subscription provides separate files, open both and merge their records before passing them to the normalizer. Every enricher node needs local access to the required file or files.

import os
from datetime import datetime, timezone

import maxminddb


class MmdbLookup:
    """Lookups from a local IPGeolocation MMDB, reopened when the file is replaced."""

    def __init__(self, path):
        self.path = path
        self.reader = None
        self._open()

    def _open(self):
        st = os.stat(self.path)
        new = maxminddb.open_database(self.path)  # MODE_AUTO: C extension and mmap when available
        old, self.reader = self.reader, new
        self.identity = (st.st_ino, st.st_mtime_ns)
        built = datetime.fromtimestamp(new.metadata().build_epoch, timezone.utc)
        self.as_of = built.isoformat(timespec="seconds")  # the build date, not the lookup time
        if old is not None:
            old.close()

    def resolve(self, keys):
        # Updates rename a new file over the old one, so a new inode means a new build.
        try:
            st = os.stat(self.path)
            if (st.st_ino, st.st_mtime_ns) != self.identity:
                self._open()
                log.info("reloaded %s, build %s", self.path, self.as_of)
        except (FileNotFoundError, maxminddb.InvalidDatabaseError) as exc:
            log.warning("keeping current database, reload failed: %s", exc)
        out = {}
        for ip in keys:
            rec = self.reader.get(ip)
            out[ip] = ("ok", from_mmdb(rec), self.as_of) if rec else ("not_found", None, self.as_of)
        return out

Read the file with a generic MMDB reader such as the maxminddb package, not MaxMind's typed GeoIP2 classes from the geoip2 package. The typed classes map MaxMind's own keys (country.iso_code, city.names) onto models. Against a file in this layout, city() returns a model with every field set to None, or raises TypeError on its database-type check, depending on the file's metadata. Anything built on those classes inherits that, including Kafka Connect transforms written for GeoLite2.

Two details keep the reload safe. Database updates are installed by renaming a new file over the old one, so a changed inode is a reliable "new build" signal, and a file that's missing mid-swap just means "keep the reader you have". And as_of here is the build date, not the lookup time. A replay next month should tell you which build answered.

In testing, I renamed a new file into place while the consumer was running. The next batch logged the reload and picked up the new data, with no restart.

Things that bite in production

The lookup's worst case has to fit inside the poll interval. Keep your connect/read timeout plus any safe retry backoff well under max.poll.interval.ms, or the group can rebalance you out mid-batch and replay it. The retry policy above does not retry a POST after a read timeout, because the server may already have processed that batch. If you'd rather stop than fail open, consumer.pause() the assignment until the API recovers.

Keying by user and keying by IP pull in opposite directions. Keying the raw topic by client IP pins each address to one partition and one cache, which cuts credits. It also breaks per-user ordering if anything downstream relies on it. I keep the user key and accept the duplicate lookups until the bill says otherwise. A shared Redis cache is the next step after that, not before.

The IP has to be the client's. If your producers sit behind a load balancer, client_ip must come from a trusted X-Forwarded-For hop at the edge. Otherwise your entire user base appears to live wherever your load balancer does.

Schema Registry users: add the new fields with defaults. In Avro, adding geo, asn, and enrichment as nullable fields with a null default keeps the change backward compatible. Without defaults, old readers break the day you deploy.

When not to build this

Not every Kafka data enrichment job needs its own consumer. If Fluent Bit or Vector already sits in front of Kafka, enrich there. Both have MMDB lookups, and the events arrive enriched without a consumer to run. If your data enters through Kafka Connect and you're on MaxMind or IP2Location data, single message transforms do this job inside Connect: IP2Location publishes its own, and the MaxMind one is a community project. If you already run Flink for streaming data enrichment elsewhere, its async I/O operator handles API lookups with backpressure built in.

The consumer above earns its keep when you want lookups batched and priced deliberately, a contract that doesn't change when the data source does, and failure behavior you chose rather than inherited.

Before you ship it, set the cache TTL to how fast your answer actually goes stale. Then alert on the ratio of unavailable to ok in enrichment.status rather than on exceptions, and re-check the credit counter after the first week of real traffic, because distinct-IP counts are the one number nobody guesses right up front.