org.opennms.features.kafka.producer

Class OpennmsKafkaProducer

java.lang.Object

org.opennms.features.kafka.producer.OpennmsKafkaProducer

All Implemented Interfaces:

AlarmFeedbackListener, AlarmLifecycleListener, EventListener, ThreadAwareEventListener, OnmsTopologyConsumer

public class OpennmsKafkaProducer
extends Object
implements AlarmLifecycleListener, EventListener, AlarmFeedbackListener, OnmsTopologyConsumer, ThreadAwareEventListener

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, OnmsTopologyDao topologyDao)

Method Summary

Modifier and Type	Method and Description
void	consume(OnmsTopologyMessage message)
void	destroy()
AlarmCallbackStateTracker	getAlarmCallbackStateTracker()
CountDownLatch	getAlarmFeedbackForwardedLatch()
CountDownLatch	getAlarmForwardedLatch()
String	getEncoding()
CountDownLatch	getEventForwardedLatch()
CountDownLatch	getForwardedTopologyEdgeMessage()
CountDownLatch	getForwardedTopologyVertexMessage()
String	getName() Return the id of the listener
CountDownLatch	getNodeForwardedLatch()
int	getNumEventListenerThreads()
int	getNumThreads()
Set<OnmsTopologyProtocol>	getProtocols()
void	handleAlarmFeedback(List<AlarmFeedback> alarmFeedback) Handle the newly generated collection of alarm feedback.
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(IEvent 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	setAlarmFeedbackTopic(String alarmFeedbackTopic)
void	setAlarmFilter(String alarmFilter)
void	setAlarmTopic(String alarmTopic)
OpennmsKafkaProducer	setDataSync(KafkaAlarmDataSync dataSync)
void	setEncoding(String encoding)
void	setEventFilter(String eventFilter)
void	setEventTopic(String eventTopic)
void	setKafkaSendQueueCapacity(int kafkaSendQueueCapacity)
void	setNodeTopic(String nodeTopic)
void	setNumEventListenerThreads(int numEventListenerThreads)
void	setSuppressIncrementalAlarms(boolean suppressIncrementalAlarms)
void	setTopologyEdgeTopic(String topologyEdgeTopic)
void	setTopologyVertexTopic(String topologyVertexTopic)
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,
                            OnmsTopologyDao topologyDao)

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

Specified by:

getName in interface OnmsTopologyConsumer

Returns:

a String object.

onEvent

public void onEvent(IEvent event)

Description copied from interface: EventListener

Process a sent event.

Specified by:

onEvent in interface EventListener

Parameters:

event - a IEvent object.

getProtocols

public Set<OnmsTopologyProtocol> getProtocols()

Specified by:

getProtocols in interface OnmsTopologyConsumer

consume

public void consume(OnmsTopologyMessage message)

Specified by:

consume in interface OnmsTopologyConsumer

setTopologyVertexTopic

public void setTopologyVertexTopic(String topologyVertexTopic)

setTopologyEdgeTopic

public void setTopologyEdgeTopic(String topologyEdgeTopic)

setEventTopic

public void setEventTopic(String eventTopic)

setAlarmTopic

public void setAlarmTopic(String alarmTopic)

setNodeTopic

public void setNodeTopic(String nodeTopic)

setAlarmFeedbackTopic

public void setAlarmFeedbackTopic(String alarmFeedbackTopic)

setEventFilter

public void setEventFilter(String eventFilter)

setAlarmFilter

public void setAlarmFilter(String alarmFilter)

setDataSync

public OpennmsKafkaProducer setDataSync(KafkaAlarmDataSync dataSync)

handleAlarmFeedback

public void handleAlarmFeedback(List<AlarmFeedback> alarmFeedback)

Description copied from interface: AlarmFeedbackListener

Handle the newly generated collection of alarm feedback.

Specified by:

handleAlarmFeedback in interface AlarmFeedbackListener

Parameters:

alarmFeedback - the collection of alarm feedback

isForwardingAlarms

public boolean isForwardingAlarms()

getEventForwardedLatch

public CountDownLatch getEventForwardedLatch()

getAlarmForwardedLatch

public CountDownLatch getAlarmForwardedLatch()

getNodeForwardedLatch

public CountDownLatch getNodeForwardedLatch()

getAlarmFeedbackForwardedLatch

public CountDownLatch getAlarmFeedbackForwardedLatch()

setSuppressIncrementalAlarms

public void setSuppressIncrementalAlarms(boolean suppressIncrementalAlarms)

getAlarmCallbackStateTracker

public AlarmCallbackStateTracker getAlarmCallbackStateTracker()

setKafkaSendQueueCapacity

public void setKafkaSendQueueCapacity(int kafkaSendQueueCapacity)

getNumThreads

public int getNumThreads()

Specified by:

getNumThreads in interface ThreadAwareEventListener

getForwardedTopologyVertexMessage

public CountDownLatch getForwardedTopologyVertexMessage()

getForwardedTopologyEdgeMessage

public CountDownLatch getForwardedTopologyEdgeMessage()

getEncoding

public String getEncoding()

setEncoding

public void setEncoding(String encoding)

getNumEventListenerThreads

public int getNumEventListenerThreads()

setNumEventListenerThreads

public void setNumEventListenerThreads(int numEventListenerThreads)