Kapacitor .id() field limitation in alert node partitioning issues with Kafka

InfluxDB 1 Kapacitor
, , , ,
1

:pushpin: Environment
Kapacitor Version: 1.7.7

  • Deployment: Docker container
  • Output Handler: Kafka (.topic() used to send alerts)
    Alert Streams: Multiple alert branches (flat_threshold, analytic, etc.)
    Kafka Setup: 4 partitions, expecting even distribution

:clipboard: Problem Statement
We are writing alerts to a Kafka topic using Kapacitor’s .topic() handler. For partitioning to work correctly, Kafka uses the message key, which Kapacitor derives from the .id() field of the alert.

However:
1.) Setting .id('') causes errors like:
err=“key required”

2.) Setting .id("static_value") sends all messages to one Kafka partition only, causing load imbalance

We want to use a dynamic .id()
Attempts like:
1.) .id(‘{{ .Tags.metric_id_string | default “fallback-{{ .Time.UnixNano }}” }}’)
Fail with the below error
template: id:1:86: executing “id” at <.Time.UnixNano>: can’t evaluate field Time in type kapacitor.idInfo

Tried multiple things to fix this but did’nt worked.

Even when I enable logging to see the available tags they are present but when we try to access it in alert id it provides key error

Point Log:
ts=2025-07-06T15:40:01.895Z lvl=info msg="batch point" service=kapacitor task_master=main task=filter_ai_metric node=log15 prefix="ALERT FINAL UUID" name= group= tag_metric_type=health tag_level5=m2a tag_metadata="{\"ai_params\":null,\"alert_description\":\"Dummy metric\",\"analytics\":\"flat_threshold\",\"attributes\":{\"tags\":{\"Applications\":[\"atomiq\"]}},\"collection_id\":null,\"collector_template\":null,\"db\":\"sitescope\",\"insights\":null,\"last_seen\":\"2025-07-01 13:11:01.967429\",\"last_update\":\"07-06-2025, 14:51:06.522731\",\"measurement\":\"SIS\",\"metric_id_string\":\"ATOMIQ-MON-m2a_latency-dummy_metric-atomiq_operability-Telegraf-e2e\",\"rules\":{\"flat_threshold\":{\"additional_params\":{\"outlier_threshold\":null,\"reset_threshold\":null,\"time_window\":null,\"tw_timestamp\":null},\"critical\":{\"numeric\":[{\"condition\":\"Gt\",\"threshold\":[5]}],\"text\":{\"thresholds\":[{}]}}}},\"selfhealing\":false,\"severity\":{\"alert_severity\":6,\"previous_severity\":6,\"severity\":6,\"severity_count\":2},\"update_by\":{\"insight\":\"BP\",\"tags\":\"custom\",\"threshold\":\"BP\"}}" tag_level7=_ tag_service=monitoring tag_level3=atomiq tag_source=Telegraf tag_metric=e2e tag_level10=_ tag_level4=monitoring tag_monitor=m2a_latency tag_level6=dummy tag_level9=_ tag_uuid=7cdb8a61-7203-49a1-930f-a03d39a8de0b tag_level1=atomiq_operability tag_metric_id_string=ATOMIQ-MON-m2a_latency-dummy_metric-atomiq_operability-Telegraf-e2e tag_host=49b046b37108 tag_level2=Amdocs tag_level8=_ field_VALUE=-1751816400464002000 time=2025-07-06T15:40:01Z

What We’re Trying to Achieve
1.) Use a dynamic .id() in Kapacitor that:

  • Leverages a tag like metric_id_string if available
  • Falls back to another tag or runtime-unique value (e.g., timestamp or UUID)
  • Ensure alerts are evenly partitioned in Kafka
  • Avoid "key required" log spam
  • Avoid .id("static"), which locks all Kafka messages to a single partition

### :folded_hands: Help Needed
Any suggestions on:

  • Safe .id() generation that allows for unique keys per message
  • Making .Time available to .id() template
  • Alternative design pattern to support even Kafka partitioning

Tick script:

dbrp "TOKEN_INFLUXDB_LT_DB_SITESCOPE"."autogen"

var data = stream
	|from()
		.measurement('SIS')
    @filter_metrics()
		.redis_host('TOKEN_REDIS_HOST')
		.redis_port(TOKEN_REDIS_PORT)
		.redis_db(0)
		.worker_size(2000)
		.tls_enabled(TOKEN_REDIS_TLS_MODE)

var md_init = data
	|where(lambda: isPresent("IsRapidMetadataPoint"))
	|log()
		.prefix('MD INIT processing')
	@window()
		.every(30s)
		.everyCount(500)
		.createAlignmentPoint(TRUE)
    |alert()
		.details(''' TEST ''')
		.info(lambda: TRUE)
		.topic('metadata-init')

var m2a_processing = data
	|where(lambda: !isPresent("IsRapidMetadataPoint") AND isPresent("analytics"))

m2a_processing
        |where(lambda: strContains("analytics", 'flat_threshold'))
        |delete()
			.tag('analytics')
        @window()
			.everyCount(500)
			.every(500ms)
			.createAlignmentPoint(TRUE)
        |alert()
			.details(''' TEST ''')
			.info(lambda:TRUE)
			.id('')
			.topic('m2a_kafka')

m2a_processing
        |where(lambda: strContains("analytics", 'analytic'))
        |delete()
			.tag('analytics')
        @window()
			.everyCount(500)
			.every(500ms)
			.createAlignmentPoint(TRUE)
        |alert()
			.details(''' TEST ''')
			.info(lambda:TRUE )
			.id('')
			.topic('m2a_ai_kafka')

var final = data
final
    |where(lambda: !isPresent("IsRapidMetadataPoint") AND !isPresent("analytics"))
    @window()
        .every(30s)
        .createAlignmentPoint(TRUE)
    |alert()
                .details(''' TEST ''')
                .info(lambda: TRUE)
                .topic('influx_data_point')

I am totally new to kapacitor and blocked with this