Kafka subscriber plugin

Note: This plugin requires InfluxDB 3 Core.8.2 or later.

The Kafka Subscriber Plugin enables real-time ingestion of Kafka messages into InfluxDB 3 Core. Subscribe to Kafka topics and automatically transform messages into time-series data with support for JSON, Line Protocol, and custom text formats. The plugin uses consumer groups for reliable message delivery and provides flexible offset commit policies for different use cases. The consumer is reused across scheduled invocations to avoid a consumer-group rebalance every cycle, and it optionally supports a dead-letter queue topic for failed messages and id-based deduplication for messages without their own timestamp.

Binary formats: avro, jsonschema, and protobuf (Confluent wire format) are decoded via Confluent Schema Registry. avro can also be decoded as raw schemaless Avro with a local .avsc schema, and protobuf with a local .proto schema. See Binary formats.

Configuration

Plugin parameters may be specified as key-value pairs in the --trigger-arguments flag (CLI) or in the trigger_arguments field (API) when creating a trigger. This plugin supports TOML configuration files for complex mapping scenarios, which can be specified using the config_file_path parameter.

Plugin metadata

This plugin includes a JSON metadata schema in its docstring that defines supported trigger types and configuration parameters. This metadata enables the InfluxDB 3 Explorer UI to display and configure the plugin.

Required parameters

In TOML configuration, bootstrap_servers, topics, and group_id are placed under the [kafka] section; table_name and table_name_field are placed under [mapping.json] or [mapping.text] section.

Connection parameters

In TOML configuration, these parameters are placed under the [kafka] section.

Offset Commit Policy

In TOML configuration, this parameter is placed under the [kafka] section.

Policy behavior:

When to use each policy:

• on_success (recommended): Use when data integrity is important. Failed messages will be reprocessed on the next trigger execution.
• always: Use in high-throughput scenarios where occasional data loss is acceptable, or when you have external error handling (failed messages are logged to kafka_exceptions table).

Interaction with dlq_topic:

Because the consumer is reused across invocations, its fetch position advances as messages are polled. Under on_success, when a message still needs reprocessing — a transient InfluxDB write failure, or a parse failure (poison message) that could not be offloaded — the plugin rewinds the consumer to the lowest such offset on each affected partition, so those messages are re-read on the next cycle. Only the affected partitions are rewound; partitions whose messages all succeeded are not. A parse failure that is successfully published to dlq_topic is treated as handled: the consumer is not rewound for it, and it is preserved in the DLQ instead of being re-read.

Without dlq_topic, a poison message under on_success is rewound and reprocessed on every cycle until it is resolved (its original payload is not stored in kafka_exceptions, so configure dlq_topic if you need to skip past and preserve poison messages). With always, the offset is committed immediately regardless and the consumer is never rewound, so a failed message is not reprocessed; dlq_topic simply preserves the failed messages that would otherwise be dropped.

Note: Rewinding re-reads from the failed offset, so messages on the same partition that succeeded after the failed one are re-read and re-written. Enable deduplication (or use messages carrying their own timestamp) to avoid duplicate points from this at-least-once retry.

Security parameters

In TOML configuration, this parameter is placed under the [kafka] section.

SASL Authentication parameters

In TOML configuration, these parameters are placed under the [kafka.sasl] section with shortened names.

Note: All three SASL parameters must be provided together when using SASL authentication.

SSL/TLS parameters

In TOML configuration, these parameters are placed under the [kafka.ssl] section with shortened names.

Note: For mutual TLS, both ssl_cert and ssl_key must be provided together.

Logging parameters

In TOML configuration, enable_full_logging is placed directly under the [kafka] section.

Dead-letter queue parameters

In TOML configuration, dlq_topic is placed under the [kafka] section.

Behavior:

• Only parse failures (poison messages) are routed to the DLQ topic — including messages missing a required dedup_id_field. Transient write failures to InfluxDB are not sent to the DLQ; they keep their offsets uncommitted (with on_success) so they are retried on the next cycle.
• Failures continue to be recorded in the kafka_exceptions table regardless of dlq_topic.
• The DLQ producer reuses the same bootstrap_servers and security settings as the consumer.
• The DLQ topic must already exist on the broker. The plugin does not create it. If your broker has auto.create.topics.enable=false (typical for production and managed services such as Confluent Cloud or MSK), create the topic beforehand. When the topic is missing, delivery fails and — under on_success — the offset is not committed, so the batch is reprocessed until the topic exists. The real broker reason (for example Unknown topic or partition) is logged so you can fix it quickly.

