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

forward.metrics

Set this value to true to enable forwarding of metrics.

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

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

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 Horizon 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.