/**
 * Obtain the default JBoss JCA WorkManager through a JMX lookup
 * for the JBossWorkManagerMBean.
 * @param mbeanName the JMX object name to use
 * @see org.jboss.resource.work.JBossWorkManagerMBean
 */
public static WorkManager getWorkManager(String mbeanName) {
	Assert.hasLength(mbeanName, "JBossWorkManagerMBean name must not be empty");
	try {
		Class<?> mbeanClass = JBossWorkManagerUtils.class.getClassLoader().loadClass(JBOSS_WORK_MANAGER_MBEAN_CLASS_NAME);
		InitialContext jndiContext = new InitialContext();
		MBeanServerConnection mconn = (MBeanServerConnection) jndiContext.lookup(MBEAN_SERVER_CONNECTION_JNDI_NAME);
		ObjectName objectName = ObjectName.getInstance(mbeanName);
		Object workManagerMBean = MBeanServerInvocationHandler.newProxyInstance(mconn, objectName, mbeanClass, false);
		Method getInstanceMethod = workManagerMBean.getClass().getMethod("getInstance");
		return (WorkManager) getInstanceMethod.invoke(workManagerMBean);
	}
	catch (Exception ex) {
		throw new IllegalStateException(
				"Could not initialize JBossWorkManagerTaskExecutor because JBoss API is not available", ex);
	}
}
 

public static void main(String... args) throws Exception {
    MBeanServerConnection msc = ManagementFactory.getPlatformMBeanServer();
    HotSpotDiagnosticMXBean mxbean =
        ManagementFactory.getPlatformMXBean(msc, HotSpotDiagnosticMXBean.class);


    String[] signatures = new String[] {
        String.class.getName()
    };
    Object obj = msc.invoke(mxbean.getObjectName(), "getVMOption",
        new String[] { "PrintVMOptions"}, signatures);

    CompositeData data = (CompositeData)obj;
    validateType(data);

    VMOption option = mxbean.getVMOption("PrintVMOptions");
    VMOption o = VMOption.from(data);
    assertEquals(option, o);
}
 

private void checkMemberView(boolean expectedBeans,
    MBeanServerConnection mbeanServer, String memberId, String serverGroup)
    throws MalformedObjectNameException, IOException, TestException {
  Log.getLogWriter().info("Checking GFXDMemberMBean for server group " + serverGroup + " and memberId : "  + memberId);
  String memberONstr = memberONTemplate.replace("{0}", serverGroup);
  memberONstr = memberONstr.replace("{1}", memberId);
  ObjectName memberON = new ObjectName(memberONstr);

  // check memberMBean
  Log.getLogWriter().info("Querying for member instance " + memberON);

  try {
    ObjectInstance instance = mbeanServer.getObjectInstance(memberON);
    Log.getLogWriter().info("Found instance " + memberON + " Expected " + expectedBeans);
    if (!expectedBeans && instance != null)
      throw new TestException("Found member instance " + memberON + " when not expected ");
  } catch (InstanceNotFoundException e) {
    if (expectedBeans) {
      logMBeans(MessageFormat.format(ManagementConstants.OBJECTNAME__GFXDMEMBER_MXBEAN, new Object[] {"*", "*"}), mbeanServer);
      throw new TestException("checkMemberViewFromManager\n" + TestHelper.getStackTrace(e));
    }
  }
}
 

private Number getAggregatedValuesForCounter(String counter, MBeanServerConnection mbeanServer) {
  String[] members = null;
  try {
    members = getGfxdClusterMembers(mbeanServer);
  } catch (Exception e) {
  }
  Number aggregatedValue = 0;
  for (String member : members) {
    String memberName = member.substring(member.indexOf('(') + 1, member.indexOf(":"));
    @SuppressWarnings("unchecked")
    String clientName = ((Map<String, String>)SQLBB.getBB().getSharedMap().get(MEMBERS_MAPPING)).get(memberName);
    String counterName = counter + clientName;
    double doubleValue = getCounterNumber(counterName).doubleValue();
    aggregatedValue  = aggregatedValue.doubleValue() + doubleValue;
    Log.getLogWriter().info(counterName + " value for member : " + member + " is " + doubleValue);
  }
  return aggregatedValue;
}
 