Deduplication parameters

In TOML configuration, dedup_window is placed under the [kafka] section; dedup_id_field and dedup_id_name are placed under [mapping.json] or [mapping.text] section.

Behavior:

• Deduplication is only active when no timestamp_field is configured. When messages carry their own timestamp, InfluxDB already overwrites duplicates by (measurement, tags, time), so no extra check is needed.
• When active, the plugin extracts the id, writes it under the dedup_id_name field (default dedup_id), and — before writing — checks whether that id already exists within dedup_window. The check is a single query per table per cycle, constrained by both the time window and the batch’s ids (... WHERE time >= <cutoff> AND "<dedup_id_name>" IN (...)), so it returns few rows. Duplicates within the same batch are also dropped.
• If the dedup query fails, the plugin fails open (writes the records) rather than dropping data.
• A message that is missing the configured id is treated as a parse failure (routed to the DLQ / kafka_exceptions). For JSON arrays, individual elements missing the id are skipped with a warning.
• Performance: for very high-volume topics prefer a shorter dedup_window to keep the query light.
• Pick a dedup_id_name that does not collide with a field you already map — it would be written twice.

Message format parameters

In TOML configuration, format is placed under the [kafka] section; timestamp_field is placed under [mapping.json] or [mapping.text] section.

Supported formats:

Format-specific timestamp_field syntax:

Supported timestamp formats:

• ns - nanoseconds (Unix timestamp)
• ms - milliseconds (Unix timestamp)
• s - seconds (Unix timestamp)
• datetime - ISO 8601 string (for example, “2021-12-01T12:00:00Z”)

For avro and jsonschema, a field decoded as a native datetime/date (Avro/JSON Schema logical type) is converted to a timestamp automatically, regardless of the configured format specifier.

JSON format parameters

In TOML configuration, tags are placed under [mapping.json.tags] section and fields under [mapping.json.fields] section.

Field specification format: "temp:float=temperature hum:int=humidity status:bool=online"

Supported field types: int, uint, float, string, bool

Text format parameters

In TOML configuration, tags are placed under [mapping.text.tags] section and fields under [mapping.text.fields] section.

Binary formats

The decoded record is a dict (or a top-level list, expanded into one record per element) for all binary formats, so field/tag/timestamp mapping uses the same JSONPath syntax as the JSON format, under [mapping.avro], [mapping.jsonschema], or [mapping.protobuf].

Per-message decode failures (schema id not found, malformed payload) are handled like parse failures: the message is logged to kafka_exceptions, sent to dlq_topic if configured, or replayed otherwise. Binary payloads are forwarded to the DLQ as the original raw bytes.

An unreachable or misconfigured Schema Registry is caught when the decoder is first built: the cycle is aborted before any message is polled or committed, so nothing is lost, and the run retries on the next schedule.

Avro / JSON Schema (Schema Registry)

Set format to avro or jsonschema to consume binary messages in the Confluent wire format (magic byte + 4-byte schema id + payload). The schema is fetched from the Schema Registry by the id embedded in each message — no local schema files are needed.

The parameter names are identical in CLI/API arguments and TOML. In TOML they are placed under the [kafka.schema_registry] section.

Note: avro and jsonschema require the extra packages httpx, fastavro, and jsonschema.

Avro (local .avsc schema, schemaless)

Set format to avro and avro_schema_file to a local .avsc schema to consume Avro messages produced without a Schema Registry — raw schemaless encoding with no Confluent wire header. The whole message value is the Avro body and is decoded with the local schema; the Schema Registry is not contacted.

The parameter name is identical in CLI/API arguments and TOML. In TOML it is placed under the [kafka.avro] section.

Note: requires the fastavro package. The schema is parsed once and cached. Writer and reader schema are assumed identical (no schema evolution).

Protobuf (local .proto or Schema Registry)

Set format to protobuf to consume Protobuf messages in the Confluent wire format. The schema source is chosen implicitly, like avro:

