org.opennms.features.kafka.producer

Class OpennmsKafkaProducer

java.lang.Object

org.opennms.features.kafka.producer.OpennmsKafkaProducer

All Implemented Interfaces:

AlarmLifecycleListener, EventListener

public class OpennmsKafkaProducer
extends Object
implements AlarmLifecycleListener, EventListener

Field Summary

Fields

Modifier and Type	Field and Description
static String	KAFKA_CLIENT_PID

Constructor Summary

Constructors

Constructor and Description
OpennmsKafkaProducer(ProtobufMapper protobufMapper, NodeCache nodeCache, org.osgi.service.cm.ConfigurationAdmin configAdmin, EventSubscriptionService eventSubscriptionService)

Method Summary

Modifier and Type	Method and Description
void	destroy()
AlarmCallbackStateTracker	getAlarmCallbackStateTracker()
CountDownLatch	getAlarmForwardedLatch()
CountDownLatch	getEventForwardedLatch()
String	getName() Return the id of the listener
CountDownLatch	getNodeForwardedLatch()
void	handleAlarmSnapshot(List<OnmsAlarm> alarms) Called periodically with a complete set of alarms as present in the database at the given timestamp.
void	handleDeletedAlarm(int alarmId, String reductionKey) Called when an alarm has been deleted.
void	handleNewOrUpdatedAlarm(OnmsAlarm alarm) Called when an alarm has been created or updated.
void	init()
boolean	isForwardingAlarms()
void	onEvent(Event event) Process a sent event.
void	postHandleAlarmSnapshot() Called after AlarmLifecycleListener.handleAlarmSnapshot(java.util.List<org.opennms.netmgt.model.OnmsAlarm>) has been called on all the listeners, and after the session & transaction used to perform the snapshot has been closed.
void	preHandleAlarmSnapshot() Called before the transaction is opened and the alarms are read for subsequent calls to AlarmLifecycleListener.handleAlarmSnapshot(java.util.List<org.opennms.netmgt.model.OnmsAlarm>).
void	setAlarmFilter(String alarmFilter)
void	setAlarmTopic(String alarmTopic)
OpennmsKafkaProducer	setDataSync(KafkaAlarmDataSync dataSync)
void	setEventFilter(String eventFilter)
void	setEventTopic(String eventTopic)
void	setKafkaSendQueueCapacity(int kafkaSendQueueCapacity)
void	setNodeTopic(String nodeTopic)
void	setSuppressIncrementalAlarms(boolean suppressIncrementalAlarms)
boolean	shouldForwardAlarm(OnmsAlarm alarm)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

KAFKA_CLIENT_PID

public static final String KAFKA_CLIENT_PID

See Also:

Constant Field Values

Constructor Detail

OpennmsKafkaProducer

public OpennmsKafkaProducer(ProtobufMapper protobufMapper,
                            NodeCache nodeCache,
                            org.osgi.service.cm.ConfigurationAdmin configAdmin,
                            EventSubscriptionService eventSubscriptionService)

Method Detail

init

public void init()
          throws IOException

Throws:

IOException

destroy

public void destroy()

shouldForwardAlarm

public boolean shouldForwardAlarm(OnmsAlarm alarm)

handleAlarmSnapshot

public void handleAlarmSnapshot(List<OnmsAlarm> alarms)

Description copied from interface: AlarmLifecycleListener

Called periodically with a complete set of alarms as present in the database at the given timestamp. This should be used to synchronize any state to ensure it matches what is currently in the database. Note that it is possible that the *current* state of alarms is different from the state at the time at which the snapshot was taken. Implementations should take this in consideration when performing any state synchronization. This method will be called while the related session & transaction that created the alarm are still open. All of the listeners are invoked serially, so the implementors should avoid blocking when possible.

Specified by:

handleAlarmSnapshot in interface AlarmLifecycleListener

Parameters:

alarms - canonical set of alarms in the database

preHandleAlarmSnapshot

public void preHandleAlarmSnapshot()

Description copied from interface: AlarmLifecycleListener

Called before the transaction is opened and the alarms are read for subsequent calls to AlarmLifecycleListener.handleAlarmSnapshot(java.util.List<org.opennms.netmgt.model.OnmsAlarm>). This can be used to trigger any necessary state tracking to accurately handle the snapshot results.

Specified by:

preHandleAlarmSnapshot in interface AlarmLifecycleListener

postHandleAlarmSnapshot

public void postHandleAlarmSnapshot()

Description copied from interface: AlarmLifecycleListener

Called after AlarmLifecycleListener.handleAlarmSnapshot(java.util.List<org.opennms.netmgt.model.OnmsAlarm>) has been called on all the listeners, and after the session & transaction used to perform the snapshot has been closed. This can be used to trigger any necessary post-processing of the results once the related session has been closed. This function may be called immediately after a call to AlarmLifecycleListener.preHandleAlarmSnapshot() if an error occurred while preparing the snapshot i.e. when opening the transaction.

Specified by:

postHandleAlarmSnapshot in interface AlarmLifecycleListener

handleNewOrUpdatedAlarm

public void handleNewOrUpdatedAlarm(OnmsAlarm alarm)

Description copied from interface: AlarmLifecycleListener

Called when an alarm has been created or updated. This method will be called while the related session & transaction that created the alarm are still open. All of the listeners are invoked serially, so the implementors should avoid blocking when possible.

Specified by:

handleNewOrUpdatedAlarm in interface AlarmLifecycleListener

Parameters:

alarm - a newly created or updated alarm

handleDeletedAlarm

public void handleDeletedAlarm(int alarmId,
                               String reductionKey)

Description copied from interface: AlarmLifecycleListener

Called when an alarm has been deleted. This method will be called while the related session & transaction that created the alarm are still open. All of the listeners are invoked serially, so the implementors should avoid blocking when possible.

Specified by:

handleDeletedAlarm in interface AlarmLifecycleListener

Parameters:

alarmId - id of the alarm that was deleted

reductionKey - reduction key of the alarm that was deleted

getName

public String getName()

Description copied from interface: EventListener

Return the id of the listener

Specified by:

getName in interface EventListener

Returns:

a String object.

onEvent

public void onEvent(Event event)

Description copied from interface: EventListener

Process a sent event.

Specified by:

onEvent in interface EventListener

Parameters:

event - a Event object.

setEventTopic

public void setEventTopic(String eventTopic)

setAlarmTopic

public void setAlarmTopic(String alarmTopic)

setNodeTopic

public void setNodeTopic(String nodeTopic)

setEventFilter

public void setEventFilter(String eventFilter)

setAlarmFilter

public void setAlarmFilter(String alarmFilter)

setDataSync

public OpennmsKafkaProducer setDataSync(KafkaAlarmDataSync dataSync)

isForwardingAlarms

public boolean isForwardingAlarms()

getEventForwardedLatch

public CountDownLatch getEventForwardedLatch()

getAlarmForwardedLatch

public CountDownLatch getAlarmForwardedLatch()

getNodeForwardedLatch

public CountDownLatch getNodeForwardedLatch()

setSuppressIncrementalAlarms

public void setSuppressIncrementalAlarms(boolean suppressIncrementalAlarms)

getAlarmCallbackStateTracker

public AlarmCallbackStateTracker getAlarmCallbackStateTracker()

setKafkaSendQueueCapacity

public void setKafkaSendQueueCapacity(int kafkaSendQueueCapacity)