private MBeanInfo testMBeanForDatasource() throws Exception {
    Map<String, String[]> env = new HashMap<>();
    String[] credentials = { "admin", "admin" };
    env.put(JMXConnector.CREDENTIALS, credentials);
    try {
        String url = "service:jmx:rmi://localhost:12311/jndi/rmi://localhost:11199/jmxrmi";
        JMXServiceURL jmxUrl = new JMXServiceURL(url);
        JMXConnector jmxConnector = JMXConnectorFactory.connect(jmxUrl, env);
        MBeanServerConnection mBeanServer = jmxConnector.getMBeanServerConnection();
        ObjectName mbeanObject = new ObjectName(dataSourceName + ",-1234:type=DataSource");
        MBeanInfo mBeanInfo = mBeanServer.getMBeanInfo(mbeanObject);
        return mBeanInfo;
    } catch (MalformedURLException | MalformedObjectNameException | IntrospectionException | ReflectionException e) {
        throw new AxisFault("Error while connecting to MBean Server " + e.getMessage(), e);
    }
}
 

@Override
public void setLoggersLevels(JmxInstance instance, final Map<String, String> updates) throws LogControlException {
    withConnection(instance, new JmxAction<Void>() {
        @Override
        public Void perform(JmxInstance jmx, MBeanServerConnection connection) throws Exception {
            JmxLogControlMBean logControlMBean = getRemoteLogControl(connection);
            logControlMBean.setLoggersLevels(updates);
            return null;
        }
    });

    for (Map.Entry<String, String> logger : updates.entrySet()) {
        log.info(String.format("Level for logger '%s' set to '%s' on '%s'",
                logger.getKey(), logger.getValue(), instance.getNodeName()));
    }
}
 

private static void connect(String pid, String address) throws Exception {
    if (address == null) {
        throw new RuntimeException("Local connector address for " +
                                   pid + " is null");
    }

    System.out.println("Connect to process " + pid + " via: " + address);

    JMXServiceURL url = new JMXServiceURL(address);
    JMXConnector c = JMXConnectorFactory.connect(url);
    MBeanServerConnection server = c.getMBeanServerConnection();

    System.out.println("Connected.");

    RuntimeMXBean rt = newPlatformMXBeanProxy(server,
        RUNTIME_MXBEAN_NAME, RuntimeMXBean.class);
    System.out.println(rt.getName());

    // close the connection
    c.close();
}
 

public JmxMemoryPoolManager(MBeanServerConnection connection) throws IOException {
	List<MemoryPoolMXBean> memoryPoolMXBeans = ManagementFactory.getPlatformMXBeans(connection,
			MemoryPoolMXBean.class);
	for (MemoryPoolMXBean memoryPool : memoryPoolMXBeans) {
		String name = memoryPool.getName().trim();
		String lowerCaseName = name.toLowerCase();
		if (lowerCaseName.contains(SURVIVOR)) {
			survivorMemoryPool = memoryPool;
		} else if (lowerCaseName.contains(EDEN)) {
			edenMemoryPool = memoryPool;
		} else if (lowerCaseName.contains(OLD) || lowerCaseName.contains(TENURED)) {
			oldMemoryPool = memoryPool;
		} else if (lowerCaseName.contains(PERM) || lowerCaseName.contains(METASPACE)) {
			permMemoryPool = memoryPool;
		} else if (lowerCaseName.contains(CODE_CACHE)) {
			codeCacheMemoryPool = memoryPool;
		} else if (lowerCaseName.contains(COMPRESSED_CLASS_SPACE)) {
			compressedClassSpaceMemoryPool = memoryPool;
		}
	}
}
 