• Local .proto — set protobuf_schema_file. The wire framing is parsed locally and the schema is read from the file; the Schema Registry is not contacted.
• Schema Registry — omit protobuf_schema_file and set schema_registry_url. The .proto (and any referenced schemas) is fetched by the wire-frame schema id, compiled with protoc, and cached per schema id. The [kafka.schema_registry] section (url, auth, TLS) is shared with avro/jsonschema.

The message index selects which message type to decode, or set protobuf_message_type to choose it explicitly. In registry mode the schema matches the producer’s, so the message index is always correct and protobuf_message_type is usually unnecessary.

The parameter names are identical in CLI/API arguments and TOML. In TOML they are placed under the [kafka.protobuf] section.

Note: protobuf requires grpcio-tools (bundles protoc to compile the .proto); it is included in the plugin’s dependencies and installed automatically. The .proto is compiled once and cached. MessageToDict semantics apply: proto3 scalar fields equal to their default (0/""/false) are omitted from the record; int64/uint64 become strings; enums become their names; bytes become base64 strings.

See examples 11–14 in kafka_config_example.toml.

TOML configuration

File path resolution

All file paths in the plugin (configuration file, TLS certificates) follow the same resolution logic:

• Absolute paths (for example, /etc/kafka/config.toml) are used as-is
• Relative paths (for example, config.toml, certs/ca.crt) are resolved against the plugin directory, taken from PLUGIN_DIR, then INFLUXDB3_PLUGIN_DIR, then the parent of VIRTUAL_ENV

If a relative path is specified and none of these can be resolved, the plugin will return an error.

Example TOML configuration

kafka_config_example.toml - comprehensive configuration example with all formats and security options

Data requirements

The plugin automatically creates the target measurement table on first write. Field mappings are required for JSON and Text formats to specify which fields to extract and their data types.

Message encoding requirements

• Text formats (json, text, lineprotocol): payloads must be UTF-8 encoded.
• Binary formats (avro, jsonschema, protobuf): payloads are decoded from binary (see Binary formats).
• Non-empty payloads: Empty or whitespace-only messages are automatically skipped.

Software Requirements

• InfluxDB 3 Core: with the Processing Engine enabled
• Python packages (all bundled in manifest.toml and installed automatically with the plugin):
  • confluent-kafka>=2.15.0 (Kafka client based on librdkafka) — all formats
  • jsonpath-ng (JSONPath field mapping) — all formats
  • fastavro — avro format (both Schema Registry and local .avsc)
  • httpx, authlib, cachetools — any Schema Registry format (avro / jsonschema / registry-mode protobuf)
  • jsonschema (library) — jsonschema message format only
  • protobuf, grpcio-tools — protobuf format only

Installation steps

1. Start InfluxDB 3 Core with the Processing Engine enabled (--plugin-dir /path/to/plugins):

   influxdb3 serve \
     --node-id node0 \
     --object-store file \
     --data-dir ~/.influxdb3 \
     --plugin-dir ~/.plugins

2. Install required Python packages:

   influxdb3 install package confluent-kafka
   influxdb3 install package jsonpath-ng

Trigger setup

Scheduled ingestion with TOML configuration

Recommended for production use with complex mappings:

# 1. Set PLUGIN_DIR environment variable
export PLUGIN_DIR=~/.plugins

# 2. Copy and edit configuration file
cp kafka_config_example.toml $PLUGIN_DIR/my_kafka_config.toml
# Edit my_kafka_config.toml with your Kafka cluster and mapping settings

# 3. Create the trigger
influxdb3 create trigger \
  --database mydb \
  --plugin-filename gh:influxdata/kafka_subscriber/kafka_subscriber.py \
  --trigger-spec "every:10s" \
  --trigger-arguments config_file_path=my_kafka_config.toml \
  kafka_ingestion

Scheduled ingestion with command-line arguments

For simple JSON message ingestion:

influxdb3 create trigger \
  --database mydb \
  --plugin-filename gh:influxdata/kafka_subscriber/kafka_subscriber.py \
  --trigger-spec "every:5s" \
  --trigger-arguments 'bootstrap_servers=kafka1:9092 kafka2:9092,topics=sensors.temperature sensors.humidity,group_id=influxdb3_consumer,format=json,table_name=sensor_data,fields=temp:float=temperature hum:int=humidity,tags=location sensor_id' \
  kafka_sensors

Secure Kafka connection with SASL_SSL

