docs: 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);
          }
      }