private void registerNotificationListener ( String type , BiConsumer < ObjectName , Tags > perObject ) { NotificationListener notificationListener = ( notification , handback ) -> { MBeanServerNotification mbs = ( MBeanServerNotification ) notification ; ObjectName o = mbs . getMBeanName ( ) ; perObject . accept ( o , Tags . concat ( tags , nameTag ( o ) ) ) ; } ; NotificationFilter filter = ( NotificationFilter ) notification -> { if ( ! MBeanServerNotification . REGISTRATION_NOTIFICATION . equals ( notification . getType ( ) ) ) return false ; ObjectName obj = ( ( MBeanServerNotification ) notification ) . getMBeanName ( ) ; return obj . getDomain ( ) . equals ( JMX_DOMAIN ) && obj . getKeyProperty ( "type" ) . equals ( type ) ; } ; try { mBeanServer . addNotificationListener ( MBeanServerDelegate . DELEGATE_NAME , notificationListener , filter , null ) ; notificationListenerCleanUpRunnables . add ( ( ) -> { try { mBeanServer . removeNotificationListener ( MBeanServerDelegate . DELEGATE_NAME , notificationListener ) ; } catch ( InstanceNotFoundException | ListenerNotFoundException ignored ) { } } ) ; } catch ( InstanceNotFoundException e ) { throw new RuntimeException ( "Error registering Kafka MBean listener" , e ) ; } }
This notification listener should remain indefinitely since new Kafka consumers can be added at any time .