influxdb3 create trigger \
  --database mydb \
  --plugin-filename gh:influxdata/kafka_subscriber/kafka_subscriber.py \
  --trigger-spec "every:10s" \
  --trigger-arguments 'bootstrap_servers=kafka1:9093 kafka2:9093,topics=secure.data,group_id=influxdb3_secure,format=json,table_name=secure_data,security_protocol=SASL_SSL,sasl_mechanism=SCRAM-SHA-512,sasl_username=myuser,sasl_password=mypass,ssl_ca_cert=certs/ca.crt,fields=value:float=value' \
  secure_kafka

Message Formats

JSON Format

The primary use case for structured data. Supports nested fields using JSONPath expressions.

TOML Configuration

[kafka]
bootstrap_servers = ["kafka:9092"]
topics = ["sensors.temperature"]
group_id = "influxdb3_json"
format = "json"

[mapping.json]
table_name = "sensor_data"
timestamp_field = "$.timestamp:ms"

[mapping.json.tags]
location = "$.location"
sensor_id = "$.sensor.id"

[mapping.json.fields]
temperature = ["$.temp", "float"]
humidity = ["$.humidity", "int"]
status = ["$.online", "bool"]

Example Message

{
  "timestamp": 1638360000000,
  "location": "warehouse_a",
  "sensor": {
    "id": "sensor_001"
  },
  "temp": 22.5,
  "humidity": 65,
  "online": true
}

Resulting Data

sensor_data,location=warehouse_a,sensor_id=sensor_001 temperature=22.5,humidity=65i,status=true 1638360000000000000

JSON Array Support

Process batch messages containing arrays of JSON objects:

[
  {"timestamp": 1638360000000, "sensor_id": "001", "temperature": 22.5},
  {"timestamp": 1638360001000, "sensor_id": "002", "temperature": 23.1},
  {"timestamp": 1638360002000, "sensor_id": "003", "temperature": 21.8}
]

Array processing behavior:

• Each array element is processed independently as a separate data point
• If one element fails to parse, the others continue processing (partial success)
• Parse errors for individual elements are logged as warnings and the element is skipped (not written to kafka_exceptions)
• Statistics count 1 Kafka message = 1 unit (regardless of array size)

Line Protocol Format

For messages already in InfluxDB line protocol format, use passthrough mode. No mapping configuration needed.

TOML Configuration

[kafka]
bootstrap_servers = ["kafka:9092"]
topics = ["influxdb.metrics"]
group_id = "influxdb3_lineprotocol"
format = "lineprotocol"

Example Message

sensor_data,location=warehouse_a,sensor_id=001 temperature=22.5,humidity=65i 1638360000000000000

Text Format

Parse plain text messages using regular expressions:

TOML Configuration

[kafka]
bootstrap_servers = ["kafka:9092"]
topics = ["legacy.logs"]
group_id = "influxdb3_text"
format = "text"

[mapping.text]
table_name = "sensor_logs"
timestamp_field = "ts:(\\d+):ms"

[mapping.text.tags]
location = "location=([^,\\s]+)"

[mapping.text.fields]
temperature = ["temp:([\\d.]+)", "float"]
humidity = ["hum:(\\d+)", "int"]

Example Message

location=warehouse_a,temp:22.5,hum:65,ts:1638360000000

Avro Format (Schema Registry)

Decode Avro messages in the Confluent wire format. The schema is resolved from the Schema Registry by the id embedded in each message; mapping uses JSONPath on the decoded record.

TOML Configuration

[kafka]
bootstrap_servers = ["kafka:9092"]
topics = ["sensors.avro"]
group_id = "influxdb3_avro"
format = "avro"

[kafka.schema_registry]
schema_registry_url = "http://schema-registry:8081"
# schema_registry_username = "sr_api_key"     # optional basic auth
# schema_registry_password = "sr_api_secret"

[mapping.avro]
table_name = "sensor_data"
timestamp_field = "$.timestamp:ms"

[mapping.avro.tags]
device_id = "$.device_id"

[mapping.avro.fields]
temperature = ["$.temperature", "float"]
humidity = ["$.humidity", "float"]

Avro Format (local .avsc schema, schemaless)

Decode raw schemaless Avro (no Schema Registry, no wire header) with a local .avsc schema. Set avro_schema_file to switch format = "avro" to this mode.

