Configure the Kafka Producer

The Kafka Producer exposes the following options to help fine-tune its behavior.

Name Description Default

eventTopic

Name of the topic used for events. Set this to an empty string to disable forwarding events.

events

alarmTopic

Name of the topic used for alarms. Set this to an empty string to disable forwarding alarms.

alarms

alarmFeedbackTopic

Name of the topic used for alarm feedback. Set this to an empty string to disable forwarding alarm feedback.

alarmFeedback

nodeTopic

Name of the topic used for nodes. Set this to an empty string to disable forwarding nodes. Set this to an empty string to disable forwarding topologies.

nodes

topologyVertexTopic

Name of the topic used for topology vertices.

vertices

topologyEdgeTopic

Name of the topic used for topology edges.

edges

metricTopic

Name of the topic used for metrics.

metrics

eventFilter

A Spring SpEL expression (see below) used to filter events. Set this to an empty string to disable filtering, and forward all events.

none

alarmFilter

A Spring SpEL expression (see below) to filter alarms. Set this to an empty string to disable filtering, and forward all alarms.

none

metricFilter

A Spring SpEL expression (see below) to filter metrics. Set this to an empty string to disable filtering, and forward all metrics.

none

metricRouting.enabled

Set this to true to route metrics to different topics based on meta-data (see Route metrics to multiple topics).

false

metricRouting.contextKey

The meta-data key that names the destination topic, as context:key.

requisition:metricRouting.package

metricRouting.cacheTimeoutMs

Number of milliseconds a resolved topic is cached before the meta-data is read from the database again.

300000 (5 minutes)

forward.metrics

Set this value to true to enable forwarding of metrics.

false

metrics.useJson

Set this value to true to forward metrics encoded as JSON instead of protobuf (see below).

false

nodeRefreshTimeoutMs

Number of milliseconds to wait before looking up a node in the database again. Decrease this value to improve accuracy at the cost of additional database lookups.

300000 (5 minutes)

suppressIncrementalAlarms

Suppresses forwarding alarms that differ only by count or last event time. Set this to false to prevent suppressing these alarms.

true

kafkaSendQueueCapacity

The capacity for the queue of Kafka messages that is used when a Kafka message is pushed but Kafka is unavailable.

1000

startAlarmSyncWithCleanState

Set this to true to force the Kafka Streams client to start with a clean state on every boot.

false

alarmSync

Set this to false to disable synchronization of the alarms topics. This is automatically disabled when alarm forwarding is not enabled.

true

alarmSyncClear

Changes how alarm synchronization reconciles reduction keys that are on the topic but no longer in the database (see Alarm synchronization). When true, such keys are published as a record with severity CLEARED instead of a null tombstone. Only takes effect when alarmSync is also enabled.

false

alarmStream

Set this to true to make the alarms topic append-only: deletes are silent instead of emitting a null tombstone (see Alarm stream mode). Intended for consumers that treat the topic as an event stream rather than a compacted KTable.

false
Some of these configuration parameters require the bundle to be restarted for the changes to take effect.

Configure filtering

Use filtering to selectively forward events and/or alarms to the Kafka topics.

Filtering is performed using a Spring SpEL expression, which is evaluated against each object to determine if it should be forwarded. The expression must return a boolean value.

Enable event filtering

To enable event filtering, set the value of the eventFilter property to a valid SpEL expression.

$ ssh -p 8101 admin@localhost
...
admin@opennms()> config:edit org.opennms.features.kafka.producer
admin@opennms()> config:property-set eventFilter 'getUei().equals("uei.opennms.org/internal/discovery/newSuspect")'
admin@opennms()> config:update

In the example above, the filter is configured such that only events with the given UEI are forwarded. Consult the source code of the org.opennms.netmgt.xml.event.Event class in your distribution for a complete list of available properties.

Enable Alarm Filtering

To enable alarm filtering, set the value of the alarmFilter property to a valid SpEL expression.

$ ssh -p 8101 admin@localhost
...
admin@opennms()> config:edit org.opennms.features.kafka.producer
admin@opennms()> config:property-set alarmFilter 'getTTicketId() != null'
admin@opennms()> config:update