public static List<String> getBeanProxiesForBlackboardMap(MBeanServerConnection server, Map<String,String> map) throws IOException{
  String clientName = RemoteTestModule.getMyClientName();
  String array[] = clientName.split("_");
  logFine("clientName split Array DSName = " + HydraUtil.ObjectToString(array));
  logFine("clientName split Array length = " + array.length);
  String dsName= null;
  if(array.length>=4){
    //wan test assume last part if DS Name
    dsName = array[array.length-1];
    logInfo("Considering names of clients with DSName = " + dsName);
  }
  //int count = map.size();
  int count = 0;
  List<String> listOFNS = new ArrayList<String>();
  for(String s: map.values()){
    if(dsName==null || (dsName!=null&& s.contains(dsName))){
      listOFNS.add(s);
      count++;
    }
  }    
  logInfo("getBeanProxiesForBlackboardMap :  " + listOFNS);
  return listOFNS;
}
 

public synchronized MBeanServerConnection
        getMBeanServerConnection(Subject delegationSubject)
        throws IOException {

    if (terminated) {
        if (logger.traceOn())
            logger.trace("getMBeanServerConnection","[" + this.toString() +
                    "] already closed.");
        throw new IOException("Connection closed");
    } else if (!connected) {
        if (logger.traceOn())
            logger.trace("getMBeanServerConnection","[" + this.toString() +
                    "] is not connected.");
        throw new IOException("Not connected");
    }

    return getConnectionWithSubject(delegationSubject);
}
 

public static List<String> checkForMBeanProxies(MBeanServerConnection server,List<String> listOFNS, Map<String,String> map) throws IOException{
  logInfo("Checking " + listOFNS + " names on server "+ server);
  try {      
    Set<ObjectName> objectNames = server.queryNames(null, null);
    logFine("Mbean Set registered at platform mbean server after federation : " + objectNames);      
    for(ObjectName n : objectNames){
      for(String s : map.values())
        if(n.toString().contains(s)){
          listOFNS.remove(s);
        }
    }
    logInfo("Final count of remaining mbeans not present in Manager view : " + listOFNS.size() + " List : " + listOFNS);
    return listOFNS;
  } finally{      
  }
}
 

/**
 * Execute the specified command. This logic only performs the common
 * attribute validation required by all subclasses; it does not perform any
 * functional logic directly.
 * 
 * @exception BuildException
 *                if a validation error occurs
 */
@Override
public void execute() throws BuildException {
    if (testIfCondition() && testUnlessCondition()) {
        try {
            String error = null;

            MBeanServerConnection jmxServerConnection = getJMXConnection();
            error = jmxExecute(jmxServerConnection);
            if (error != null && isFailOnError()) {
                // exception should be thrown only if failOnError == true
                // or error line will be logged twice
                throw new BuildException(error);
            }
        } catch (Exception e) {
            if (isFailOnError()) {
                throw new BuildException(e);
            } else {
                handleErrorOutput(e.getMessage());
            }
        } finally {
            closeRedirector();
        }
    }
}
 

/**
 * Transfroms a proxy implementing T in a proxy implementing T plus
 * NotificationEmitter
 *
 **/
public static <T> T makeNotificationEmitter(T proxy,
                    Class<T> mbeanInterface) {
    if (proxy instanceof NotificationEmitter)
        return proxy;
    if (proxy == null) return null;
    if (!(proxy instanceof Proxy))
        throw new IllegalArgumentException("not a "+Proxy.class.getName());
    final Proxy p = (Proxy) proxy;
    final InvocationHandler handler =
            Proxy.getInvocationHandler(proxy);
    if (!(handler instanceof MBeanServerInvocationHandler))
        throw new IllegalArgumentException("not a JMX Proxy");
    final MBeanServerInvocationHandler h =
            (MBeanServerInvocationHandler)handler;
    final ObjectName name = h.getObjectName();
    final MBeanServerConnection mbs = h.getMBeanServerConnection();
    final boolean isMXBean = h.isMXBean();
    final T newProxy;
    if (isMXBean)
        newProxy = JMX.newMXBeanProxy(mbs,name,mbeanInterface,true);
    else
        newProxy = JMX.newMBeanProxy(mbs,name,mbeanInterface,true);
    return newProxy;
}
 