[kafka]
bootstrap_servers = ["kafka:9092"]
topics = ["sensors.avro"]
group_id = "influxdb3_avro_file"
format = "avro"

[kafka.avro]
avro_schema_file = "schemas/sensor.avsc"

[mapping.avro]
table_name = "sensor_data"
timestamp_field = "$.timestamp:ms"

[mapping.avro.tags]
device_id = "$.device_id"

[mapping.avro.fields]
temperature = ["$.temperature", "float"]
humidity = ["$.humidity", "float"]

JSON Schema Format (Schema Registry)

Decode JSON-Schema messages in the Confluent wire format. Configuration mirrors the Avro example, using format = "jsonschema" and [mapping.jsonschema].

[kafka]
bootstrap_servers = ["kafka:9092"]
topics = ["events.jsonschema"]
group_id = "influxdb3_jsonschema"
format = "jsonschema"

[kafka.schema_registry]
schema_registry_url = "http://schema-registry:8081"

[mapping.jsonschema]
table_name = "events"
timestamp_field = "$.ts:ms"

[mapping.jsonschema.fields]
value = ["$.value", "float"]

Protobuf Format (local .proto schema)

Decode Protobuf messages in the Confluent wire format using a local .proto file. The Schema Registry is not contacted; the message type is selected by the wire message index, or set protobuf_message_type to choose it explicitly.

[kafka]
bootstrap_servers = ["kafka:9092"]
topics = ["sensors.protobuf"]
group_id = "influxdb3_protobuf"
format = "protobuf"

[kafka.protobuf]
protobuf_schema_file = "schemas/sensor.proto"
# protobuf_include_dir = "schemas"               # optional, for .proto imports
# protobuf_message_type = "sensors.SensorReading" # optional, defaults to wire message index

[mapping.protobuf]
table_name = "sensor_data"
timestamp_field = "$.timestamp:ms"

[mapping.protobuf.tags]
device_id = "$.device_id"

[mapping.protobuf.fields]
temperature = ["$.temperature", "float"]
humidity = ["$.humidity", "float"]

Protobuf Format (Schema Registry)

Omit protobuf_schema_file and set schema_registry_url to fetch and compile the .proto (and any referenced schemas) by the wire-frame schema id. The [kafka.protobuf] section can be omitted entirely.

[kafka]
bootstrap_servers = ["kafka:9092"]
topics = ["sensors.protobuf"]
group_id = "influxdb3_protobuf_sr"
format = "protobuf"

[kafka.schema_registry]
schema_registry_url = "http://schema-registry:8081"

[mapping.protobuf]
table_name = "sensor_data"
timestamp_field = "$.timestamp:ms"

[mapping.protobuf.tags]
device_id = "$.device_id"

[mapping.protobuf.fields]
temperature = ["$.temperature", "float"]
humidity = ["$.humidity", "float"]

Statistics and Monitoring

The plugin tracks comprehensive statistics and writes them to the kafka_stats table on every plugin invocation.

kafka_stats Table

Querying Statistics

# Get latest statistics
influxdb3 query --database mydb \
  "SELECT * FROM kafka_stats ORDER BY time DESC LIMIT 10"

# Success rate by topic
influxdb3 query --database mydb \
  "SELECT topic, partition, success_rate, messages_processed, messages_failed
   FROM kafka_stats
   WHERE time > now() - INTERVAL '1 hour'
   ORDER BY time DESC"

# Check consumer lag indicators
influxdb3 query --database mydb \
  "SELECT topic, partition, last_offset
   FROM kafka_stats
   WHERE consumer_group = 'influxdb3_consumer'
   ORDER BY time DESC LIMIT 10"

Error Handling

Parse errors and message processing failures are logged to the kafka_exceptions table:

kafka_exceptions Table

Checking for Errors

influxdb3 query --database mydb \
  "SELECT * FROM kafka_exceptions ORDER BY time DESC LIMIT 10"

Dead-letter queue

When dlq_topic is configured, messages that fail to parse are additionally republished to that Kafka topic with the original payload and key, so they can be inspected or replayed without re-reading the source topic. Error context is attached as Kafka headers:

Transient InfluxDB write failures are not sent to the DLQ — they are retried on the next cycle (with offset_commit_policy=on_success).