In the example above, the filter is configured such that only alarms that are associated with a ticket ID are forwarded. Consult the source code of the org.opennms.netmgt.model.OnmsAlarm class in your distribution for a complete list of available properties.

Enable Metric Filtering

To enable metric filtering, set the value of the metricFilter property to a valid SpEL expression.

$ ssh -p 8101 admin@localhost
...
admin@opennms()> config:edit org.opennms.features.kafka.producer
admin@opennms()> config:property-set metricFilter 'getLocation() == "Raleigh"'
admin@opennms()> config:update

In the example above, the filter is configured such that only metrics associated with the given location are forwarded. Available properties are nodeId, nodeLabel, foreignSource, foreignId, location, resourceTypeName, resourceName, resourceId, ifIndex and instance.

Enable metric forwarding

To enable metric forwarding, set the value of the forward.metrics property to true.

$ ssh -p 8101 admin@localhost
...
admin@opennms()> config:edit org.opennms.features.kafka.producer
admin@opennms()> config:property-set forward.metrics true
admin@opennms()> config:update

Kafka Producer sends a full CollectionSet as one message to Kafka. Large network equipment might have hundreds of thousands of metrics that could generate messages over the Kafka limit (1MB in size by default). If the message size is over 1MB , they will broken up into smaller messages that are less than 1MB. When sending one CollectionSet over multiple messages, their timestamps will remain same. === Disable metrics splitting into multiple kafka messages To disable metrics splitting into multiple kafka messages, set the value of disable.metrics.splitting property to true.

$ ssh -p 8101 admin@localhost
...
admin@opennms()> config:edit org.opennms.features.kafka.producer
admin@opennms()> config:property-set disable.metrics.splitting true
admin@opennms()> config:update

Or add following line at $OPENNMS_HOME/etc/org.opennms.features.kafka.producer.cfg

disable.metrics.splitting = true

Forward metrics as JSON

By default, metrics are forwarded as protobuf messages (see collectionset.proto in your distribution for the model definitions). To forward metrics encoded as JSON instead, set the value of the metrics.useJson property to true.

$ ssh -p 8101 admin@localhost
...
admin@opennms()> config:edit org.opennms.features.kafka.producer
admin@opennms()> config:property-set metrics.useJson true
admin@opennms()> config:update

Or add following line at $OPENNMS_HOME/etc/org.opennms.features.kafka.producer.cfg

metrics.useJson = true
The metric forwarder reads metrics.useJson (like forward.metrics and disable.metrics.splitting) only when the opennms-kafka-producer feature starts. After changing it, restart Meridian (or restart the feature) for the change to take effect.

The JSON representation follows the canonical protobuf JSON mapping of the same CollectionSet model. Note that JSON messages are larger than their protobuf equivalents; the message splitting described above accounts for the encoded size, so large collection sets may split into more messages when JSON is enabled. This setting only affects metrics; events, alarms, nodes, and topology messages are always forwarded as protobuf.

JSON has no representation for non-finite numbers, so the protobuf JSON mapping encodes them as the strings "NaN", "Infinity", and "-Infinity". A numeric attribute with a missing sample (for example, an uninitialized counter) is therefore forwarded with a string where consumers otherwise see a JSON number. Strongly typed consumers must either tolerate this mixed type or parse the payload with a protobuf JSON parser, which accepts these strings for numeric fields.

Route metrics to multiple topics

By default, every collected metric goes to metricTopic. Metric routing sends metrics to a topic named by meta-data on the node, interface, or service they were collected from, so that (for example) each tenant’s metrics can be separated at the broker instead of at the consumer.

To enable it, set metricRouting.enabled to true (metric forwarding must be enabled as well):

$ ssh -p 8101 admin@localhost
...
admin@opennms()> config:edit org.opennms.features.kafka.producer
admin@opennms()> config:property-set forward.metrics true
admin@opennms()> config:property-set metricRouting.enabled true
admin@opennms()> config:update

Then define the meta-data named by metricRouting.contextKey, by default requisition:metricRouting.package. Its value is the topic name. For example, to send everything collected on one interface to the topic foobar:

<node foreign-id="node1" node-label="node1">
  <interface ip-addr="10.0.0.1" status="1" snmp-primary="P">
    <meta-data context="requisition" key="metricRouting.package" value="foobar"/>
    <monitored-service service-name="SNMP"/>
  </interface>