public MBeanServerConnection call() {
    MBeanServer mbs = MBeanServerFactory.newMBeanServer();
    try {
        JMXServiceURL url = new JMXServiceURL("service:jmx:rmi://");
        JMXConnectorServer cs =
            JMXConnectorServerFactory.newJMXConnectorServer(
                url, null, mbs);
        cs.start();
        JMXServiceURL addr = cs.getAddress();
        connector = JMXConnectorFactory.connect(addr);
        return connector.getMBeanServerConnection();
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}
 

@Override
public String jmxExecute(MBeanServerConnection jmxServerConnection)
    throws Exception {

    if (getName() == null) {
        throw new BuildException("Must specify a 'name'");
    }
    if ((attribute == null)) {
        throw new BuildException(
                "Must specify a 'attribute' for get");
    }
    return  jmxGet(jmxServerConnection, getName());
 }
 

@Test
public void testInvokeOperationStandalone() throws Exception {
    MBeanServerConnection connection = setupAndGetConnection(new MBeanInfoAdditionalInitialization(ProcessType.STANDALONE_SERVER, new TestExtension()));

    ObjectName name = createObjectName(LEGACY_DOMAIN + ":subsystem=test");

    ModelControllerResourceDefinition.VoidOperationNoParams.INSTANCE.invoked = false;
    Assert.assertNull(connection.invoke(name, ModelControllerResourceDefinition.VoidOperationNoParams.OPERATION_JMX_NAME, null, null));
    Assert.assertTrue(ModelControllerResourceDefinition.VoidOperationNoParams.INSTANCE.invoked);

    String result = assertCast(String.class, connection.invoke(
            name,
            ModelControllerResourceDefinition.IntOperationWithParams.OPERATION_JMX_NAME,
            new Object[]{100L, new String[]{"A"}, Collections.singletonMap("test", 3), 5, 5},
            new String[]{Long.class.getName(), String[].class.getName(), Map.class.getName(), Integer.class.getName(), Integer.class.getName()}));
    Assert.assertEquals("A105", result);
    Assert.assertTrue(ModelControllerResourceDefinition.IntOperationWithParams.INSTANCE_NO_EXPRESSIONS.invoked);

    MBeanInfo info = connection.getMBeanInfo(name);
    CompositeType complexType = assertCast(CompositeType.class, findAttribute(info.getAttributes(), "complex").getOpenType());
    CompositeData complexData = createComplexData(connection, complexType, 5, BigDecimal.valueOf(10.3d));
    Assert.assertEquals(complexData, assertCast(CompositeData.class, connection.invoke(
            name,
            ModelControllerResourceDefinition.ComplexOperation.OPERATION_NAME,
            new Object[]{complexData},
            new String[]{CompositeData.class.getName()})));
}
 

private Supplier<MBeanServerConnection> getJmxServerConnection(JmxInfinispanCacheStatistics annotation) throws MalformedURLException {
    final String host;
    final int port;

    if (annotation.dc() != DC.UNDEFINED && annotation.dcNodeIndex() != -1) {
        ContainerInfo node = suiteContext.get().getAuthServerBackendsInfo(annotation.dc().getDcIndex()).get(annotation.dcNodeIndex());
        Container container = node.getArquillianContainer();
        if (container.getDeployableContainer() instanceof KeycloakOnUndertow) {
            return () -> ManagementFactory.getPlatformMBeanServer();
        }
        host = "localhost";
        port = container.getContainerConfiguration().getContainerProperties().containsKey("managementPort")
          ? Integer.valueOf(container.getContainerConfiguration().getContainerProperties().get("managementPort"))
          : 9990;
    } else {
        host = annotation.host().isEmpty()
          ? System.getProperty((annotation.hostProperty().isEmpty()
            ? "keycloak.connectionsInfinispan.remoteStoreServer"
            : annotation.hostProperty()))
          : annotation.host();

        port = annotation.managementPort() == -1
          ? Integer.valueOf(System.getProperty((annotation.managementPortProperty().isEmpty()
            ? "cache.server.management.port"
            : annotation.managementPortProperty())))
          : annotation.managementPort();
    }


    JMXServiceURL url = new JMXServiceURL("service:jmx:remote+http://" + host + ":" + port);
    return () -> {
        try {
            return jmxConnectorRegistry.get().getConnection(url).getMBeanServerConnection();
        } catch (IOException ex) {
            throw new RuntimeException(ex);
        }
    };
}
 

private final int doMemoryMXBeanTest(MBeanServerConnection mbsc) {
    int errorCount = 0 ;
    System.out.println("---- MemoryMXBean") ;

    try {
        ObjectName memoryName =
                new ObjectName(ManagementFactory.MEMORY_MXBEAN_NAME) ;
        MBeanInfo mbInfo = mbsc.getMBeanInfo(memoryName);
        errorCount += checkNonEmpty(mbInfo);
        System.out.println("getMBeanInfo\t\t"
                + mbInfo);
        MemoryMXBean memory = null ;

        memory =
                JMX.newMXBeanProxy(mbsc,
                memoryName,
                MemoryMXBean.class,
                true) ;
        System.out.println("getMemoryHeapUsage\t\t"
                + memory.getHeapMemoryUsage());
        System.out.println("getNonHeapMemoryHeapUsage\t\t"
                + memory.getNonHeapMemoryUsage());
        System.out.println("getObjectPendingFinalizationCount\t\t"
                + memory.getObjectPendingFinalizationCount());
        System.out.println("isVerbose\t\t"
                + memory.isVerbose());

        System.out.println("---- OK\n") ;

    } catch (Exception e) {
        Utils.printThrowable(e, true) ;
        errorCount++ ;
        System.out.println("---- ERROR\n") ;
    }

    return errorCount ;
}
 

private static void mbeanOp(MBeanServerConnection mserver, ObjectName name, boolean adding)
        throws Exception {
    if (adding) {
        mserver.createMBean("javax.management.timer.Timer", name);
    } else {
        mserver.unregisterMBean(name);
    }
}
 

private final int doCompilationMXBeanTest(MBeanServerConnection mbsc) {
    int errorCount = 0 ;
    System.out.println("---- CompilationMXBean") ;

    try {
        ObjectName compilationName =
                new ObjectName(ManagementFactory.COMPILATION_MXBEAN_NAME);

        if ( mbsc.isRegistered(compilationName) ) {
            MBeanInfo mbInfo = mbsc.getMBeanInfo(compilationName);
            errorCount += checkNonEmpty(mbInfo);
            System.out.println("getMBeanInfo\t\t" + mbInfo);
            CompilationMXBean compilation = null ;

            compilation =
                    JMX.newMXBeanProxy(mbsc,
                    compilationName,
                    CompilationMXBean.class) ;
            System.out.println("getName\t\t"
                    + compilation.getName());
            boolean supported =
                    compilation.isCompilationTimeMonitoringSupported() ;
            System.out.println("isCompilationTimeMonitoringSupported\t\t"
                    + supported);

            if ( supported ) {
                System.out.println("getTotalCompilationTime\t\t"
                        + compilation.getTotalCompilationTime());
            }
        }

        System.out.println("---- OK\n") ;
    } catch (Exception e) {
        Utils.printThrowable(e, true) ;
        errorCount++ ;
        System.out.println("---- ERROR\n") ;
    }

    return errorCount ;
}
 

static MXBeanLookup lookupFor(MBeanServerConnection mbsc) {
    synchronized (mbscToLookup) {
        WeakReference<MXBeanLookup> weakLookup = mbscToLookup.get(mbsc);
        MXBeanLookup lookup = (weakLookup == null) ? null : weakLookup.get();
        if (lookup == null) {
            lookup = new MXBeanLookup(mbsc);
            mbscToLookup.put(mbsc, new WeakReference<MXBeanLookup>(lookup));
        }
        return lookup;
    }
}
 

/**
 * Prints all MBeans whatever the domain is.
 */
private static void printMBeans(MBeanServerConnection mbsc) throws Exception {
    Set<ObjectName> set = mbsc.queryNames(null, null);
    System.out.println("---- MBeans found :");

    for (Iterator<ObjectName> iter = set.iterator(); iter.hasNext(); ) {
        System.out.println(iter.next().toString());
    }

    System.out.println("\n") ;
}
 

private final int doMemoryMXBeanTest(MBeanServerConnection mbsc) {
    int errorCount = 0 ;
    System.out.println("---- MemoryMXBean") ;

    try {
        ObjectName memoryName =
                new ObjectName(ManagementFactory.MEMORY_MXBEAN_NAME) ;
        MBeanInfo mbInfo = mbsc.getMBeanInfo(memoryName);
        errorCount += checkNonEmpty(mbInfo);
        System.out.println("getMBeanInfo\t\t"
                + mbInfo);
        MemoryMXBean memory = null ;

        memory =
                JMX.newMXBeanProxy(mbsc,
                memoryName,
                MemoryMXBean.class,
                true) ;
        System.out.println("getMemoryHeapUsage\t\t"
                + memory.getHeapMemoryUsage());
        System.out.println("getNonHeapMemoryHeapUsage\t\t"
                + memory.getNonHeapMemoryUsage());
        System.out.println("getObjectPendingFinalizationCount\t\t"
                + memory.getObjectPendingFinalizationCount());
        System.out.println("isVerbose\t\t"
                + memory.isVerbose());

        System.out.println("---- OK\n") ;

    } catch (Exception e) {
        Utils.printThrowable(e, true) ;
        errorCount++ ;
        System.out.println("---- ERROR\n") ;
    }

    return errorCount ;
}
 

@Override
public LoggingHostInfo getHostInfo(JmxInstance instance) {
    return withConnection(instance, new JmxAction<LoggingHostInfo>() {
        @Override
        public LoggingHostInfo perform(JmxInstance jmx, MBeanServerConnection connection) throws Exception {
            JmxLogControlMBean logControlMBean = getRemoteLogControl(connection);
            return new LoggingHostInfo(
                    logControlMBean.getLoggerNames(),
                    logControlMBean.getAppenders(),
                    logControlMBean.getLogFileNames()
            );
        }
    });
}
 

/**
 * Listener for a specific MBean.
 * Caller must call {@link #start} to start listening.
 * @param server
 * @param objectName
 * @param callback
 */
public MBeanListener(
        final MBeanServerConnection server,
        final ObjectName objectName,
        final T callback)
{
    mMBeanServer = server;
    mObjectName = objectName;
    mJMXDomain = null;
    mType = null;
    mName = null;
    mCallback = callback;
}
 

private JvmReadingInstance getDaemonThreadsCount(
                                                  MBeanServerConnection connection,
                                                  ReadingBean reading ) {

    String jvmPort = reading.getParameter("JMX_PORT");
    final MBeanWrapper mbeanWrapper = mbeanWrappers.get(jvmPort);

    return new JvmReadingInstance(connection,
                                  String.valueOf(reading.getDbId()),
                                  reading.getMonitorName(),
                                  getName(reading, jvmPort),
                                  reading.getUnit(),
                                  1) {
        private static final long serialVersionUID = 1L;

        @Override
        public void init() {

            mBeanName = mbeanWrapper.getObjectName("java.lang:type=Threading");
        }

        @Override
        public float poll() {

            return fixLongValue(Long.valueOf(mbeanWrapper.getMBeanAttribute(mBeanName,
                                                                            "DaemonThreadCount")
                                                         .toString()))
                   * normalizationFactor;
        }
    };
}
 

/**
 * Dumps a local or remote MBeanServer's entire object tree for support purposes. Nested arrays and CompositeData
 * objects in MBean attribute values are handled.
 * 
 * @param connection
 *            the server connection (or server itself)
 * @param out
 *            PrintWriter to write the output to
 * @throws IOException
 *             Signals that an I/O exception has occurred.
 */
public static void dumpConnection(MBeanServerConnection connection, PrintWriter out) throws IOException
{
    JmxDumpUtil.showStartBanner(out);
    
    // Get all the object names
    Set<ObjectName> objectNames = connection.queryNames(null, null);

    // Sort the names (don't assume ObjectName implements Comparable in JDK 1.5)
    Set<ObjectName> newObjectNames = new TreeSet<ObjectName>(new Comparator<ObjectName>()
    {
        public int compare(ObjectName o1, ObjectName o2)
        {
            return o1.toString().compareTo(o2.toString());
        }
    });
    newObjectNames.addAll(objectNames);
    objectNames = newObjectNames;

    
    // Dump each MBean
    for (ObjectName objectName : objectNames)
    {
        try
        {
            printMBeanInfo(connection, objectName, out);
        }
        catch (JMException e)
        {
            // Sometimes beans can disappear while we are examining them
        }
    }
}
 

/**
 * Get kafka version.
 */
public String getKafkaVersion(String host, int port, String ids, String clusterAlias) {
	JMXConnector connector = null;
	String version = "-";
	String JMX = "service:jmx:rmi:///jndi/rmi://%s/jmxrmi";
	try {
		JMXServiceURL jmxSeriverUrl = new JMXServiceURL(String.format(JMX, host + ":" + port));
		connector = JMXFactoryUtils.connectWithTimeout(jmxSeriverUrl, 30, TimeUnit.SECONDS);
		MBeanServerConnection mbeanConnection = connector.getMBeanServerConnection();
		if (CollectorType.KAFKA.equals(SystemConfigUtils.getProperty(clusterAlias + ".kafka.eagle.offset.storage"))) {
			version = mbeanConnection.getAttribute(new ObjectName(String.format(BrokerServer.BROKER_VERSION.getValue(), ids)), BrokerServer.BROKER_VERSION_VALUE.getValue()).toString();
		} else {
			version = mbeanConnection.getAttribute(new ObjectName(KafkaServer8.VERSION.getValue()), KafkaServer8.VALUE.getValue()).toString();
		}
	} catch (Exception ex) {
		LOG.error("Get kafka version from jmx has error, msg is " + ex.getMessage());
	} finally {
		if (connector != null) {
			try {
				connector.close();
			} catch (IOException e) {
				LOG.error("Close jmx connector has error, msg is " + e.getMessage());
			}
		}
	}
	return version;
}
 

public void doTestMBeanServerNotification_REGISTRATION_NOTIFICATION(MBeanServerConnection connection, boolean mustReceiveNotification) throws Exception {
    final ObjectName testObjectName = createObjectName(LEGACY_DOMAIN + ":subsystem=test");
    final ObjectName childObjectName = createObjectName(LEGACY_DOMAIN + ":subsystem=test,single=only");

    final CountDownLatch notificationEmitted = new CountDownLatch(1);
    final AtomicReference<Notification> notification = new AtomicReference<>();
    NotificationListener listener = new MbeanServerNotificationListener(notification, notificationEmitted, LEGACY_DOMAIN);
    NotificationFilterSupport filter = new NotificationFilterSupport();
    filter.enableType(MBeanServerNotification.REGISTRATION_NOTIFICATION);

    connection.addNotificationListener(MBeanServerDelegate.DELEGATE_NAME, listener, filter, null);

    // add a management resource
    connection.invoke(testObjectName, "addSingleOnly", new Object[]{123}, new String[]{String.class.getName()});

    if (mustReceiveNotification) {
        Assert.assertTrue("Did not receive expected notification", notificationEmitted.await(1, TimeUnit.SECONDS));
        Notification notif = notification.get();
        Assert.assertNotNull(notif);
        Assert.assertTrue(notif instanceof MBeanServerNotification);
        MBeanServerNotification mBeanServerNotification = (MBeanServerNotification) notif;
        Assert.assertEquals(MBeanServerNotification.REGISTRATION_NOTIFICATION, notif.getType());
        Assert.assertEquals(childObjectName, mBeanServerNotification.getMBeanName());
    } else {
        Assert.assertFalse("Did receive unexpected notification", notificationEmitted.await(500, TimeUnit.MILLISECONDS));
    }

    connection.removeNotificationListener(MBeanServerDelegate.DELEGATE_NAME, listener, filter, null);
}
 

public MBeanServerConnection call() {
    MBeanServer mbs = MBeanServerFactory.newMBeanServer();
    try {
        JMXServiceURL url = new JMXServiceURL("service:jmx:rmi://");
        JMXConnectorServer cs =
            JMXConnectorServerFactory.newJMXConnectorServer(
                url, null, mbs);
        cs.start();
        JMXServiceURL addr = cs.getAddress();
        connector = JMXConnectorFactory.connect(addr);
        return connector.getMBeanServerConnection();
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}