StreamsInstanceTracker.java 文件源码

private synchronized void addJobToMap(BigInteger jobid) {
      InstanceMXBean instance = null;
      LOGGER.debug("AddJobToMap({})...", jobid);
      StopWatch sw = new StopWatch();
      sw.start();

      try {
          instance = this.jmxContext
                  .getBeanSourceProvider()
                  .getBeanSource()
                  .getInstanceBean(domainName,
                          this.instanceInfo.getInstanceName());

          ObjectName tJobNameObj = instance.registerJob(jobid);

          JobMXBean jobBean = JMX.newMXBeanProxy(jmxContext
                  .getBeanSourceProvider().getBeanSource()
                  .getMBeanServerConnection(), tJobNameObj, JobMXBean.class,
                  true);
          jobMap.addJobToMap(jobid, new JobDetails(this, jobid, jobBean));

      } catch (IOException e) {
          LOGGER.warn("New Job Initialization received IO Exception from JMX Connection Pool.  Resetting monitor.  Exception Message: "
                  + e.getLocalizedMessage());
          resetTracker();
      }

      sw.stop();
      LOGGER.debug("** addJobToMap (jobid: " + jobid + ") time: "
              + sw.getTime());

metricsExporter.getStreamsMetric("jobCount", StreamsObjectType.INSTANCE, this.domainName, this.instanceInfo.getInstanceName()).set(jobMap.size());

  }