</node>

Meta-data can be set on nodes, interfaces, and services, in the requisition, through the web UI, or through the REST API. A metric whose node, interface, and service all lack the key goes to metricTopic, so enabling routing never drops metrics.

Data collection is always triggered by a monitored service on an interface of a node, and routing follows that collection rather than the individual metric. The topic is determined by the first of the following that carries the key:

  1. the meta-data of the collecting service

  2. the meta-data of the interface the collecting service is assigned to

  3. the meta-data of the node the collecting service belongs to

  4. otherwise, the default metricTopic

For SNMP collected through 10.0.0.1 on a node:

Meta-data set on What is routed

the SNMP service on 10.0.0.1

everything that SNMP collection produces, overriding the interface and the node

interface 10.0.0.1

the same, unless the SNMP service itself is tagged

the node

every metric collected from the node, unless a collecting service or its interface is tagged

Meta-data on an interface only affects routing when a monitored service is assigned to that interface, because it is the collecting service’s interface that is consulted. Meta-data on the other interfaces of a node — including the SNMP interfaces the metrics actually describe — is not consulted. So you cannot move a single interface’s traffic metrics into their own topic by tagging that interface; tag the service, its interface, or the node to move a whole collection. You can use different collection packages bound to different service names to split interface metrics.

Response time metrics carry their own address and service, so they are routed by the polled service, then its interface, then the node.

Streaming telemetry, flows, and collections submitted through the integration API are not triggered by a monitored service, so only the exporter’s interface and the node apply to them.

Use opennms:kafka-metric-routing-test to check what a given node, interface, or service resolves to before relying on it — see Shell Commands.

Topic names

The meta-data value is used verbatim as the topic name. It is not interpolated, so a value such as ${requisition:tenant} is taken literally and rejected.

A value is only used if it is a legal Kafka topic name: at most 249 characters, made up of a-z, A-Z, 0-9, ., , and -, and not . or ... Values that fail are logged and the metrics go to metricTopic instead; they are never renamed automatically, because a rewritten name could silently collide with a different topic (Kafka treats . and as equivalent in its own metric names).

Routed topics must already exist, or the broker must be configured to create topics automatically. Sending to an unknown topic blocks the collection thread for up to max.block.ms (60 seconds by default) before failing, which delays collection. If you cannot pre-create topics, consider lowering max.block.ms in org.opennms.features.kafka.producer.client.metrics.

What consumers need to know

Because routing is decided per collection, a collection is still sent as a single message, keyed as it always was. Different services of the same node can nevertheless resolve to different topics, and Kafka provides no ordering guarantee across topics. A consumer that reconstructs per-node state therefore has to merge the topics it reads.

Monitoring

Metric routing exposes counters over JMX, and through the shell:

$ ssh -p 8101 admin@localhost
...
admin@opennms()> opennms:metrics-display -m org.opennms.features.kafka.producer
Counter Description

metricRouting.routed.topic

Resources published to topic.

metricRouting.unmapped

Resources whose meta-data key was not set, or was blank.

metricRouting.sanitizeRejected

Resources whose meta-data value was not a legal topic name.

metricRouting.unresolvedNode

Resources for which no node could be determined, so no meta-data could be read.

cache.kafkaMetricRouting.hitRate

Hit rate of the resolved-topic cache.

A resolved topic is cached for metricRouting.cacheTimeoutMs, so a meta-data change made through a requisition import or the REST API takes up to that long to take effect. Run opennms:kafka-metric-routing-test --clear-cache to apply it immediately.

Enable exclusive metric forwarding

Once metric forwarding is enabled, you can use this as the exclusive persistence strategy by setting the following system property:

echo 'org.opennms.timeseries.strategy=osgi' > "$OPENNMS_HOME/etc/opennms.properties.d/kafka-for-metrics.properties"

Configure topic names

By default five topics are created i.e., events, alarms, nodes,vertices, and edges . To change these, you can use:

$ ssh -p 8101 admin@localhost
...
admin@opennms()> config:edit org.opennms.features.kafka.producer
admin@opennms()> config:property-set eventTopic ""
admin@opennms()> config:property-set nodeTopic "opennms-nodes"
admin@opennms()> config:update