Troubleshooting

Check Plugin Logs

influxdb3 query --database _internal \
  "SELECT * FROM system.processing_engine_logs
   WHERE trigger_name = 'kafka_ingestion'
   ORDER BY time DESC LIMIT 20"

Common Issues

“confluent-kafka library not installed” or “No module named ‘confluent_kafka’”

influxdb3 install package confluent-kafka

If you encounter librdkafka related errors:

• Ubuntu/Debian: sudo apt-get install librdkafka-dev
• macOS: brew install librdkafka

“Configuration file not found”

• For relative paths, ensure PLUGIN_DIR environment variable is set
• For absolute paths, verify the file exists at the specified location

# For relative paths
export PLUGIN_DIR=~/.plugins
ls $PLUGIN_DIR/my_kafka_config.toml

# Or use absolute path
ls /etc/kafka/my_kafka_config.toml

“Failed to connect to Kafka cluster”

• Verify bootstrap_servers addresses and ports
• Check network connectivity to Kafka brokers
• For SSL connections, verify certificate paths
• For SASL authentication, verify credentials

“SASL mechanism required when security_protocol includes SASL”

When using SASL_PLAINTEXT or SASL_SSL, you must provide:

• sasl_mechanism
• sasl_username
• sasl_password

“No fields were mapped from JSON data”

• Verify JSONPath expressions in field mappings (use $. prefix)
• Check that JSON structure matches your paths
• Review kafka_exceptions table for detailed errors

Messages not being processed

• Check trigger status: influxdb3 show summary --database mydb
• Verify Kafka connection in plugin logs
• Check consumer group lag using Kafka tools
• Increase trigger frequency (for example, from every:10s to every:5s)
• If many messages are queued, increase max_poll_records or set to 0 (unlimited)

Offset commit issues with on_success policy

If using offset_commit_policy=on_success and seeing repeated message processing:

• Check kafka_exceptions table for processing errors
• Fix the root cause of failures
• Consider using offset_commit_policy=always if some data loss is acceptable

Architecture

How It Works

1. Scheduled Trigger: Plugin runs on schedule (for example, every:10s)
2. Configuration Caching: Plugin configuration is parsed once and cached between executions
3. Persistent Consumer: The Kafka consumer is created once and reused across invocations (cached), so it stays in its consumer group and no rebalance happens every cycle
4. Consumer Poll: Consumer polls for available messages (drains all available messages)
5. Offset Commit: Based on offset_commit_policy:
   • always: Commits immediately after poll
   • on_success: Commits only after successful processing of all messages
6. Deduplication (optional): When dedup_id_field is set and no message timestamp is configured, duplicate ids within dedup_window are skipped
7. Parse & Write: Messages parsed according to format and written to InfluxDB
8. Error Tracking: Parse errors logged to kafka_exceptions table (and dlq_topic if configured)
9. Statistics: Written to kafka_stats table on every plugin invocation

Performance Optimization

The plugin includes several optimizations for high-throughput scenarios:

• Persistent Consumer: The consumer is reused across trigger executions instead of reconnecting each cycle, eliminating the consumer-group rebalance (and its latency/broker overhead) that a connect/disconnect-per-cycle would cause. It stays in the group while the schedule interval stays below max_poll_interval.
• Configuration Caching: Plugin configuration is parsed once and reused across all trigger executions
• Pre-compiled Patterns: JSONPath expressions and regex patterns are compiled once during parser initialization
• Batch Polling: Consumer drains available messages in a single trigger execution (limited by max_poll_records, default 500)
• Manual Offset Commit: Provides control over exactly-once vs at-least-once semantics

Consumer Group Behavior

• A single consumer connection is created and reused across trigger executions; it is only rebuilt after a fatal connection/poll error or a configuration change
• Set max_poll_interval above your trigger interval so the broker does not evict the reused consumer between polls (this would force a rebalance)
• Consumer group ID ensures partitions are assigned consistently
• Offsets are committed to Kafka for durability
• Multiple plugin instances with the same group_id will share partitions (load balancing)

Report an issue

For plugin issues, see the Plugins repository issues page.

Find support for InfluxDB 3 Core

The InfluxDB Discord server is the best place to find support for InfluxDB 3 Core and InfluxDB 3 Enterprise. For other InfluxDB versions, see the Support and feedback options.