In the example above, we disable event forwarding by setting an empty topic name and change the node topic name to opennms-nodes.

$ ssh -p 8101 admin@localhost
...
admin@opennms()> config:edit org.opennms.features.kafka.producer
admin@opennms()> config:property-set topologyVertexTopic "opennms-bridge-vertex"
admin@opennms()> config:property-set topologyEdgeTopic "opennms-edge-vertex"
admin@opennms()> config:update

In the example above, we set the vertex and edge topics to be different by default.

Alarm synchronization

When alarm forwarding is enabled, the producer runs a periodic reconciliation loop (alarmSync, on by default). It compares a materialized view of the alarm topic against a database snapshot and emits corrections so the topic converges on the database over time. This is eventually consistent: corrections are applied on the next snapshot cycle, not the moment a discrepancy appears. It mainly recovers from cases the live path can miss, such as a dropped send or an alarm removed without a lifecycle callback (for example, a node-deletion cascade).

For a reduction key that is on the topic but no longer in the database, there are two strategies:

  • Tombstone (default): the key is removed with a null-value record. This is standard KTable semantics.

  • Clear (alarmSyncClear = true): the key’s last-known value is republished with severity CLEARED, so no null records are written. A note is appended to the alarm’s log message recording that it was auto-cleared by synchronization because the source alarm no longer exists in the database. This can help consumers that do not model the topic as a compacted KTable and mishandle null values, such as some Kafka Connect sinks. The clear is idempotent, so an already-cleared key is not re-emitted.

Because the clear strategy never tombstones, keys are not dropped from the materialized store, so it can grow larger over a long-running session.

Alarm stream mode

By default the alarms topic behaves like a changelog: each reduction key holds the alarm’s current state, and a delete is a null-value tombstone. This suits consumers that read the topic as a compacted KTable, but the tombstones confuse consumers that just want a record for every alarm change.

Set alarmStream to true to treat the topic as an append-only event stream instead. Updates are still keyed by reduction key, but deletes are silent: no tombstone is emitted. Consumers see the rising edge (an alarm is raised or escalates) and the falling edge (the alarm’s CLEARED update) as ordinary records, and simply stop seeing a key once it is resolved.

Silent deletes rely on an alarm being cleared before it is removed, which is the normal lifecycle. An alarm that leaves the database without being cleared first (for example, a node-deletion cascade) would otherwise stay active downstream. Reconciliation (alarmSync) covers this case, but only if it clears orphaned alarms rather than tombstoning them, so stream mode cannot be used with the default tombstone reconciliation.

To enable stream mode, set alarmStream = true, then use one of these two combinations:

  • alarmStream = true, alarmSync = true, alarmSyncClear = true (recommended): reconciliation runs and clears orphaned alarms instead of tombstoning them.

  • alarmStream = true, alarmSync = false: reconciliation is turned off entirely. Uncleared cascade deletes are not reconciled, so they stay active downstream.

Setting alarmStream = true and alarmSync = true while leaving alarmSyncClear = false is not permitted, because reconciliation would write tombstones onto the stream. In that case Meridian logs a warning and disables synchronization at startup.

The recommended configuration sets both alarmStream and alarmSyncClear to true (alarmSync is already true by default):

$ ssh -p 8101 admin@localhost
...
admin@opennms()> config:edit org.opennms.features.kafka.producer
admin@opennms()> config:property-set alarmStream true
admin@opennms()> config:property-set alarmSyncClear true
admin@opennms()> config:update

Configure the alarm sync streams client

When producing alarms to a topic, we also automatically enable a Kafka Stream client to help synchronize the contents of the topic and ensure eventual consistency with the database.

The streams client takes different properties than the producer and requires a separate configuration map.

We automatically pull known properties from the producer configuration, allowing it to work without further configuration in most cases.

If your producer takes a special configuration directive, or you would like to tune the behavior of the stream client, you can set properties in org.opennms.features.kafka.producer.streams:

$ ssh -p 8101 admin@localhost
...
admin@opennms()> config:edit org.opennms.features.kafka.producer.streams
admin@opennms()> config:property-set default.dsl.store rocksDB
admin@opennms()> config:update

Any property set in org.opennms.features.kafka.producer.streams will override those inherited from org.opennms.features.kafka.producer.client.